一、Future模式

1）用Java实现Future模式

我们先用一个例子来对future模式进行形象的分析：
情景1：
假如你突然想做饭，但是没有厨具，也没有食材。
在往常，我们可能会先去菜场买菜，然后去超市购买厨具，最后回家烹饪，这是一个迭代的流程。
在厨具买回来前，你没办法进行烹饪，这显然相当浪费时间。面对这种尴尬的处境，程序员就会想：在主线程买菜的期间，能不能发起一个子线程去移步买菜？
Where there is a will，there is a way，future 模式应运而生！
情景2：
我们去超市买食材，在此期间我们当然不会闲着，用手机发起了一次网购，相当于在主线程里面另起一个子线程去网购厨具，或者说发起一个Ajax请求。我们无需一直在家里等待快递配送，而是可以继续买菜，在我们回家的时候，快递送来了厨具。

执行步骤：

1.创建一个Data接口

代理数据和真实数据都实现了这个接口并重写getRequest方法用来返回数据

  1. public interface Data {
  2.     String getRequest();
  3. }

image.gif

2.代理数据FutureData

实现接口并封装了RealData ，如果没有装载好getRequest就处于阻塞状态，否则就正常执行

  1. public class FutureData implements Data{
  2.     private RealData realData ;
  3.     private boolean isReady = false;
  4.     public synchronized void setRealData(RealData realData) {
  5.         //如果已经装载完毕了，就直接返回
  6.         if(isReady){
  7.             return;
  8.         }
  9.         //如果没装载，进行装载真实对象
  10.         this.realData = realData;
  11.         isReady = true;
  12.         //进行通知
  13.         notify();
  14.     }
  15.     @Override
  16.     public synchronized String getRequest() {
  17.         //如果没装载好 程序就一直处于阻塞状态
  18.         while(!isReady){
  19.             try {
  20.                 wait();
  21.             } catch (InterruptedException e) {
  22.                 e.printStackTrace();
  23.             }
  24.         }
  25.         //装载好直接获取数据即可
  26.         return this.realData.getRequest();
  27.     }
  28. }

image.gif

3.真实数据RealData

实现接口，实例化时购买厨具，模拟耗时5秒

  1. public class RealData implements Data{
  2.     private String result ;
  3.     public RealData (String queryStr){
  4.         System.out.println("根据" + queryStr + "购买厨具，这是一个很耗时的操作..");
  5.         try {
  6.             Thread.sleep(5000);
  7.         } catch (InterruptedException e) {
  8.             e.printStackTrace();
  9.         }
  10.         System.out.println("操作完毕，获取结果");
  11.         result = "《查询的结果》";
  12.     }
  13.     @Override
  14.     public String getRequest() {
  15.         return result;
  16.     }
  17. }

image.gif

4.客户端Client向Service发起请求，网购餐具

  1. public class Main {
  2.     public static void main(String[] args) throws InterruptedException {        
  3.         FutureService fc = new FutureService();
  4.         Data data = fc.request("网购厨具");
  5.         System.out.println("下单成功!");
  6.         System.out.println("去菜场买菜...");        
  7.         String result = data.getRequest();
  8.         System.out.println(result);
  9.     }
  10. }

image.gif

5.Server向数据源Data慢慢请求真实数据，并返回给客户端

  1. public class FutureService {
  2.     public Data request(final String queryStr){
  3.         //1 我想要一个代理对象（Data接口的实现类）先返回给发送请求的客户端，让他可以做其他的事情
  4.         final FutureData futureData = new FutureData();
  5.         //2 启动一个新的线程，去加载真实的数据，传递给这个代理对象
  6.         new Thread(new Runnable() {
  7.             @Override
  8.             public void run() {
  9.                 //3 这个新的线程可以去慢慢的加载真实对象，然后传递给代理对象
  10.                 RealData realData = new RealData(queryStr);
  11.                 futureData.setRealData(realData);
  12.             }
  13.         }).start();        
  14.         return futureData;
  15.     }    
  16. }

image.gif
以下是执行结果：
JAVA的并发编程（六）： 多线程的设计模式 - 图6image.gif

2）JDK自带的Callable、Future和FutureTask

1.Callable与Runnable

先说一下java.lang.Runnable吧，它是一个接口，在它里面只声明了一个run()方法：

  1. public interface Runnable {
  2.     public abstract void run();
  3. }

