多线程设计模式（五）

并行设计模式属于优化的一部分，它是对一些常用的多线程结构的总结和抽象。与串行相比，并行程序的结构通常更为复杂。因此合理的使用并行模式在多线程开发中更具有意义。

Future模式

Future模式有点类似商品订单。比如网购时，当看重某一件商品时，就可以提交订单，当订单处理完成后，在家里等待商品送货上门即可。或者说更形象的我们的Ajax请求的时候，页面是异步的进行后台处理，用户无须一直等待请求结果，可以继续浏览或操作其他内容。

public class Main {

	public static void main(String[] args) {
		FutureClient fc = new FutureClient();
		Data data = fc.request("请求参数");
		System.out.println("请求发送成功！");
		System.out.println("FutureClient做其他的业务操作");
		
		String result = data.getRequest();
		System.out.println(result);
	}

}

public class FutureClient {

	public Data request(final String queryStr){
		//1.生成一个代理对象（Data接口的实现类）先返回给客户端，告诉客户端请求已经收到，客户端可以继续做其他事
		final FutureData futureData = new FutureData();
		//2.启动一个新的线程，去加载真实的数据，并传给这个带你对象
		new Thread(new Runnable(){
			@Override
			public void run() {
				//3.加载真实的数据，然后传递给代理对象
				RealData realData = new RealData(queryStr);
				futureData.setRealData(realData);
			}}).start();
		return futureData;
	}
}

public interface Data {

	String getRequest();
}

public class FutureData implements Data{

	private RealData realData;
	
	private boolean isReady = false;
	
	public synchronized void setRealData(RealData realData) {
		//如果加载完毕，则直接返回
		if(isReady){
			return;
		}
		//如果没又装载，进行装载真是对象
		this.realData = realData;
		isReady = true;
		//进行通知
		notify();
	}

	@Override
	public String getRequest() {
		//如果没装载好，程序就一直处于阻塞状态
		while(!isReady){
			try{
				wait();
			} catch (InterruptedException e){
				e.printStackTrace();
			}
		}
		//装载好之后直接返回数据即可
		return this.realData.getRequest();
	}

}

public class RealData implements Data{

	private String result;
	
	public RealData(String queryStr) {
		System.out.println("根据" + queryStr + "进行查询，这是一个耗时的操作...");
		try{
			Thread.sleep(5000);
		} catch (InterruptedException e){
			e.printStackTrace();
		}
		System.out.println("操作完毕，获取结果");
		result = "查询结果";
	}

	@Override
	public String getRequest() {
		return result;
	}
}

Master-Worker模式

Master-Worker模式是常用的并行计算模式。它的核心思想是系统有两类进程协作工作：Master进程和Worker进程。Master负责接收和分配任务，Worker负责处理子任务。当各个Worker子进程处理完毕后，会将结果返回给Master，有Master做归纳总结。其好处就是将一个大任务分解成若干和小任务，并行执行，从而提高系统的吞吐量。

public class Main {

	public static void main(String[] args) {
		Master master = new Master(new Worker(), 10);
		Random r = new Random();
		for(int i = 0; i < 100; i++){
			Task task = new Task();
			task.setId(i);
			task.setName("任务"+i);
			task.setPrice(r.nextInt(1000));
			master.submit(task);
		}
		master.execute();
		
		long start = System.currentTimeMillis();
		
		while(true){
			if(master.isComplate()){
				long end = System.currentTimeMillis();
				int ret = master.getResult();
				System.out.println("执行结果 " + ret + ",执行耗时 " + (end-start));
				break;
			}
		}
	}

}

public class Master {

	//1.声明承载任务的容器
	private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<>();
	
	//2.使用HashMap去承载所有的Worker对象
	private HashMap<String,Thread> workers = new HashMap<>();
	
	//3.使用一个容器承载每一个Worker并发执行任务的结果集
	private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();
	
	//4.构造方法
	public Master(Worker worker,int workerCount){
		
		//每一个Worker对象都需要有Master节点的引用，workQueue用于任务的领取，resultMap用于任务的提交
		worker.setWorkQueue(this.workQueue);
		worker.setResultMap(this.resultMap);
		
		for(int i = 0; i < workerCount; i++){
			//key表示每一个worker的名字，value表示线程执行对象
			workers.put("子节点" + Integer.toString(i), new Thread(worker));
		}
	}
	
	//5.提交方法
	public void submit(Task task){
		this.workQueue.add(task);
	}
	
	//6.启动worker的方法（启动应用程序，让所有的Worker工作）
	public void execute(){
		for(Map.Entry<String, Thread> worker : workers.entrySet()){
			worker.getValue().start();
		}
	}
	
	//7.判断线程是否执行完毕
	public boolean isComplate(){
		for(Map.Entry<String, Thread> worker : workers.entrySet()){
			if(worker.getValue().getState() == Thread.State.TERMINATED)
				return true;
		}
		return false;
	}
	
	//8.返回结果集数据
	public int getResult(){
		int result = 0;
		for(Map.Entry<String, Object> res : resultMap.entrySet()){
			//汇总逻辑
			result += (Integer)res.getValue();
		}
		return result;
	}
}

public class Task {

	private int id;
	
	private String name;
	
	private int price;

	//getter&setter...
}

public class Worker implements Runnable{

	//用于接收master节点的ConcurrentLinkedQueue的引用
	private ConcurrentLinkedQueue<Task> workQueue;
	
	//用于接收master节点的ConcurrentHashMap的引用
	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(null == input) break;
			//Task处理任务逻辑
			Object output = handle(input); 
			this.resultMap.put(Integer.toString(input.getId()), output);
		}
	}

	private Object handle(Task input) {
		Object output = null;
		try {
			//表示task处理业务逻辑耗时 
			Thread.sleep(500);
			output = input.getPrice();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return output;
	}
}

生产者-消费者

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