多生产者多消费者多线程Java

Question

我正在尝试多生产者 - 生产者 - 消费者问题的多个消费者使用案例.我正在使用BlockingQueue在多个生产者/消费者之间共享公共队列.

以下是我的代码.
制片人

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

    private BlockingQueue inputQueue;
    private static volatile int i = 0;
    private volatile boolean isRunning = true;

    public Producer(BlockingQueue q){
        this.inputQueue=q;
    }

    public synchronized void run() {

        //produce messages
        for(i=0; i<10; i++) 
        {
            try {
                inputQueue.put(new Integer(i));

                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("Produced "+i);
        }
        finish();
    }

    public void finish() {
        //you can also clear here if you wanted
        isRunning = false;
    }

}

---

消费者

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {

    private BlockingQueue inputQueue;
    private volatile boolean isRunning = true;

    private final Integer POISON_PILL = new Integer(-1);

    Consumer(BlockingQueue queue) {
        this.inputQueue = queue;
    }

    public void run() {
        //worker loop keeps taking en element from the queue as long as the producer is still running or as 
        //long as the queue is not empty:
        while(!inputQueue.isEmpty()) {

            try {
                Integer queueElement = (Integer) inputQueue.take();
                System.out.println("Consumed : " + queueElement.toString());

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.out.println("Queue ");
    }

    //this is used to signal from the main thread that he producer has finished adding stuff to the queue
    public void finish() {
        //you can also clear here if you wanted
        isRunning = false;
        inputQueue.add(POISON_PILL);
    }
}

---

测试类

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;


public class ProducerConsumerService {

    public static void main(String[] args) {

        //Creating BlockingQueue of size 10
        BlockingQueue queue = new ArrayBlockingQueue(10);

        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        //starting producer to produce messages in queue
        new Thread(producer).start();

        //starting producer to produce messages in queue
        new Thread(producer).start();

        //starting consumer to consume messages from queue
        new Thread(consumer).start();

        //starting consumer to consume messages from queue
        new Thread(consumer).start();

        System.out.println("Producer and Consumer has been started");
    }

}

运行以下代码时,我看不到正确的输出.

我在这里做错了吗？

Answer 1

你的代码中有相当一部分没有意义.我建议你坐下来弄清楚为什么代码在那里以及它在做什么.

如果删除了isFinshed标志,则不会发生任何变化.

如果您删除了synchronized生产者中的使用,您将拥有并发生产者.创建仅在同步块volatile中访问的字段没有任何好处.

对于生产者来说,共享循环计数器是没有意义的,如果它们是并发的.通常,生产者发送毒丸,而消费者不会消费药丸.例如,如果你有两个消费者,一个可能会添加药丸,另一个可能会消耗它.你的消费者忽略毒丸,因为它忽略了isFinished旗帜.

您不希望仅因为队列暂时为空而停止使用者.否则它将不会看到生产者产生的所有消息,可能没有消息.

Answer 2

包含多个生产者和多个消费者的示例代码

import java.util.concurrent.*;

public class ProducerConsumerDemo {

    public static void main(String args[]){

     BlockingQueue<Integer> sharedQueue = new LinkedBlockingQueue<Integer>();

     Thread prodThread1 = new Thread(new Producer(sharedQueue,1));
     Thread prodThread2 = new Thread(new Producer(sharedQueue,2));
     Thread consThread1 = new Thread(new Consumer(sharedQueue,1));
     Thread consThread2 = new Thread(new Consumer(sharedQueue,2));

     prodThread1.start();
     prodThread2.start();
     consThread1.start();
     consThread2.start();
    }

}

class Producer implements Runnable {

    private final BlockingQueue<Integer> sharedQueue;
    private int threadNo;

    public Producer(BlockingQueue<Integer> sharedQueue,int threadNo) {
        this.threadNo = threadNo;
        this.sharedQueue = sharedQueue;
    }

    @Override
    public void run() {
        for(int i=1; i<= 5; i++){
            try {
                int number = i+(10*threadNo);
                System.out.println("Produced:" + number + ":by thread:"+ threadNo);
                sharedQueue.put(number);
            } catch (Exception err) {
                err.printStackTrace();
            }
        }
    }

}

class Consumer implements Runnable{

    private final BlockingQueue<Integer> sharedQueue;
    private int threadNo;
    public Consumer (BlockingQueue<Integer> sharedQueue,int threadNo) {
        this.sharedQueue = sharedQueue;
        this.threadNo = threadNo;
    }

    @Override
    public void run() {
        while(true){
            try {
                int num = sharedQueue.take();
                System.out.println("Consumed: "+ num + ":by thread:"+threadNo);
            } catch (Exception err) {
               err.printStackTrace();
            }
        }
    }   
}

输出:

Produced:11:by thread:1
Produced:21:by thread:2
Produced:22:by thread:2
Produced:23:by thread:2
Produced:24:by thread:2
Produced:25:by thread:2
Consumed: 11:by thread:1
Consumed: 22:by thread:1
Consumed: 23:by thread:1
Consumed: 24:by thread:1
Consumed: 25:by thread:1
Produced:12:by thread:1
Consumed: 21:by thread:2
Consumed: 12:by thread:1
Produced:13:by thread:1
Produced:14:by thread:1
Produced:15:by thread:1
Consumed: 13:by thread:2
Consumed: 14:by thread:1
Consumed: 15:by thread:2

本文提供了BlockingQueue的简单示例生产者和消费者问题

您的代码更改:

1. 不同的生产者将生成不同的输出而不是相同的输出 生产者线程1生成11-15的数字,生产者线程2生成21-25的数字
2. 任何Consumer线程都可以使用任何Producer中的数据.与生产者不同,消费者没有使用数据的约束.
3. 我在Producer和Consumer中都添加了Thread号.

您可以在以下位置找到替代方案ExecutorService:

使用队列的生产者/消费者线程