并发编程(13)-生产者消费者

偏执的太偏执、 2022-06-06 11:21 174阅读 0赞

生产者和消费者是一个非常经典的多线程模式，我们在实际开发中应用非常广泛的思想理念。在生产-消费模式中：通常由两类线程，即若干个生产者的线程和若干个消费者的线程。生产者线程负责提交用户请求，消费者线程则负责具体处理生产者提交的任务，在生产者和消费者之间通过共享内存缓存区进行通信。  
![这里写图片描述][SouthEast]

创建队列需要的data类：

public final class Data { 
    
        private String id;
        private String name;
    
        public Data(String id, String name){
            this.id = id;
            this.name = name;
        }
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        @Override
        public String toString(){
            return "{id: " + id + ", name: " + name + "}";
        }
    
    }

创建生产者：

import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class Provider implements Runnable{ 
        //共享缓存区
        private BlockingQueue<Data> queue;
        //多线程间是否启动变量，有强制从主内存中刷新的功能。即时返回线程的状态
        private volatile boolean isRunning = true;
        //id生成器
        private static AtomicInteger count = new AtomicInteger();
        //随机对象
        private static Random r = new Random(); 
    
        public Provider(BlockingQueue queue){
            this.queue = queue;
        }
    
        public void run() {
            while(isRunning){
                try {
                    //随机休眠0 - 1000 毫秒 表示获取数据(产生数据的耗时) 
                    Thread.sleep(r.nextInt(1000));
                    //获取的数据进行累计...
                    int id = count.incrementAndGet();
                    //比如通过一个getData方法获取了
                    Data data = new Data(Integer.toString(id), "数据" + id);
                    System.out.println("当前线程:" + Thread.currentThread().getName() + ", 获取了数据，id为:" + id + ", 进行装载到公共缓冲区中...");
                    if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
                        System.out.println("提交缓冲区数据失败....");
                        //do something... 比如重新提交
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        public void stop(){
            this.isRunning = false;
        }   
    }

创建消费者：

package com.thread.mythread.conn017;
    
    import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    public class Consumer implements Runnable{ 
    
        private BlockingQueue<Data> queue;
    
        public Consumer(BlockingQueue queue){
            this.queue = queue;
        }
    
        //随机对象
        private static Random r = new Random(); 
    
        public void run() {
            while(true){
                try {
                    //获取数据
                    Data data = this.queue.take();
                    //进行数据处理。休眠0 - 1000毫秒模拟耗时
                    Thread.sleep(r.nextInt(1000));
                    System.out.println("当前消费线程：" + Thread.currentThread().getName() + "， 消费成功，消费数据为id: " + data.getId());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

创建测试：

package com.thread.mythread.conn017;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class Main { 
    
        public static void main(String[] args) throws Exception {
            //内存缓冲区
            BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(10);
            //生产者
            //如果想往队列中放入和取出数据，生产者和消费者都需要持有该队列的引用，所以这里讲队列出入到每个生产者和消费者
            Provider p1 = new Provider(queue);
            Provider p2 = new Provider(queue);
            Provider p3 = new Provider(queue);
            //消费者
            Consumer c1 = new Consumer(queue);
            Consumer c2 = new Consumer(queue);
            Consumer c3 = new Consumer(queue);
    
            //创建线程池运行,这是一个缓存的线程池，可以创建无穷大的线程，没有任务的时候不创建线程。空闲线程存活时间为60s（默认值）
            ExecutorService cachePool = Executors.newCachedThreadPool();
            //启动线程
            cachePool.execute(p1);
            cachePool.execute(p2);
            cachePool.execute(p3);
            cachePool.execute(c1);
            cachePool.execute(c2);
            cachePool.execute(c3);
    
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //生产者完成生产后，将生产者关闭掉
            p1.stop();
            p2.stop();
            p3.stop();
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }       
    // cachePool.shutdown(); 
    // cachePool.shutdownNow();
        }
    
    }

[SouthEast]: /images/20220606/cebbd9d807574574b0e0af957a3e83d9.png