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