架构师入门笔记四初识多线程设计模式

本是古典何须时尚 2022-06-10 02:23 141阅读 0赞

![aHR0cHM6Ly9pbWcyMDIwLmNuYmxvZ3MuY29tL2Jsb2cvODA2OTU2LzIwMjAwNy84MDY5NTYtMjAyMDA3MzEyMzQ1MDc3MjItMTExNjgwOTkyMi5naWY][]

**架构师入门笔记四 初识多线程设计模式**

这章讲解常用的并行设计模式，为后续的多线程打基础。常用的并行设计模式有：Future 模式、Master-Worker模式 和 生产者-消费者模式。其中生产者-消费者模式是最为经典的常用模型。（注：并行设计模式是对一些常用的多线程结构的总结和抽象，属于设计优化的一部分）

# 1 Future 模式 #

## 1.1 核心思想  ##

除去主线程等待时间，将其等待时间处理其他业务逻辑。这也是它的应用场景

## 1.2 原理图 ##

借用网上的图片，前者是传统模式，后者是Future模式

![20170808115214693][]

## 1.3 模拟 Futrue 模式 代码 ##

参考：  [http://www.cnblogs.com/winkey4986/p/6203225.html][http_www.cnblogs.com_winkey4986_p_6203225.html]  （写的很好，代码一定要自己敲才会有感觉）

# 2 Master-Worker模式  #

## 2.1 核心思想 ##

系统由两类进程协作工作：Master负责接收和分配任务，Worker负责处理Master分配的任务。当每个Worker子进程处理完成后，会将结果返回给Master，再由Master做数据的归纳总结。这种将大任务分解成若干个小任务，再由多个小任务并发完成的设计模式，从很大程度上提高了系统的吞吐量。

## 2.2 原理图 ##

![20170808122518860][]

## 2.3 模拟Master-Worker代码 ##

首先分析一下Master 端具体需要做什么？

1）存放任务的容器；

2）存放分配子任务的普通集合 HashMap，key为线程名字，value为对应的线程；

3）存放收集子任务返回结果的集合，该集合必须是线程安全的 ConcurrentHashMap。（为什么分配子任务时不考虑线程安全呢？因为在分配任务时，是一对多的场景。而在收集结果的时候，是多对一的场景，存在高并发的情况。）

4）Master的构造器，分配 N 个Worker 线程；

5）检查Worker线程是否全部执行成功；

6）最后一步，收集总结返回的结果

import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    public class Master {
    
    	//1 有一个盛放任务的容器
    	private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
    	//2 需要有一个盛放worker的集合
    	private HashMap<String, Thread> workers = new HashMap<String, Thread>();
    	//3 需要有一个盛放每一个worker执行任务的结果集合
    	private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
    	
    	//4 构造方法
    	public Master(Worker worker , int workerCount){
    		worker.setWorkQueue(this.workQueue);
    		worker.setResultMap(this.resultMap);
    		for(int i = 0; i < workerCount; i ++){
    			this.workers.put(Integer.toString(i), new Thread(worker));
    		}
    	}
    	
    	//5 需要一个提交任务的方法
    	public void submit(Task task){
    		this.workQueue.add(task);
    	}
    	
    	//6 需要有一个执行的方法，启动所有的worker方法去执行任务
    	public void execute(){
    		for(Map.Entry<String, Thread> me : workers.entrySet()){
    			me.getValue().start();
    		}
    	}
    
    	//7 判断是否运行结束的方法
    	public boolean isComplete() {
    		for(Map.Entry<String, Thread> me : workers.entrySet()){
    			if(me.getValue().getState() != Thread.State.TERMINATED){
    				return false;
    			}
    		}		
    		return true;
    	}
    
    	//8 计算结果方法
    	public int getResult() {
    		int priceResult = 0;
    		for(Map.Entry<String, Object> me : resultMap.entrySet()){
    			priceResult += (Integer)me.getValue();
    		}
    		return priceResult;
    	}
    	
    }

Worker 端就比较简单了

1）也要一个存放任务的容器；

2）存放结果的ConcurrentHashMap

import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    public class Worker implements Runnable {
    
    	private ConcurrentLinkedQueue<Task> workQueue;
    	private ConcurrentHashMap<String, Object> resultMap;
    	
    	public void setWorkQueue(ConcurrentLinkedQueue<Task> workQueue) {
    		this.workQueue = workQueue;
    	}
    
    	public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
    		this.resultMap = resultMap;
    	}
    	
    	@Override
    	public void run() {
    		while(true){
    			Task input = this.workQueue.poll(); // 从头部取出元素，并从队列里删除
    			if(input == null) break;
    			Object output = handle(input);
    			this.resultMap.put(input.getId().toString(), output); // key是对象id，value是对象price
    		}
    	}
    
    	private Object handle(Task input) {
    		Object output = null;
    		try {
    			//处理任务的耗时。。 比如说进行操作数据库。。。
    			Thread.sleep(500);
    			output = input.getPrice();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		return output;
    	}
    
    }

Task 模拟的是实际开发中的数据对象：

public class Task {
    	private Integer id;
    	private Integer price;
    	
    	public Task() {}
    	
    	public Task(Integer id, Integer price) {
    		this.id = id;
    		this.price = price;
    	}
    
    	public Integer getId() {
    		return id;
    	}
    
    	public void setId(Integer id) {
    		this.id = id;
    	}
    
    	public Integer getPrice() {
    		return price;
    	}
    
    	public void setPrice(Integer price) {
    		this.price = price;
    	}
    }

最后是启动程序Main方法：

public class Main {
    	public static void main(String[] args) {
    		// 分配20个子任务
    		Master master = new Master(new Worker(), 20);
    		// 对象赋值
    		for(int i = 1; i <= 100; i++){
    			Task t = new Task();
    			t.setId(i);
    			t.setPrice(i*10);
    			master.submit(t);
    		}
    		// Master启动worker去执行任务
    		master.execute();
    		long start = System.currentTimeMillis();
    		while(true){
    			// 判断Worker子任务是否全部执行成功
    			if(master.isComplete()){
    				int priceResult = master.getResult(); // 获得master 收集总结的值
    				long end = System.currentTimeMillis() - start;
    				System.out.println("最终结果：" + priceResult + ", 执行时间：" + end);
    				break;
    			}
    		}	
    	}
    }

总结：Master容器，接收任务，然后将任务细化分给若干个子任务。当系统检查到全部子任务执行完成后，Master容器收集结果，给出答复。（打个比方：100人报数，假设每个报数需要一秒的时间，若从头到尾报数则要100秒。如果将100人分成10组，每组同时进行，最后可能只需10几秒就完成任务。）

# 3 生产者-消费者模式 #

## 3.1 核心思想 ##

通常由两类线程，即若干个生产者线程和若干个消费者线程。生产者线程负责提交用户请求，消费者线程负责具体处理生产者提交的任务，在生产者和消费者之间通过共享内存缓存区进行通信（其核心组件是共享内存缓冲区，是两者的通信桥梁，起到解耦作用，优化系统整体结构。）。

## 3.2 原理图 ##

![20170808153359479][]

## 3.3 模拟生产者-消费者代码 ##

首先是生产者：

1）要有一个内存缓冲区

2）要有一个向内存缓冲区（队列）中加入数据的操作

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<Task> queue;
    	//多线程间是否启动变量，有强制从主内存中刷新的功能。即时返回线程的状态
    	private volatile boolean isRunning = true;
    	//id生成器 Atomic 单个方法执行是线程安全的
    	private static AtomicInteger count = new AtomicInteger();
    	//随机对象
    	private static Random r = new Random(); 
    	
    	public Provider(BlockingQueue<Task> queue){
    		this.queue = queue;
    	}
    
    	@Override
    	public void run() {
    		while(isRunning){
    			try {
    				//随机休眠0 - 1000 毫秒 表示获取数据(产生数据的耗时) 
    				Thread.sleep(r.nextInt(1000));
    				//获取的数据进行累计...
    				int id = count.incrementAndGet();
    				Task task = new Task(id,  10*id);
    				System.out.println("当前线程:" + Thread.currentThread().getName() + ", 获取了数据，id为:" + id + ", 进行装载到公共缓冲区中...");
    				// 指定时间内将数据加入队列中
    				if(!this.queue.offer(task, 2, TimeUnit.SECONDS)){
    					System.out.println("提交缓冲区数据失败....");
    					//do something... 比如重新提交
    				}
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    	public void stop(){
    		this.isRunning = false;
    	}
    	
    }

消费者：

1）要有一个内存缓冲区

2）要有一个向内存缓冲区（队列）取值的方法

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

最后是Main方法：

1）长度为5的队列（内存缓冲区）

2）三个生产者，三个消费者。一个线程池，后续会介绍线程池的用法

3）两秒后，生产者不再生成数据，八秒后关闭线程池。

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<Task> queue = new LinkedBlockingQueue<Task>(5);
    		//生产者
    		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(2000);
    			p1.stop();
    			p2.stop();
    			p3.stop();
    			Thread.sleep(8000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}		
    		cachePool.shutdown(); // 在终止前允许执行以前提交的任务
    //		cachePool.shutdownNow(); // 阻止等待任务启动并试图停止当前正在执行的任务
    	}
    	
    }

总结：从打印的结果可以看出，生产者停止生成后不久，消费者也停止消耗。生产者-消费者模型是非常经典常用的设计模型，生产者负责将任务丢到容器中，再由容器提供给消费者（在Disruptor框架中，是消费者主动从容器中获取任务，可以理解为高效版的生产者-消费者模型）。若容器满了，生产者就处于等待状态。若容器空了，消费者就处于等待状态。

以上内容便是学习多线程设计模式记录的笔记，方便自己以后查阅，同时也希望帮助到读者。下一章java线程池

更多干货尽在  [ITDragon博客][ITDragon]

\*注 ： 以上代码均来之网络，但也要自己动手敲一遍，看懂仅仅是不够的。

学习博客：

[并发模型（一）——Future模式][Future]

[并发模型（二）——Master-Worker模式][Master-Worker]

[并发模式（三）——生产者-消费模式][-]

[aHR0cHM6Ly9pbWcyMDIwLmNuYmxvZ3MuY29tL2Jsb2cvODA2OTU2LzIwMjAwNy84MDY5NTYtMjAyMDA3MzEyMzQ1MDc3MjItMTExNjgwOTkyMi5naWY]: /images/20220610/c503d187ad0f436bbf049f36331b19e0.png
[20170808115214693]: /images/20220610/aa019d0dc9184e948c1707ea22e3f2f5.png
[http_www.cnblogs.com_winkey4986_p_6203225.html]: http://www.cnblogs.com/winkey4986/p/6203225.html
[20170808122518860]: /images/20220610/7e62cf700f174846819d50d3dbd2d38c.png
[20170808153359479]: /images/20220610/56231a23c720497298c6940516e07499.png
[ITDragon]: http://www.itit123.cn/
[Future]: http://blog.csdn.net/lmdcszh/article/details/39696357
[Master-Worker]: http://blog.csdn.net/lmdcszh/article/details/39698189
[-]: http://blog.csdn.net/lmdcszh/article/details/39699261