image.gif
由于run()方法返回值为void类型，所以在执行完任务之后无法返回任何结果。
Callable位于java.util.concurrent包下，它也是一个接口，在它里面也只声明了一个方法，只不过这个方法叫做call()：

  1. public interface Callable<V> {    
  2.     V call() throws Exception;
  3. }

image.gif
可以看到，这是一个泛型接口，call()函数返回的类型就是传递进来的V类型。
　　那么怎么使用Callable呢？一般情况下是配合ExecutorService线程池来使用的，在ExecutorService接口中声明了若干个submit方法的重载版本：

  1. <T> Future<T> submit(Callable<T> task);
  2. <T> Future<T> submit(Runnable task, T result);
  3. Future<?> submit(Runnable task);

image.gif
　　第一个submit方法里面的参数类型就是Callable。
　　暂时只需要知道Callable一般是和ExecutorService配合来使用的，具体的使用方法讲在后面讲述。
　　一般情况下我们使用第一个submit方法和第三个submit方法，第二个submit方法很少使用。
submit方法和excute方法的区别：submit可以传入Callable接口，而且他有返回值

2.Future

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。
　　Future类位于java.util.concurrent包下，它是一个接口：

  1. public interface Future<V> {
  2.     boolean cancel(boolean mayInterruptIfRunning);
  3.     boolean isCancelled();
  4.     boolean isDone();
  5.     V get() throws InterruptedException, ExecutionException;
  6.     V get(long timeout, TimeUnit unit)
  7.         throws InterruptedException, ExecutionException, TimeoutException;
  8. }

image.gif
cancel方法用来取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，即如果取消已经完成的任务会返回false；如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若mayInterruptIfRunning设置为false，则返回false；如果任务还没有执行，则无论mayInterruptIfRunning为true还是false，肯定返回true。 　　在Future接口中声明了5个方法，下面依次解释每个方法的作用：

  • isCancelled方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
  • isDone方法表示任务是否已经完成，若任务完成，则返回true；
  • get()方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回；
  • get(long timeout, TimeUnit unit)用来获取执行结果，如果在指定时间内，还没获取到结果，就直接返回null。

　　也就是说Future提供了三种功能：
　　1）判断任务是否完成；
　　2）能够中断任务；
　　3）能够获取任务执行结果。
　　因为Future只是一个接口，所以是无法直接用来创建对象使用的，因此就有了下面的FutureTask。

3.FutureTask

我们先来看一下FutureTask的实现：

  1. public class FutureTask<V> implements RunnableFuture<V>

image.gif
　　FutureTask类实现了RunnableFuture接口，我们看一下RunnableFuture接口的实现：

  1. public interface RunnableFuture<V> extends Runnable, Future<V> {
  2.     void run();
  3. }

image.gif
　　可以看出RunnableFuture继承了Runnable接口和Future接口，而FutureTask实现了RunnableFuture接口。所以它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。
　　FutureTask提供了2个构造器：

  1. public FutureTask(Callable<V> callable) {}
  2. public FutureTask(Runnable runnable, V result) {}

image.gif
　　事实上，FutureTask是Future接口的一个唯一实现类。

4.使用示例

