生产者-消费者线程间通信

Question

在线程间通信方面遇到问题，并通过到处使用“虚拟消息”来“解决”它。这是一个坏主意吗？有哪些可能的解决方案？

我遇到的示例问题。

主线程启动一个线程来处理和插入数据库中的记录。 主线程读取一个可能很大的文件并将一个记录（对象）一个接一个地放入阻塞队列中。处理线程从队列中读取并开始工作。

如何告诉“处理线程”停止？ 队列可以是空的，但工作没有完成，当处理线程完成工作并且不能中断它时，主线程现在也没有。

处理线程确实如此

while (queue.size() > 0 || !Thread.currentThread().isInterrupted()) {
    MyObject object= queue.poll(100, TimeUnit.MILLISECONDS);
    if (object != null) {
        String data = object.getData();
        if (data.equals("END")) {
            break;
        }
    // do work
    }
}
// clean-up
synchronized queue) {
    queue.notifyAll();
}
return;

和主线程

// ...start processing thread...
while(reader.hasNext(){
    // ...read whole file and put data in queue...
}
MyObject dummy = new MyObject();
dummy.setData("END");
queue.put(dummy);
//Note: empty queue here means work is done
while (queue.size() > 0) {
    synchronized (queue) {
        queue.wait(500); // over-cautios locking prevention i guess
    }
}

注意插入必须在同一个事务中，不能处理事务 通过主线程。

有什么更好的方法来做到这一点？ （我正在学习，不想开始“做错事”）

Answer 1

这些虚拟消息是有效的。它被称为“毒药”。生产者发送给消费者以使其停止的东西。

另一种可能性是在主线程中的某处调用 Thread.interrupt() 并在工作线程中相应地捕获和处理 InterruptedException。

Answer 2

通过到处使用“虚拟消息”来“解决”它。这是一个 馊主意？有哪些可能的解决方案？

这不是一个坏主意，它被称为“毒丸”，是停止基于线程的服务的合理方法。

但它只有在生产者和消费者的数量已知时才有效。

在您发布的代码中，有两个线程，一个是“主线程”，它产生数据，另一个是“处理线程”，它消耗数据，“毒丸”适用于这种情况。

但是想象一下，如果你还有其他生产者，那么消费者如何知道何时停止（只有当所有生产者都发送“毒丸”时），你需要确切知道所有生产者的数量，并检查数量消费者中“毒丸”的数量，如果等于生产者的数量，即所有生产者停止工作，则消费者停止。

在“主线程”中，您需要捕获InterruptedException，因为如果没有，“主线程”可能无法设置“毒丸”。你可以像下面那样做，

...
try {
    // do normal processing
} catch (InterruptedException e) { /*  fall through  */  }
finally {
    MyObject dummy = new MyObject();
    dummy.setData("END");
    ...
}
...

另外，您可以尝试使用ExecutorService 来解决您的所有问题。

（当你只需要做一些工作，然后在所有工作完成后停止）

void doWorks(Set<String> works, long timeout, TimeUnit unit)
    throws InterruptedException {
    ExecutorService exec = Executors.newCachedThreadPool();
    try {
        for (final String work : works)
            exec.execute(new Runnable() {
                    public void run() {
                        ...
                    }
                });
    } finally {
        exec.shutdown();
        exec.awaitTermination(timeout, unit);
    }
}

我正在学习，不想开始“做错事”

您可能需要阅读这本书：Java 并发实践。相信我，这是最好的。

Answer 3

你可以做的（我在最近的一个项目中做过）是包装队列，然后添加一个'isOpen()'method。

class ClosableQ<T> {

   boolean isOpen = true;

   private LinkedBlockingQueue<T> lbq = new LinkedBlockingQueue<T>();

   public void put(T someObject) {
      if (isOpen) {
         lbq.put(someObject);
      }
   }

   public T get() {
      if (isOpen) {
         return lbq.get(0);
      }
   }

   public boolean isOpen() {
      return isOpen;
   }

   public void open() {
      isOpen = true;
   }

   public void close() {
      isOpen = false;
   }
}

所以你的作家线程变成了这样的：

while (reader.hasNext() ) {
  // read the file and put it into the queue
  dataQ.put(someObject);
} 
// now we're done
dataQ.close();

和读者线程：

while (dataQ.isOpen) {
   someObject = dataQ.get();
}

您当然可以改为扩展列表，但这会为用户提供您可能不想要的访问级别。而且你需要在这段代码中添加一些并发的东西，比如 AtomicBoolean。