在生产者线程之后结束消费者线程并消耗所有元素答案

【问题标题】：end consumer thread after producer thread and consuming all the elements在生产者线程之后结束消费者线程并消耗所有元素
【发布时间】：2015-07-17 10:49:41
【问题描述】：

我已经编写了代码来使用等待和通知来实现生产者消费者问题。它工作正常，但问题是消费者线程正在无限循环中运行，并且即使在生产者线程完成并且消费者已经消耗了列表中的所有元素之后，它也会继续等待。

public class Practice {

    public static void main(String[] args) {

        List<Employee> empList = new ArrayList<Employee>();
        Thread producer = new Thread(new Producer(empList , 2) , "Producer");
        Thread consumer = new Thread(new Consumer(empList , 2) , "Consumer");

        producer.start();
        consumer.start();
    }

}

class Employee 
{
    private String name;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

}
class Producer implements Runnable 
{
    List<Employee> empList; 
    int size;

    public Producer(final List<Employee> empList  , final int size)
    {
        this.empList = empList;
        this.size = size;
    }

    @Override    
    public void run()
    {
        for(int i=0; i<5;i++)
        {
            try {
                produce(new Employee());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void produce(Employee e) throws InterruptedException
    {
        synchronized(empList){
            while(empList.size()==size) // If list is full then will have to wait
            {

                System.out.println("List is full "+Thread.currentThread().getName()+" Is waiting and" + " Size is "+empList.size());
                empList.wait();
            }
        }

        synchronized(empList)
        {
            System.out.println("Producing");
            empList.add(e);
            empList.notifyAll();
        }
    }
}


class Consumer implements Runnable 
{
    List<Employee> empList; 
    int size;

    public Consumer(final List<Employee> empList  , final int size)
    {
        this.empList = empList;
        this.size = size;
    }

    @Override    
    public void run()
    {
        while(true)
        {
            try {
                System.out.println("Consumed ");
                consume();
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void consume() throws InterruptedException
    {
        synchronized(empList){
            while(empList.isEmpty()) // If list is empty then will have to wait
            {
                System.out.println("List is empty "+Thread.currentThread().getName()+" Is waiting and " + "Size is "+empList.size());
                empList.wait();
            }
        }

        synchronized(empList)
        {
            empList.notifyAll();
            empList.remove(0);
        }

    }    

}

请告诉我如何在生产者完成并且消费者消耗完列表中的所有元素后停止我的消费者线程。请帮我写代码。提前致谢

【问题讨论】：

你怎么知道制片人什么时候结束？
@PM77-1 执行生产者的 run 方法后，它结束了，因为它只循环了 5 次。消费完所有元素后，消费者放弃锁定并等待任何通知到来，但没有通知到来，因此它继续等待。如果我错了，请纠正我
所以你想让消费者超时？
是的......但只有在生产者完成并消耗所有元素之后。
您无法预测消费者在生产者完成循环后开始工作。一旦你启动两个线程就可以并行运行。当生产者结束生产时，您无法做出预测，除非您将其按顺序进行。如果将其设为顺序，则无需锁定或使用单独的线程。

标签： java multithreading producer-consumer

【解决方案1】：

只需添加标志，在生产者和消费者之间共享：

static bool producerFinished = false;

在生成所有元素后在生产者中设置：

public void run()
{
    // ...
    synchronized(empList){
        producerFinished = true;
        empList.notifyAll();
    }
}

并在消费者的 while() 循环中检查它以及列表是否为空

public void consume()
{
    synchronized(empList){
        while(empList.isEmpty() && !producerFinished)
        {
            empList.wait();
        }

        if(!empList.isEmpty())
        {
            //consume element
        }
        else
        {
            //List is empty and producer has finished.
        }
    }
}

【讨论】：

技术上，producerFinished 应该声明为volatile。
@tariksbl: 因为这个标志是在锁定下访问的，所以不需要声明volatile。

【解决方案2】：

生产者-消费者的习惯用法是毒丸模式：生产者在完成生产后将一个魔法值放入队列中，然后退出。当消费者读取魔法值时，它会退出。

毒丸定义在生产者和消费者之间共享：

static final Employee POISON_PILL = new Employee();

制作人：

public void run()
{
    for(int i=0; i<5;i++)
    {
        produce(new Employee());
    }
    produce(POISON_PILL);
}

消费者：

public void consume() throws InterruptedException
{
    synchronized(empList){
        while(empList.isEmpty())
        {
            empList.wait();
        }
    }

    synchronized(empList)
    {
        // Note == not .equals() is intended here
        if (empList.remove(0) == POISON_PILL) {
            // done consuming
        }
        empList.notifyAll();
    }

}

【讨论】：