使用Callable+Future获取执行结果

  • 新建Task实现Callable接口

    1. class Task implements Callable<Integer> {
    2.   @Override
    3.   public Integer call() throws Exception {
    4.       System.out.println("子线程在进行计算");
    5.       Thread.sleep(3000);
    6.       int sum = 0;
    7.       for(int i=0;i<100;i++)
    8.           sum += i;
    9.       return sum;
    10.   }
    11. }

    image.gif

  • 调用submit方法生成Futuer对象，调用future.get() 阻塞直到获取真实数据；

    1. public class Test {
    2.   public static void main(String[] args) {
    3.       ExecutorService executor = Executors.newCachedThreadPool();
    4.       Task task = new Task();
    5.       Future<Integer> future = executor.submit(task);
    6.       executor.shutdown();
    8.       try {
    9.           Thread.sleep(1000);
    10.       } catch (InterruptedException e1) {
    11.           e1.printStackTrace();
    12.       }         
    13.       System.out.println("主线程在执行任务");
    15.       try {
    16.           System.out.println("task运行结果"+future.get());
    17.       } catch (InterruptedException e) {
    18.           e.printStackTrace();
    19.       } catch (ExecutionException e) {
    20.           e.printStackTrace();
    21.       }
    22.       System.out.println("所有任务执行完毕");
    23.   }
    24. }

    image.gif
    　　执行结果：
    JAVA的并发编程（六）： 多线程的设计模式 - 图17image.gif
    　　2.使用Callable+FutureTask获取执行结果

    1. public class Test2 {
    2.   public static void main(String[] args) {
    3.       //第一种方式
    4.       ExecutorService executor = Executors.newCachedThreadPool();
    5.       Task2 task = new Task2();
    6.       FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
    7.       executor.submit(futureTask);
    8.       executor.shutdown();
    9.       //第二种方式，注意这种方式和第一种方式效果是类似的，只不过一个使用的是ExecutorService，一个使用的是Thread
    10.       /*Task task = new Task();
    11.       FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
    12.       Thread thread = new Thread(futureTask);
    13.       thread.start();*/
    14.       try {
    15.           Thread.sleep(1000);
    16.       } catch (InterruptedException e1) {
    17.           e1.printStackTrace();
    18.       }
    19.       System.out.println("主线程在执行任务");
    21.       try {
    22.           System.out.println("task运行结果"+futureTask.get());
    23.       } catch (InterruptedException e) {
    24.           e.printStackTrace();
    25.       } catch (ExecutionException e) {
    26.           e.printStackTrace();
    27.       }
    28.       System.out.println("所有任务执行完毕");
    29.   }
    30. }
    31. class Task2 implements Callable<Integer> {
    32.   @Override
    33.   public Integer call() throws Exception {
    34.       System.out.println("子线程在进行计算");
    35.       Thread.sleep(3000);
    36.       int sum = 0;
    37.       for(int i=0;i<100;i++)
    38.           sum += i;
    39.       return sum;
    40.   }
    41. }

    image.gif
    　　如果为了可取消性而使用 Future 但又不提供可用的结果，则可以声明 Future<?> 形式类型、并返回 null 作为底层任务的结果。

    二、Master-worker模式

    1）简介

    Master-Worker模式是常用的并行设计模式。它的核心思想是，系统有两个进程协议工作：Master进程和Worker进程。Master进程负责接收和分配任务，Worker进程负责处理子任务。当各个Worker进程将子任务处理完后，将结果返回给Master进程，由Master进行归纳和汇总，从而得到系统结果。处理过程如下图：
    JAVA的并发编程（六）： 多线程的设计模式 - 图20image.gif
    Master-Worker模式的好处是，它能将大任务分解成若干个小任务，并发执行，从而提高系统性能。而对于系统请求者Client来说，任务一旦提交，Master进程就会立刻分配任务并立即返回，并不会等系统处理完全部任务再返回，其处理过程是异步的。
    Master-Worker模式的主要结构如下图：
    JAVA的并发编程（六）： 多线程的设计模式 - 图22image.gif

　如上图所示
Master：主要进程，用于任务的分配和最终结果的合并，它维护着一个Worker进程队列、子任务队列和子结果集，Worker进程中的Worker进程不断的从任务队列中提取要处理的子任务，并将子任务的处理结果放入到子结果集中。这里使用上一个笔记JAVA的并发编程（五）： 同步类容器和并发类容器中提到的并发类容器，ConcurrentLinkedQueue用于存放任务，HashMap存放Worker，ConcurrentHashMap存放结果集。
Worker：用于实际处理一个任务；客户端进程：用于启动系统，调度开启Master。

2）代码实现

1.Main主程序（依据处理器数量限制开启的线程数）

  1. public class Main {
  2.     public static void main(String[] args) {
  3.         System.out.println("我的机器可用Processor数量:" + Runtime.getRuntime().availableProcessors());
  4.         Master master = new Master(new MyWorker(), Runtime.getRuntime().availableProcessors()*25);
  5.         Random r = new Random();
  6.         //插入任务
  7.         for(int i = 1; i<= 100; i++){
  8.             Task t = new Task();
  9.             t.setId(i);
  10.             t.setName("任务"+i);
  11.             t.setPrice(r.nextInt(1000));
  12.             master.submit(t);
  13.         }
  14.         //执行任务
  15.         master.execute();
  16.         long start = System.currentTimeMillis();
  17.         while(true){
  18.             if(master.isComplete()){
  19.                 long end = System.currentTimeMillis() - start;
  20.                 int ret = master.getResult();
  21.                 System.out.println("最终结果：" + ret + "， 执行耗时：" + end);
  22.                 break;
  23.             }
  24.         }
  25.     }
  26. }

image.gif

