Java - 多个队列生成者消费者

3 回答

• 0

  如果要阻塞多个队列，那么一个选项是使用java的Lock and Condition objects and then use the signal method .

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

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

  这种方式只有在提供信号时，消费者才会去检查队列 .

  回复于 2024-05-02T17:19:44+08:00
• 1

  @Abe很接近 . 我会使用信号并等待 - 使用Object类内置函数，因为它们是最轻的权重 .

  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 释放同步锁定，直到通知 - 并且需要锁定同步才能成功调用wait方法（这提醒程序员确实需要某种类型的关键部分才能使其可靠地工作） .

  顺便说一下，如果可行的话，我会推荐一个专门针对消费者（或消费者群体）的队列，而不是专用于 生产环境 者的队列 . 它将简化解决方案 .

  回复于 2024-05-02T17:19:44+08:00
• 0

  我按照@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);
      }
  
  }
  
  }
  

  回复于 2024-05-02T17:19:44+08:00