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