在N次执行后协调多个线程

Question

我有一个Java/Groovy多线程进程，我想在N次执行后“同步”： - 每次线程执行后共享计数器递减 - 目标是重置计数器一次它达到0，而没有其他线程正在访问它。

我试过ReadWriteReentrantLock但它看起来我有一个竞争条件的递减阶段。这里是我的测试代码

public static void main(String[] args) { 
    AtomicInteger counter = new AtomicInteger(Decrementer.MAX_SIZE) 
    ReadWriteLock lock = new ReentrantReadWriteLock() 

    for (int i = 1; i <= 10; i++) { 
     Decrementer d = new Decrementer(counter, lock) 
     new Thread(d).start() 
    } 
} 

public class Decrementer implements Runnable { 

    public final static int MAX_SIZE = 5 
    private ReadWriteLock lock 
    private AtomicInteger counter 

    public Decrementer(AtomicInteger counter, ReadWriteLock lock) { 
     this.counter = counter 
     this.lock = lock 
    } 

    public void run() { 
     while (true) { 
      try{ 
       lock.readLock().lock() 
       int current = this.counter.decrementAndGet() 
       System.out.println(Thread.currentThread().getName() + " at counter " + current) 
       Thread.sleep(762) 
      } finally { 
       lock.readLock().unlock() 
      } 

      try { 
       lock.writeLock().lock() 
       int current = this.counter.get() 
       if (current <= 0) { 
        this.counter.set(Decrementer.MAX_SIZE) 
        System.out.println(Thread.currentThread().getName() + " reset " + current + " to " + Decrementer.MAX_SIZE) 
        Thread.sleep(4217) 
       } 
      } finally { 
       lock.writeLock().unlock() 
      } 
     } 
    } 
} 

这给下面的怪异输出（带负计数器值），可能是由于的AtomicInteger价值的缺失“同步”的检查。

Thread-3 at counter 2 
Thread-2 at counter 4 
Thread-1 at counter 3 
Thread-4 at counter 1 
Thread-5 at counter 0 
Thread-6 at counter -1 
Thread-7 at counter -2 
Thread-8 at counter -3 
Thread-9 at counter -4 
Thread-10 at counter -5 
Thread-2 reset -5 to 5 
Thread-3 at counter 4 
Thread-4 at counter 2 
Thread-2 at counter 3 
Thread-1 at counter 1 
Thread-5 at counter -3 
Thread-10 at counter -4 
Thread-7 at counter -1 
Thread-6 at counter -2 
Thread-8 at counter 0 
Thread-9 at counter -5 
Thread-9 reset -5 to 5 

我也看到了CountDownLatch和CyclicBarrier类，但我的目标是不同步的所有线程但要保证计数器复位是原子和其他线程排除其他修改。

您是否看到我在代码中遗漏的任何明显的并发问题？

Answer 1

你的锁反之亦然。你正在读取锁定，然后进行“写入”。这应该工作：

public static void main(String[] args) { 
    AtomicInteger counter = new AtomicInteger(Decrementer.MAX_SIZE); 
    ReadWriteLock lock = new ReentrantReadWriteLock(); 

    for (int i = 1; i <= 10; i++) { 
     Decrementer d = new Decrementer(counter, lock); 
     new Thread(d).start(); 
    } 
} 

static public class Decrementer implements Runnable { 

    public final static int MAX_SIZE = 5; 
    private ReadWriteLock lock; 
    private AtomicInteger counter; 

    public Decrementer(AtomicInteger counter, ReadWriteLock lock) { 
     this.counter = counter; 
     this.lock = lock; 
    } 

    public void run() { 
     while (true) { 
      try { 
       lock.writeLock().lock(); 
       int current = this.counter.decrementAndGet(); 
       System.out.println(Thread.currentThread().getName() + " at counter " + current); 
       if (current <= 0) { 
        this.counter.set(Decrementer.MAX_SIZE); 
        System.out.println(Thread.currentThread().getName() + " reset " + current + " to " + Decrementer.MAX_SIZE); 
        Thread.sleep(4217); 
       } 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } finally { 
       lock.writeLock().unlock(); 
      } 
     } 
    } 
} 

Answer 2

基础上，写锁获取评论，在这里工作的解决方案：

public class Decrementer implements Runnable { 

public final static int MAX_SIZE = 20; 
private ReadWriteLock lock; 
private AtomicInteger counter; 

public Decrementer(AtomicInteger counter, ReadWriteLock lock) { 
    this.counter = counter; 
    this.lock = lock; 
} 

public void run() { 

    while (true) { 
     int current 
     try { 
      lock.writeLock().lock(); 
      if (this.counter.compareAndSet(0, Decrementer.MAX_SIZE)) { 
       current = 0; 
       flushWork(current); 
      } else { 
       current = counter.decrementAndGet(); 
       System.out.println(Thread.currentThread().getName() + " decrement " + current); 
      } 
     } finally { 
      lock.writeLock().unlock(); 
     } 

     try { 
      lock.readLock().lock(); 
      doWork(current); 
     } finally { 
      lock.readLock().unlock(); 
     } 
    } 
} 

private void flushWork(int current) { 
    System.out.println(Thread.currentThread().getName() + " reset " + current + " to " + Decrementer.MAX_SIZE); 
    System.out.println('------------------------------------------------------------------------------------'); 
    Thread.sleep(4217); 
} 

private void doWork(int current) { 
    System.out.println(Thread.currentThread().getName() + " at counter " + current); 
    Thread.sleep(62); 
} 
}