在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); 
} 
}