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

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 Concurrency in Practice。相信我，这是最好的。

Answer 3

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

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。