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

Question

所以我模拟了我的生产者消费者问题,我有下面的代码.我的问题是:如果消费者处于不变的状态,消费者将如何停止(真实).

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

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

这个例子很好用.但在我的真实世界设计中,这不起作用(有时生产者需要更长的时间来"放置"数据,因此消费者抛出的异常是不正确的.一般来说,我知道我可以放一个'毒'数据比如Object是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,生产商已停止生产.如果生产者放慢速度怎么办？现在,消费者退出,生产者继续生产.所以'偷看' - >'休息'的想法基本上都失败了.

B:在消费者中设置'done/run'标志并在生产者中读取它也会失败,如果:

1. 消费者检查旗帜,发现它应该继续运行,然后做"拿"
2. 与此同时,制片人正在设置"不要跑"的旗帜
3. 现在消费者永远等待鬼包

相反的情况也可能发生,并且一个数据包被排除在外.

然后,为了解决这个问题,你需要在'BlockingQueue'之上与互斥锁进行额外的同步.

C: 我觉得'Rosetta Code'是决定什么是好习惯的好来源,在这种情况下:

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

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

在上面的Rosetta Code示例中,这个'对象'只是一个String名为'EOF' 的空:

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)
      {
      }
  }