生产者-消费者；消费者如何停止？

Question

所以我已经模拟了我的生产者消费者问题，我有下面的代码。我的问题是:如果消费者处于常量while(true)中，他如何停止。

在下面的代码中，我添加了

                    if (queue.peek()==null)
                         Thread.currentThread().interrupt();

这在本例中工作得很好。但在我的实际设计中，这是行不通的(有时生产者需要更长的时间来‘放置’数据，所以抛给消费者的异常是不正确的。一般来说，我知道我可以放入一个“毒药”数据，比如Object is XYZ，我可以在消费者中检查它。但是这种毒药让代码看起来真的很糟糕。不知道是否有人有不同的方法。

public class ConsumerThread implements Runnable
{
 private BlockingQueue<Integer> queue;
 private String name;
 private boolean isFirstTimeConsuming = true;
 public ConsumerThread(String name, BlockingQueue<Integer> queue)
 {
    this.queue=queue;
    this.name=name;
 }

@Override
public void run()
{
    try
    {       
        while (true)
        {   
            if (isFirstTimeConsuming)
            {
                System.out.println(name+" is initilizing...");
                Thread.sleep(4000);
                isFirstTimeConsuming=false;
            }
            try{

                if (queue.peek()==null)
                    Thread.currentThread().interrupt();

                Integer data = queue.take();

                System.out.println(name+" consumed ------->"+data);
                Thread.sleep(70);    

            }catch(InterruptedException ie)
            {
                System.out.println("InterruptedException!!!!");
                break;
            }
        }

        System.out.println("Comsumer " + this.name + " finished its job; terminating.");

    }catch (InterruptedException e)
    {
        e.printStackTrace();
    } 
}

}

Answer 1

A:不能保证仅仅因为peek返回null，生产者就已经停止生产。如果制作人只是速度变慢了怎么办？现在，消费者退出了，生产者继续生产。所以“偷看”->“中断”的想法基本上失败了。

B:在以下情况下，从消费者设置‘完成/运行’标志并在生产者中读取它的也会失败：

1. 消费者检查标志，发现它应该继续运行，然后执行
2. 。同时，producer将标志设置为“不要运行”
3. 现在消费者将永远阻塞，等待幽灵数据包

相反的情况也可能发生，一个包被遗漏了，没有被使用。

然后，为了解决这个问题，你需要在'BlockingQueue‘之上对互斥进行额外的同步。

C:我发现'Rosetta Code‘是决定什么是好的实践的很好的来源，在这样的情况下：

http://rosettacode.org/wiki/Synchronous_concurrency#Java

生产者和消费者必须就表示输入结束的对象(或对象中的属性)达成一致。然后生产者在最后一个包中设置该属性，消费者停止消费它。也就是说，你在问题中提到的“毒药”。

在上面的Rosetta Code示例中，这个“对象”只是一个名为“EOF”的空String：

final String EOF = new String();

// Producer
while ((line = br.readLine()) != null)
  queue.put(line);
br.close();
// signal end of input
queue.put(EOF);

// Consumer
while (true)
  {
    try
      {
        String line = queue.take();
        // Reference equality
        if (line == EOF)
          break;
        System.out.println(line);
        linesWrote++;
      }
    catch (InterruptedException ie)
      {
      }
  }

Answer 2

不在线程上使用中断，而是在不再需要时中断循环：

if (queue.peek()==null)
         break;

或者，您也可以使用变量将关闭操作标记为挂起，然后在以下情况下中断循环并关闭循环：

if (queue.peek()==null)
         closing = true;

//Do further operations ...
if(closing)
  break;

Answer 3

在现实世界中，大多数消息传递都带有某种类型的头，它定义了消息类型/子类型，或者可能定义了不同的对象。

您可以创建命令和控制对象或消息类型，告诉线程在获得消息时执行某些操作(如关机、重新装入表、添加新的侦听器等)。

这样，您就可以让命令和控制线程将消息发送到正常消息流中。您可以让CNC线程与大型系统中的操作终端对话，等等。