Java并发编程---生产者-消费者模式

淡淡的烟草味﹌ 2022-06-02 07:05 169阅读 0赞

**一.概念**

**生产者和消费者也是一个非常经典的多线程模式,我们在实际开发中应用非常广泛的思想理念.在生产--消费模式中:通常由两类线程,即若干个生产者的线程和若干个消费者的线程.生产者线程负责提交用户请求,消费者线程则负责具体处理生产者提交的任务,在生产者和消费者之间通过共享内存缓冲区进行通信(该模式在MQ中使用比较广泛)**

**![Center][]**

**二.代码示例**

package com.thread.provider;
    
    public final class Data {
    
    	private String id;
    	private String 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 "Data [id=" + id + ", name=" + name + "]";
    	}
    	public Data(String id, String name) {
    		super();
    		this.id = id;
    		this.name = name;
    	}
    	
    }

package com.thread.provider;
    
    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<Data> queue){
    		this.queue = queue;
    	}
    	@Override
    	public void run() {
    		// TODO Auto-generated method stub
           while(isRunning){
        	   try {
        		//随机休眠0-1000毫秒,表示获取数据(产生数据的耗时)
    			Thread.sleep(r.nextInt(1000));
    			//获取的数据进行累计
    			int id = count.getAndIncrement();
    			//比如通过一个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("提交缓冲区数据失败...");
    			}
    			
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
           }
    	}
    	
    	public void stop(){
    		this.isRunning = false;
    	}
    
    }

package com.thread.provider;
    
    import java.util.Random;
    import java.util.concurrent.BlockingQueue;
    
    public class Consumer implements Runnable {
    
    	private BlockingQueue<Data> queue;
    	
    	public Consumer(BlockingQueue<Data> queue){
    		this.queue = queue;
    	}
    	//随机对象
    	private static Random r = new Random();
    	
    	@Override
    	public void run() {
    		// TODO Auto-generated method stub
           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) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
           }
    	}
    
    }

package com.thread.provider;
    
    import java.util.concurrent.BlockingDeque;
    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 {
    		// TODO Auto-generated method stub
    		//内存缓冲区
    		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);
    		
    		//创建线程池运行,这是一个缓存的线程池,可以创建无穷大的线程,没有任务的时候不创建线程,空闲线程的存活时间为60秒(默认值)
    		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 (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		p1.stop();
    		p2.stop();
    		p3.stop();
    		try {
    			Thread.sleep(2000);
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		
    //		cachePool.shutdown();
    //		cachePool.shutdownNow();
    	}
    
    }

**3.运行结果**

当前线程:pool-1-thread-1,获取了数据,id为:0,进行装载到缓冲区...
    当前消费线程: pool-1-thread-5消费成功,消费数据为id: 0
    当前线程:pool-1-thread-1,获取了数据,id为:1,进行装载到缓冲区...
    当前线程:pool-1-thread-1,获取了数据,id为:2,进行装载到缓冲区...
    当前线程:pool-1-thread-2,获取了数据,id为:3,进行装载到缓冲区...
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 1
    当前线程:pool-1-thread-3,获取了数据,id为:4,进行装载到缓冲区...
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 4
    当前线程:pool-1-thread-2,获取了数据,id为:5,进行装载到缓冲区...
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 2
    当前消费线程: pool-1-thread-5消费成功,消费数据为id: 3
    当前线程:pool-1-thread-3,获取了数据,id为:6,进行装载到缓冲区...
    当前线程:pool-1-thread-1,获取了数据,id为:7,进行装载到缓冲区...
    当前消费线程: pool-1-thread-5消费成功,消费数据为id: 7
    当前线程:pool-1-thread-2,获取了数据,id为:8,进行装载到缓冲区...
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 5
    当前线程:pool-1-thread-1,获取了数据,id为:9,进行装载到缓冲区...
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 6
    当前线程:pool-1-thread-2,获取了数据,id为:10,进行装载到缓冲区...
    当前线程:pool-1-thread-3,获取了数据,id为:11,进行装载到缓冲区...
    当前线程:pool-1-thread-1,获取了数据,id为:12,进行装载到缓冲区...
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 10
    当前消费线程: pool-1-thread-5消费成功,消费数据为id: 8
    当前线程:pool-1-thread-3,获取了数据,id为:13,进行装载到缓冲区...
    当前线程:pool-1-thread-2,获取了数据,id为:14,进行装载到缓冲区...
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 9
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 11
    当前线程:pool-1-thread-1,获取了数据,id为:15,进行装载到缓冲区...
    当前线程:pool-1-thread-1,获取了数据,id为:16,进行装载到缓冲区...
    当前线程:pool-1-thread-3,获取了数据,id为:17,进行装载到缓冲区...
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 13
    当前线程:pool-1-thread-1,获取了数据,id为:18,进行装载到缓冲区...
    当前消费线程: pool-1-thread-5消费成功,消费数据为id: 12
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 15
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 14
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 18
    当前线程:pool-1-thread-2,获取了数据,id为:19,进行装载到缓冲区...
    当前消费线程: pool-1-thread-6消费成功,消费数据为id: 19
    当前消费线程: pool-1-thread-4消费成功,消费数据为id: 17
    当前消费线程: pool-1-thread-5消费成功,消费数据为id: 16

[Center]: /images/20220602/afd0dba064d143b1a06ce38a0a746ac0.png