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

Question

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

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

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

在这个例子中效果很好。但在我的现实世界设计中，这是行不通的(有时生产者需要更长的时间来“放置”数据，因此在消费者中抛出的异常是不正确的。一般来说，我知道我可以放置“毒”数据例如对象是 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

答: 根本无法保证仅仅因为 peek 返回 null，生产者就停止了生产。如果生产者只是放慢了速度怎么办？现在，消费者退出，生产者继续生产。所以 'peek' -> 'break' 的想法基本上失败了。

B:从消费者设置“完成/运行”标志并在生产者中读取它也会失败，如果:

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

相反的情况也可能发生，一个数据包被遗漏未被消费。

然后为了解决这个问题，您需要在“BlockingQueue”之上使用互斥体进行额外的同步。

C:在这种情况下，我发现“Rosetta 代码”是决定什么是好的做法的良好来源:

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

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

在上面的 Rosetta 代码示例中，这个“对象”只是一个名为“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)
      {
      }
  }