2.Task任务

  1. public class Task {
  2.     private int id ;
  3.     private String name;
  4.     private int price;
  5.     public int getId() { return id; }
  6.     public void setId(int id) { this.id = id; }
  7.     public String getName() { return name; }
  8.     public void setName(String name) { this.name = name; }
  9.     public int getPrice() { return price; }
  10.     public void setPrice(int price) { this.price = price; }
  11. }

image.gif

3.Worker工作类（基类）

  1. public class Worker implements Runnable {
  2.     private ConcurrentLinkedQueue<Task> workQueue;
  3.     private ConcurrentHashMap<String, Object> resultMap;
  4.     public void setWorkerQueue(ConcurrentLinkedQueue<Task> workQueue) {
  5.         this.workQueue = workQueue;
  6.     }
  7.     public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
  8.         this.resultMap = resultMap;
  9.     }
  10.     @Override
  11.     public void run() {
  12.         while(true){
  13.             Task input = this.workQueue.poll();
  14.             if(input == null) break;
  15.             //真正的去做业务处理
  16.             Object output = handle(input);
  17.             this.resultMap.put(Integer.toString(input.getId()), output);
  18.         }
  19.     }
  20.     //可以通过重新handle方法扩展
  21.     public Object handle(Task input){
  22.         return null;
  23.     }
  24. }

image.gif

4.Worker工作类（扩展类）

  1. public class MyWorker extends Worker {
  2.     @Override
  3.     public  Object handle(Task input) {
  4.         Object output = null;
  5.         try {
  6.             //表示处理task任务的耗时，可能是数据的加工，也可能是操作数据库...
  7.             Thread.sleep(500);
  8.             output = input.getPrice();
  9.         } catch (InterruptedException e) {
  10.             e.printStackTrace();
  11.         }
  12.         return output;
  13.     }
  14. }

image.gif

5.Master控制器

  1. public class Master {
  2.     //1 应该有一个承装任务的集合
  3.     private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
  4.     //2 使用HashMap去承装所有的worker对象
  5.     private HashMap<String, Thread> workers = new HashMap<String, Thread>();
  6.     //3 使用一个容器承装每一个worker并非执行任务的结果集
  7.     private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
  8.     //4 构造方法
  9.     public Master(Worker worker, int workerCount){
  10.         // 每一个worker对象都需要有Master的引用 workQueue用于任务的领取，resultMap用于任务的提交
  11.         worker.setWorkerQueue(this.workQueue);
  12.         worker.setResultMap(this.resultMap);
  13.         for(int i = 0 ; i < workerCount; i++){
  14.             //key表示每一个worker的名字, value表示线程执行对象
  15.             workers.put("子节点" + Integer.toString(i), new Thread(worker));
  16.         }
  17.     }
  18.     //5 提交方法
  19.     public void submit(Task task){
  20.         this.workQueue.add(task);
  21.     }
  22.     //6 需要有一个执行的方法（启动应用程序 让所有的worker工作）
  23.     public void execute(){
  24.         for(Map.Entry<String, Thread> me : workers.entrySet()){
  25.             me.getValue().start();
  26.         }
  27.     }
  28.     //8 判断线程是否执行完毕
  29.     public boolean isComplete() {
  30.         for(Map.Entry<String, Thread> me : workers.entrySet()){
  31.             if(me.getValue().getState() != Thread.State.TERMINATED){
  32.                 return false;
  33.             }
  34.         }        
  35.         return true;
  36.     }
  37.     //9 返回结果集数据
  38.     public int getResult() {
  39.         int ret = 0;
  40.         for(Map.Entry<String, Object> me : resultMap.entrySet()){
  41.             //汇总的逻辑..
  42.             ret += (Integer)me.getValue();
  43.         }
  44.         return ret;
  45.     }
  46. }

image.gif
运行结果为：
JAVA的并发编程（六）： 多线程的设计模式 - 图29image.gif

三、生产者-消费者模式

1）概述

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式，都会找一个第三者出来进行解耦，如工厂模式的第三者是工厂类，模板模式的第三者是模板类。在学习一些设计模式的过程中，如果先找到这个模式的第三者，能帮助我们快速熟悉一个设计模式。
JAVA的并发编程（六）： 多线程的设计模式 - 图31image.gif

2）实现

