并发控制(8)使用executor并发框架来实现的生产消费者模型

亦凉 2022-07-16 23:21 79阅读 0赞

### 并发控制(8)使用executor并发框架来实现的生产消费者模型 ###

executoer 本质上就是个线程池 根据具体业务场景，分为了

*  public static ExecutorService newFixedThreadPool(int nThreads)  
    创建固定数目线程的线程池。
 *  public static ExecutorService newCachedThreadPool()  
    创建一个可缓存的线程池，调用execute 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。
 *  public static ExecutorService newSingleThreadExecutor()  
    创建一个单线程化的Executor。
 *  public static ScheduledExecutorService  
    创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Timer类。

--------------------

#### 有如下代码，分为生产者 消费者 数据定义 和最后的测试类，等4个类 ####

##### 消费者Consumer.java #####

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(); 
    
        @Override
        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();
                }
            }
        }
    }

##### 生产者Provider.java #####

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;
        }
    
        @Override
        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;
        }
    
    }

##### 消费的类型定义文件 Data.java #####

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 + "}";
        }
    
    }

##### 测试的主函数文件Main.java #####

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();
    
    
        }
    
    }