Java 生产者消费者代码

比眉伴天荒 2023-02-23 01:28 48阅读 0赞

### 介绍 ###

生产者和消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此并不直接通信，而是通过阻塞队列进行通信，所以生产者生产完数据后不用等待消费者进行处理，而是直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列中获取数据，阻塞队列就相当于一个缓冲区，平衡生产者和消费者的处理能力。

### 单个线程实现生产者消费者 ###

生产者：

import java.util.List;
    
    public class Producer implements Runnable{
        private List<Integer> queue;
        private int n;
    
        public Producer(List<Integer> queue,int n){
            this.queue=queue;
            this.n =n;
        }
    
        public void main() throws InterruptedException {
            for (int i = 0; i < n; i++) {
                synchronized (queue) {
                    while (queue.size() == 5) {
                        System.out.println(Thread.currentThread().getName() + " i= " + i);
                        queue.wait();
                    }
                    queue.add(i);
                    queue.notify();
                }
            }
        }
    
        @Override
        public void run() {
            try {
                main();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

消费者：

import java.util.List;
    
    public class Consumer implements Runnable{
        private List<Integer> queue;
    
        public Consumer(List<Integer> queue){
            this.queue = queue;
        }
        public void main() throws InterruptedException {
            while(true) {
    
                synchronized (queue) {
                    while (queue.isEmpty()) {
                        System.out.println(" consumer ");
                        queue.wait();
                    }
                    System.out.println(" consumer " + Thread.currentThread().getName() + "  n= " + queue.get(0));
                    queue.clear();
                    queue.notify();
                }
            }
        }
    
        @Override
        public void run() {
            try {
                main();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

测试代码：

@Test
        public void testProducerConsumer(){
            List<Integer> queue = Lists.newArrayList();
            Thread thread1=new Thread(new Producer(queue,30));
            thread1.setName("生产者线程 ");
    
            Thread thread2=new Thread(new Consumer(queue));
            thread2.setName("......消费者线程 ");
    
            thread2.start();
            thread1.start();
        }

### 1个生产、多个消费 ###

生产者：

/**
     *
     */
    public class MuitiProducer implements Runnable {
    
        private GoodsQueue basket;
    
        private int n;
    
        public MuitiProducer(GoodsQueue basket,int n) {
            this.basket = basket;
            this.n=n;
        }
    
        @Override
        public void run() {
            mainTwo();
        }
    
        private void main() {
            for (int i = 0; i < n; i++) {
                GoodsDTO good = new GoodsDTO(i + 1, "面包");
                System.out.println(Thread.currentThread().getName()
                        + "生产了" + good.getId() + "号" + good.getName());
                basket.push(good);
            }
        }
    
        private void mainTwo() {
            basket.setProducerOver(Boolean.FALSE);
            for (int i = 0; i < n; i++) {
                synchronized (basket) {
                    GoodsDTO good = new GoodsDTO(i + 1, "面包");
                    System.out.println(Thread.currentThread().getName()
                            + "生产了" + good.getId() + "号" + good.getName());
    
                    //条件成立说明篮子放满了
                    if (basket.getIndex() == basket.getGoodsList().length) {
                        try {
                            //wait()后，线程会将持有的锁释放，进入阻塞状态
                            //这样其他需要锁的线程就可以获得锁
                            basket.wait();
                            //这里的含义就是执行此方法的线程暂停，进入阻塞状态
                            //等消费者消费了产品后再生产
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    basket.getGoodsList()[basket.getIndex()] = good;
                    basket.setIndex(basket.getIndex()+1);
                    //唤醒在当前对象等待池中等待的第一个线程
                    //notifyAll()唤醒所有在当前对象等待池中等待的所有线程
                    basket.notifyAll();
                }
            }
            basket.setProducerOver(Boolean.TRUE);
        }
    }

消费者

public class MuitiConsumer implements Runnable{
    
        private GoodsQueue basket;
    
        public MuitiConsumer(GoodsQueue basket) {
            this.basket = basket;
        }
    
        @Override
        public void run() {
            main();
        }
    
        /**
         * 两种方式：随机消费
         */
        public void main(){
            for (int i = 0; i < 10; i++) {
                GoodsDTO good = basket.pop();
                System.out.println(Thread.currentThread().getName()
                        + "吃了" + good.getId() + "号" + good.getName());
            }
        }
    
        /**
         * 一次消费 5个
         */
        public void mainTwo() {
            while (basket.isProducerOver() == Boolean.FALSE) {
                synchronized (basket) {
                    for (int i = 0; i < 5; i++) {
                        while (basket.getIndex() == 0) {
                            try {
                                basket.wait();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        basket.setIndex(basket.getIndex() - 1);
                        GoodsDTO good = basket.getGoodsList()[basket.getIndex()];
                        System.out.println(Thread.currentThread().getName()
                                + "吃了" + good.getId() + "号" + good.getName());
                        basket.notifyAll();
                    }
                }
            }
        }
    }

测试代码

@Test
        public void testMultiProducerConsumer() throws InterruptedException {
            GoodsQueue basket = new GoodsQueue();
            int n = 30;
            Thread thread1 = new Thread(new MuitiProducer(basket, n));
            thread1.setName(" 生产者线程 ");
            thread1.start();
    
            for (int i = 1; i < 4; i++) {
                Thread thread2 = new Thread(new MuitiConsumer(basket));
                thread2.setName("......消费者线程_" + i + "_");
                thread2.start();
            }
        }

额外处理类：

public class GoodsQueue {
    
        public GoodsDTO[] getGoodsList() {
            return goodsList;
        }
    
        public void setGoodsList(GoodsDTO[] goodsList) {
            this.goodsList = goodsList;
        }
    
        private GoodsDTO[] goodsList = new GoodsDTO[15];
    
        public int getIndex() {
            return index;
        }
    
        public void setIndex(int index) {
            this.index = index;
        }
    
        private int index;//记录篮子中面包的个数
    
    
        public boolean isProducerOver() {
            return producerOver;
        }
    
        public void setProducerOver(boolean producerOver) {
            this.producerOver = producerOver;
        }
    
        /**
         * 生产结束
         */
        private boolean producerOver;
    
        //private List<GoodsDTO> goodsList;
        public synchronized void push(GoodsDTO good) {
            //条件成立说明篮子放满了
            if (index == goodsList.length) {
                try {
                    //wait()后，线程会将持有的锁释放，进入阻塞状态
                    //这样其他需要锁的线程就可以获得锁
                    this.wait();
                    //这里的含义就是执行此方法的线程暂停，进入阻塞状态
                    //等消费者消费了产品后再生产
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            goodsList[index] = good;
            index++;
            //唤醒在当前对象等待池中等待的第一个线程
            //notifyAll()唤醒所有在当前对象等待池中等待的所有线程
            this.notifyAll();
            //如果不唤醒，以后这两个线程都会进入等待线程，没有人唤醒
        }
    
        public synchronized GoodsDTO pop() {
            //如果篮子空了就暂停消费线程，等生产线程生产出产品再消费
            if (index == 0) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            index--;
            this.notifyAll();
            return goodsList[index];
        }
    }
    
    
    
    import java.util.List;
    
    public class GoodsDTO {
    
        public GoodsDTO(int id,String name){
            this.id=id;
            this.name=name;
        }
    
        private List<Long> goodsIdList;
    
        private int id;
    
        private String name;
    
        public int getId() {
            return id;
        }
    
        public void setId(int id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    }

### BlockingQueue实现的生产者-消费者 ###

BlockingQueue自身就是线程安全的，因为我们不用考虑 并发问题，代码如下：

生产者：

import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class BlockingQueueProducer implements Runnable {
    
        private String name;
        private volatile boolean isRunning = true;
    
        private BlockingQueue queue;
        private static AtomicInteger count = new AtomicInteger();
        private static final int DEFAULT_RANGE_FOR_SLEEP = 100;
    
        public BlockingQueueProducer(BlockingQueue queue,String name) {
            this.queue = queue;
            this.name=name;
        }
    
        public void run() {
            String data = null;
            Random r = new Random();
            try {
                while (isRunning) {
                    Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
                    data = "data:" + count.incrementAndGet();
                    System.out.println(this.name+" 生产  " + data);
                    if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                        System.out.println("放入数据失败：" + data);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            } finally {
                System.out.println(Thread.currentThread().getName()+" 退出生产者线程！");
            }
        }
    
        public void stop() {
            isRunning = false;
        }
    }

消费者：

import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    public class BlockingQueueConsumer implements Runnable {
    
        private String name;
        private BlockingQueue<String> queue;
        private static final int DEFAULT_RANGE_FOR_SLEEP = 100;
    
        public BlockingQueueConsumer(BlockingQueue<String> queue,String name) {
            this.queue = queue;
            this.name = name;
        }
    
        public void run() {
            Random r = new Random();
            boolean isRunning = true;
            try {
                while (isRunning) {
                    String data = queue.poll(2, TimeUnit.SECONDS);
                    if (null != data) {
                        System.out.println(this.name+" 消费数据  " + data);
                        Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
                    } else {
                        // 超过2s还没数据，认为所有生产线程都已经退出，自动退出消费线程。
                        isRunning = false;
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            } finally {
                System.out.println(Thread.currentThread().getName()+ " 退出消费者线程！");
            }
        }
    }

测试类：

@Test
        public void blockingQueueTest() throws InterruptedException {
            // 声明一个容量为10的缓存队列
            BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
    
            BlockingQueueProducer producer1 = new BlockingQueueProducer(queue,"1");
            BlockingQueueProducer producer2 = new BlockingQueueProducer(queue,"2");
            BlockingQueueProducer producer3 = new BlockingQueueProducer(queue,"3");
    
            BlockingQueueConsumer consumer = new BlockingQueueConsumer(queue,"0001");
            BlockingQueueConsumer consumer2 = new BlockingQueueConsumer(queue,"0002");
    
            // 借助Executors
            ExecutorService service = Executors.newCachedThreadPool();
            // 启动线程
            service.execute(producer1);
            service.execute(producer2);
            service.execute(producer3);
            service.execute(consumer);
            service.execute(consumer2);
    
            // 执行10s
            Thread.sleep(2 * 1000);
            producer1.stop();
            producer2.stop();
            producer3.stop();
    
            Thread.sleep(1000);
            // 退出Executor
            service.shutdown();
        }

### 参考博客 ###

[多线程-生产者和消费者模式的四种实现][-]  
[线程并发协作(生产者/消费者模式)][Link 1]  
[BlockingQueue][]

[-]: https://zhuanlan.zhihu.com/p/74539919
[Link 1]: https://zhuanlan.zhihu.com/p/74718222
[BlockingQueue]: https://www.iteye.com/blog/wsmajunfeng-1629354