Java -多队列生产者消费者

Question

我有以下代码：

    while(!currentBoard.boardIsValid()){
        for (QueueLocation location : QueueLocation.values()){
            while(!inbox.isEmpty(location)){
                Cell c = inbox.dequeue(location);
                notifyNeighbours(c.x, c.y, c.getCurrentState(),previousBoard);
            }
        }
    }

我有一个有几个队列的使用者(他们所有的方法都是同步的)。每个生产者有一个队列。使用者遍历所有队列并检查它们是否有一个任务供他使用。如果他正在检查的队列中有一个任务，他就会消耗它。否则，他会去检查下一个队列，直到他完成对所有队列的迭代。

到目前为止，如果他遍历所有队列，并且所有队列都是空的，他就继续循环，而不是等待其中的一个包含什么(正如外部while所看到的)。

我如何让使用者等到其中一个队列中有什么东西时再等待呢？

我对下面的场景有一个问题:假设只有两个队列。消费者检查了第一个，它是空的。就在他检查第二个(也是空的)时，生产者在第一个队列中放了一些东西。对于消费者来说，队列都是空的，所以他应该等待(尽管其中一个不再是空的，他应该继续循环)。

编辑:最后一件事。这对我来说是个练习。我正在尝试自己实现同步。因此，如果任何java库都有实现此功能的解决方案，我对此并不感兴趣。我在试着理解我如何实现这一点。

Answer 1

@安倍关系很好。我会使用信号和等待-使用对象类内置，因为他们是最轻的重量。

Object sync = new Object();  // Can use an existing object if there's an appropriate one

// On submit to queue
synchronized ( sync ) {
    queue.add(...);  // Must be inside to avoid a race condition
    sync.notifyAll();
}

// On check for work in queue
synchronized ( sync ) {
    item = null;
    while ( item == null ) {
        // Need to check all of the queues - if there will be a large number, this will be slow,
        // and slow critical sections (synchronized blocks) are very bad for performance
        item = getNextQueueItem();
        if ( item == null ) {
            sync.wait();
        }
    }
}

注意，sync.wait释放同步锁，直到通知-并且成功调用等待方法需要同步锁(它提醒程序员，确实需要某种类型的关键部分才能可靠地工作)。

顺便说一句，如果可行的话，我会推荐一个专用于消费者(或一组消费者)的队列，而不是一个专用于生产者的队列。这将简化解决方案。

Answer 2

如果您想阻止跨多个队列，那么一个选项是使用java的 method。

因此，只要生产者有数据，它就应该调用signallAll。

Lock fileLock = new ReentrantLock();
Condition condition = fileLock.newCondition();
...
// producer has to signal
condition.signalAll();
...
// consumer has to await.
condition.await();

这样，只有当信号被提供时，使用者才会去检查队列。

Answer 3

我解决了类似于@Abe建议的类似情况，但我决定将Semaphore与AtomicBoolean结合起来，并将其命名为BinarySemaphore。它确实要求生产者进行修改，以便他们在有什么事情要做的时候发出信号。

下面是BinarySemaphore的代码和消费者工作循环应该是什么样子的一般概念：

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class MultipleProdOneConsumer {

BinarySemaphore workAvailable = new BinarySemaphore();

class Consumer {

    volatile boolean stop;

    void loop() {

        while (!stop) {
            doWork();
            if (!workAvailable.tryAcquire()) {
                // waiting for work
                try {
                    workAvailable.acquire();
                } catch (InterruptedException e) {
                    if (!stop) {
                        // log error
                    }
                }
            }
        }
    }

    void doWork() {}

    void stopWork() {
        stop = true;
        workAvailable.release();
    }
}

class Producer {

    /* Must be called after work is added to the queue/made available. */
    void signalSomethingToDo() {
        workAvailable.release();
    }
}

class BinarySemaphore {

    private final AtomicBoolean havePermit = new AtomicBoolean();
    private final Semaphore sync;

    public BinarySemaphore() {
        this(false);
    }

    public BinarySemaphore(boolean fair) {
        sync = new Semaphore(0, fair);
    }

    public boolean release() {

        boolean released = havePermit.compareAndSet(false, true);
        if (released) {
            sync.release();
        }
        return released;
    }

    public boolean tryAcquire() {

        boolean acquired = sync.tryAcquire();
        if (acquired) {
            havePermit.set(false);
        }
        return acquired;
    }

    public boolean tryAcquire(long timeout, TimeUnit tunit) throws InterruptedException {

        boolean acquired = sync.tryAcquire(timeout, tunit);
        if (acquired) {
            havePermit.set(false);
        }
        return acquired;
    }

    public void acquire() throws InterruptedException {

        sync.acquire();
        havePermit.set(false);
    }

    public void acquireUninterruptibly() {

        sync.acquireUninterruptibly();
        havePermit.set(false);
    }

}

}