1.Data数据

  1. public final class Data {
  2.     private String id;
  3.     private String name;
  4.     public Data(String id, String name){
  5.         this.id = id;
  6.         this.name = name;
  7.     }
  8.     public String getId() { return id; }
  9.     public void setId(String id) { this.id = id; }
  10.     public String getName() { return name; }
  11.     public void setName(String name) { this.name = name; }
  12.     @Override
  13.     public String toString(){
  14.         return "{id: " + id + ", name: " + name + "}";
  15.     }
  16. }

image.gif

2.Provider生产者

  1. public class Provider implements Runnable{
  2.     //共享缓存区
  3.     private BlockingQueue<Data> queue;
  4.     //多线程间是否启动变量，有强制从主内存中刷新的功能。即时返回线程的状态
  5.     private volatile boolean isRunning = true;
  6.     //id生成器
  7.     private static AtomicInteger count = new AtomicInteger();
  8.     //随机对象
  9.     private static Random r = new Random();
  10.     public Provider(BlockingQueue queue){
  11.         this.queue = queue;
  12.     }
  13.     @Override
  14.     public void run() {
  15.         while(isRunning){
  16.             try {
  17.                 //随机休眠0 - 1000 毫秒 表示获取数据(产生数据的耗时) 
  18.                 Thread.sleep(r.nextInt(1000));
  19.                 //获取的数据进行累计...
  20.                 int id = count.incrementAndGet();
  21.                 //比如通过一个getData方法获取了
  22.                 Data data = new Data(Integer.toString(id), "数据" + id);
  23.                 System.out.println("当前线程:" + Thread.currentThread().getName() + ", 获取了数据，id为:" + id + ", 进行装载到公共缓冲区中...");
  24.                 if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
  25.                     System.out.println("提交缓冲区数据失败....");
  26.                     //do something... 比如重新提交
  27.                 }
  28.             } catch (InterruptedException e) {
  29.                 e.printStackTrace();
  30.             }
  31.         }
  32.     }
  33.     public void stop(){
  34.         this.isRunning = false;
  35.     }
  36. }

image.gif

3.Cusumer消费者

  1. public class Consumer implements Runnable{
  2.     private BlockingQueue<Data> queue;
  3.     public Consumer(BlockingQueue queue){
  4.         this.queue = queue;
  5.     }
  6.     //随机对象
  7.     private static Random r = new Random();
  8.     @Override
  9.     public void run() {
  10.         while(true){
  11.             try {
  12.                 //获取数据
  13.                 Data data = this.queue.take();
  14.                 //进行数据处理。休眠0 - 1000毫秒模拟耗时
  15.                 Thread.sleep(r.nextInt(1000));
  16.                 System.out.println("当前消费线程：" + Thread.currentThread().getName() + "， 消费成功，消费数据为id: " + data.getId());
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.         }
  21.     }
  22. }

image.gif

4.Main主程序

  1. public class Main {
  2.     public static void main(String[] args) throws Exception {
  3.         //内存缓冲区
  4.         BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(10);
  5.         //生产者
  6.         Provider p1 = new Provider(queue);
  7.         Provider p2 = new Provider(queue);
  8.         Provider p3 = new Provider(queue);
  9.         //消费者
  10.         Consumer c1 = new Consumer(queue);
  11.         Consumer c2 = new Consumer(queue);
  12.         Consumer c3 = new Consumer(queue);
  14.         //创建线程池运行,这是一个缓存的线程池，可以创建无穷大的线程，没有任务的时候不创建线程。空闲线程存活时间为60s（默认值）
  15.         ExecutorService cachePool = Executors.newCachedThreadPool();
  16.         cachePool.execute(p1);
  17.         cachePool.execute(p2);
  18.         cachePool.execute(p3);
  19.         cachePool.execute(c1);
  20.         cachePool.execute(c2);
  21.         cachePool.execute(c3);
  22.         try {
  23.             Thread.sleep(3000);
  24.         } catch (InterruptedException e) {
  25.             e.printStackTrace();
  26.         }
  27.         p1.stop();
  28.         p2.stop();
  29.         p3.stop();
  30.         try {
  31.             Thread.sleep(2000);
  32.         } catch (InterruptedException e) {
  33.             e.printStackTrace();
  34.         }        
  35.         //cachePool.shutdown();
  36.         //cachePool.shutdownNow();
  37.     }
  38. }

image.gif
执行结果为：
JAVA的并发编程（六）： 多线程的设计模式 - 图37image.gif
这是最简单的不包含线程通信的版本，其实可以加入线程通信，让生产者和消费者之间可以互相唤醒，具体详见之前的笔记：JAVA的并发编程（四）： 线程的通信