有没有一种方法可以让两个线程不从共享集合中获取相同的项？

Question

我有一个程序有3个线程。一个线程专用于向共享集合中添加项，其余两个线程用于从集合中移除项并对其进行处理。

我创建了一个用于添加的Runnable。看起来是这样的：

public class Adder implements Runnable {

    private static final Integer MAX_NO = 1000; // I wanted this to be Integer.MAX_VALUE
    private final List<Integer> shared;
    private final AtomicBoolean isAddingDone;

    public Adder(final List<Integer> shared, final AtomicBoolean isAddingDone) {
        this.shared= shared;
        this.isAddingDone = isAddingDone;
    }

    @Override
    public void run() {
        for (int i = 0; i < MAX_NO; i++) {
            shared.add(i);
        }
        isAddingDone.set(true);
    }

}

我创建了另一个Runnable，该Runnable可从共享集合中删除项。

public class Remover implements Runnable {

    private final int id;
    private boolean stopRequested = false;
    private final List<Integer> shared;
    private final AtomicBoolean isAddingDone;

    public Remover(final int id, final List<Integer> shared, final AtomicBoolean isAddingDone) {
        this.id = id;
        this.shared = shared;
        this.isAddingDone = isAddingDone;
    }

    public synchronized void requestStop() {
        this.stopRequested = true;
    }

    public synchronized boolean isStopRequested() {
        return this.stopRequested;
    }

    @Override
    public void run() {
        System.out.println("Starting Remover" + id + "...");
        while (!isStopRequested()) {
            if (shared.isEmpty() && isAddingDone.get()) {
                this.requestStop();
            } else {
                final Integer value = shared.get(0);
                System.out.println("Remover" + this.id + " retrieved " + value + ". Removing...");
                shared.remove(value);
                // processing of value here
            }
        }
    }

}

主要方法如下：

public static void main(String[] args) {
    final List<Integer> shared = new Vector<>();
    final AtomicBoolean isAddingDone = new AtomicBoolean(false);

    final Runnable tAdder = new Adder(shared, isAddingDone);
    final Runnable tRemover0 = new Remover(0, shared, isAddingDone);
    final Runnable tRemover1 = new Remover(1, shared, isAddingDone);

    Thread t1 = new Thread(tAdder, "Thread-Adder");
    t1.start();

    Thread r0 = new Thread(tRemover0, "Thread-Remover0");
    r0.start();

    Thread r1 = new Thread(tRemover1, "Thread-Remover1");
    r1.start();
}

我在运行程序时没有错误，但我注意到，在某一点上，两个线程都在检索同一个项。

Starting Remover0...
Starting Remover1...
Remover1 retrieved 0. Removing...
Remover0 retrieved 0. Removing...
Remover1 retrieved 1. Removing...
Remover0 retrieved 1. Removing...
Remover1 retrieved 2. Removing...
Remover1 retrieved 3. Removing...

有没有一种方法可以让两个移除器不处理同一件物品？我尝试使用迭代器，但在运行时，我得到了一个ConcurrentModificationException。

Answer 1

shared.isempty()、shared.get(0)和shared.remove(value)不会自动发生，因此不能保证列表在其间没有更改。

如果要在线程之间通信，请使用blockingqueue，并使用poll()从其中删除可用的项。这允许您将IsEmpty()、Get和Remove组合到单个原子操作中。

// From the while loop inside Remover:
            Integer value = shared.poll();
            if (value == null /* indicates the queue was empty when you looked. */
                    && isAddingDone.get()) {
                this.requestStop();
            } else {
                System.out.println("Remover" + this.id + " retrieved " + value + ". Removing...");
                // No need to remove, you already did.
                // processing of value here
            }