Java高併發之設計模式.

時間 2019-11-29

標籤 java 併發設計模式欄目 Java 简体版

原文原文鏈接

本文主要講解幾種常見並行模式, 具體目錄結構以下圖.html

單例

單例是最多見的一種設計模式, 通常用於全局對象管理, 好比xml配置讀寫之類的.ajax

通常分爲懶漢式, 餓漢式.設計模式

懶漢式: 方法上加synchronized多線程

1 public static synchronized Singleton getInstance() {  
2          if (single == null) {    
3              single = new Singleton();  
4          }    
5         return single;  
6 }

這種方式, 因爲每次獲取示例都要獲取鎖, 不推薦使用, 性能較差框架

懶漢式: 使用雙檢鎖 + volatiledom

 1     private volatile Singleton singleton = null;
 2     public static Singleton getInstance() {
 3         if (singleton == null) {
 4             synchronized (Singleton.class) {
 5                 if (singleton == null) {
 6                     singleton = new Singleton();
 7                 }
 8             }
 9         }
10         return singleton;
11     }

本方式是對直接在方法上加鎖的一個優化, 好處在於只有第一次初始化獲取了鎖.異步

後續調用getInstance已是無鎖狀態. 只是寫法上稍微繁瑣點.ide

至於爲何要volatile關鍵字, 主要涉及到jdk指令重排, 詳見以前的博文: Java內存模型與指令重排函數

懶漢式: 使用靜態內部類post

1 public class Singleton {    
2     private static class LazyHolder {    
3        private static final Singleton INSTANCE = new Singleton();    
4     }    
5     private Singleton (){}    
6     public static final Singleton getInstance() {    
7        return LazyHolder.INSTANCE;    
8     }    
9 }

該方式既解決了同步問題, 也解決了寫法繁瑣問題. 推薦使用改寫法.

缺點在於沒法響應事件來從新初始化INSTANCE.

餓漢式

1 public class Singleton1 {  
2     private Singleton1() {}  
3     private static final Singleton1 single = new Singleton1();  
4     public static Singleton1 getInstance() {  
5         return single;  
6     }  
7 }

缺點在於對象在一開始就直接初始化了.

Future模式

該模式的核心思想是異步調用. 有點相似於異步的ajax請求.

當調用某個方法時, 可能該方法耗時較久, 而在主函數中也不急於馬上獲取結果.

所以可讓調用者馬上返回一個憑證, 該方法放到另外線程執行,

後續主函數拿憑證再去獲取方法的執行結果便可, 其結構圖以下

jdk中內置了Future模式的支持, 其接口以下:

經過FutureTask實現

注意其中兩個耗時操做.

若是doOtherThing耗時2s, 則整個函數耗時2s左右.
若是doOtherThing耗時0.2s, 則整個函數耗時取決於RealData.costTime, 即1s左右結束.

 1 public class FutureDemo1 {
 2 
 3     public static void main(String[] args) throws InterruptedException, ExecutionException {
 4         FutureTask<String> future = new FutureTask<String>(new Callable<String>() {
 5             @Override
 6             public String call() throws Exception {
 7                 return new RealData().costTime();  8             }
 9         });
10         ExecutorService service = Executors.newCachedThreadPool();
11         service.submit(future);
12 
13         System.out.println("RealData方法調用完畢");
14         // 模擬主函數中其餘耗時操做
15  doOtherThing(); 16         // 獲取RealData方法的結果
17         System.out.println(future.get());
18     }
19 
20     private static void doOtherThing() throws InterruptedException {
21         Thread.sleep(2000L); 22     }
23 }
24 
25 class RealData {
26 
27     public String costTime() {
28         try {
29             // 模擬RealData耗時操做
30             Thread.sleep(1000L); 31             return "result";
32         } catch (InterruptedException e) {
33             e.printStackTrace();
34         }
35         return "exception";
36     }
37 
38 }

經過Future實現

與上述FutureTask不一樣的是, RealData須要實現Callable接口.

 1 public class FutureDemo2 {
 2 
 3     public static void main(String[] args) throws InterruptedException, ExecutionException {
 4         ExecutorService service = Executors.newCachedThreadPool();
 5         Future<String> future = service.submit(new RealData2());
 6 
 7         System.out.println("RealData2方法調用完畢");
 8         // 模擬主函數中其餘耗時操做
 9         doOtherThing();
10         // 獲取RealData2方法的結果
11         System.out.println(future.get());
12     }
13 
14     private static void doOtherThing() throws InterruptedException {
15         Thread.sleep(2000L);
16     }
17 }
18 
19 class RealData2 implements Callable<String>{ 20 
21     public String costTime() {
22         try {
23             // 模擬RealData耗時操做
24             Thread.sleep(1000L);
25             return "result";
26         } catch (InterruptedException e) {
27             e.printStackTrace();
28         }
29         return "exception";
30     }
31 
32     @Override
33     public String call() throws Exception {
34         return costTime();
35     }
36 }

另外Future自己還提供了一些額外的簡單控制功能, 其API以下

 1     // 取消任務
 2     boolean cancel(boolean mayInterruptIfRunning);
 3     // 是否已經取消
 4     boolean isCancelled();
 5     // 是否已經完成
 6     boolean isDone();
 7     // 取得返回對象
 8     V get() throws InterruptedException, ExecutionException;
 9     // 取得返回對象, 並能夠設置超時時間
10     V get(long timeout, TimeUnit unit)
11             throws InterruptedException, ExecutionException, TimeoutException;

生產消費者模式

生產者-消費者模式是一個經典的多線程設計模式. 它爲多線程間的協做提供了良好的解決方案。

在生產者-消費者模式中，一般由兩類線程，即若干個生產者線程和若干個消費者線程。

生產者線程負責提交用戶請求，消費者線程則負責具體處理生產者提交的任務。

生產者和消費者之間則經過共享內存緩衝區進行通訊, 其結構圖以下

PCData爲咱們須要處理的元數據模型, 生產者構建PCData, 並放入緩衝隊列.

消費者從緩衝隊列中獲取數據, 並執行計算.

生產者核心代碼

 1         while(isRunning) {
 2             Thread.sleep(r.nextInt(SLEEP_TIME));
 3             data = new PCData(count.incrementAndGet);
 4             // 構造任務數據
 5             System.out.println(data + " is put into queue");
 6             if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
 7                 // 將數據放入隊列緩衝區中
 8                 System.out.println("faild to put data : " + data);
 9             }
10         }

消費者核心代碼

 1         while (true) {
 2             PCData data = queue.take();
 3             // 提取任務
 4             if (data != null) {
 5                 // 獲取數據, 執行計算操做
 6                 int re = data.getData() * 10;
 7                 System.out.println("after cal, value is : " + re);
 8                 Thread.sleep(r.nextInt(SLEEP_TIME));
 9             }
10         }

生產消費者模式能夠有效對數據解耦, 優化系統結構.

下降生產者和消費者線程相互之間的依賴與性能要求.

通常使用BlockingQueue做爲數據緩衝隊列, 他是經過鎖和阻塞來實現數據之間的同步,

若是對緩衝隊列有性能要求, 則可使用基於CAS無鎖設計的ConcurrentLinkedQueue.

分而治之

嚴格來說, 分而治之不算一種模式, 而是一種思想.

它能夠將一個大任務拆解爲若干個小任務並行執行, 提升系統吞吐量.

咱們主要講兩個場景, Master-Worker模式, ForkJoin線程池.

Master-Worker模式

該模式核心思想是系統由兩類進行協助工做: Master進程, Worker進程.

Master負責接收與分配任務, Worker負責處理任務. 當各個Worker處理完成後,

將結果返回給Master進行概括與總結.

假設一個場景, 須要計算100個任務, 並對結果求和, Master持有10個子進程.

Master代碼

 1 public class MasterDemo {
 2     // 盛裝任務的集合
 3     private ConcurrentLinkedQueue<TaskDemo> workQueue = new ConcurrentLinkedQueue<TaskDemo>();
 4     // 全部worker
 5     private HashMap<String, Thread> workers = new HashMap<>();
 6     // 每個worker並行執行任務的結果
 7     private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();
 8 
 9     public MasterDemo(WorkerDemo worker, int workerCount) {
10         // 每一個worker對象都須要持有queue的引用, 用於領任務與提交結果
11         worker.setResultMap(resultMap);
12         worker.setWorkQueue(workQueue);
13         for (int i = 0; i < workerCount; i++) {
14             workers.put("子節點: " + i, new Thread(worker));
15         }
16     }
17 
18     // 提交任務
19     public void submit(TaskDemo task) {
20         workQueue.add(task);
21     }
22 
23     // 啓動全部的子任務
24     public void execute(){
25         for (Map.Entry<String, Thread> entry : workers.entrySet()) {
26             entry.getValue().start();
27         }
28     }
29 
30     // 判斷全部的任務是否執行結束
31     public boolean isComplete() {
32         for (Map.Entry<String, Thread> entry : workers.entrySet()) {
33             if (entry.getValue().getState() != Thread.State.TERMINATED) {
34                 return false;
35             }
36         }
37 
38         return true;
39     }
40 
41     // 獲取最終彙總的結果
42     public int getResult() {
43         int result = 0;
44         for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
45             result += Integer.parseInt(entry.getValue().toString());
46         }
47 
48         return result;
49     }
50 
51 }

Worker代碼

 1 public class WorkerDemo implements Runnable{
 2 
 3     private ConcurrentLinkedQueue<TaskDemo> workQueue;
 4     private ConcurrentHashMap<String, Object> resultMap;
 5 
 6     @Override
 7     public void run() {
 8         while (true) {
 9             TaskDemo input = this.workQueue.poll();
10             // 全部任務已經執行完畢
11             if (input == null) {
12                 break;
13             }
14             // 模擬對task進行處理, 返回結果
15             int result = input.getPrice();
16             this.resultMap.put(input.getId() + "", result);
17             System.out.println("任務執行完畢, 當前線程: " + Thread.currentThread().getName());
18         }
19     }
20 
21     public ConcurrentLinkedQueue<TaskDemo> getWorkQueue() {
22         return workQueue;
23     }
24 
25     public void setWorkQueue(ConcurrentLinkedQueue<TaskDemo> workQueue) {
26         this.workQueue = workQueue;
27     }
28 
29     public ConcurrentHashMap<String, Object> getResultMap() {
30         return resultMap;
31     }
32 
33     public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
34         this.resultMap = resultMap;
35     }
36 }

 1 public class TaskDemo {
 2 
 3     private int id;
 4     private String name;
 5     private int price;
 6 
 7     public int getId() {
 8         return id;
 9     }
10 
11     public void setId(int id) {
12         this.id = id;
13     }
14 
15     public String getName() {
16         return name;
17     }
18 
19     public void setName(String name) {
20         this.name = name;
21     }
22 
23     public int getPrice() {
24         return price;
25     }
26 
27     public void setPrice(int price) {
28         this.price = price;
29     }
30 }

View Code

主函數測試

 1         MasterDemo master = new MasterDemo(new WorkerDemo(), 10);
 2         for (int i = 0; i < 100; i++) {
 3             TaskDemo task = new TaskDemo();
 4             task.setId(i);
 5             task.setName("任務" + i);
 6             task.setPrice(new Random().nextInt(10000));
 7             master.submit(task);
 8         }
 9 
10         master.execute();
11 
12         while (true) {
13             if (master.isComplete()) {
14                 System.out.println("執行的結果爲: " + master.getResult());
15                 break;
16             }
17         }

ForkJoin線程池

該線程池是jdk7以後引入的一個並行執行任務的框架, 其核心思想也是將任務分割爲子任務,

有可能子任務仍是很大, 還須要進一步拆解, 最終獲得足夠小的任務.

將分割出來的子任務放入雙端隊列中, 而後幾個啓動線程從雙端隊列中獲取任務執行.

子任務執行的結果放到一個隊列裏, 另起線程從隊列中獲取數據, 合併結果.

假設咱們的場景須要計算從0到20000000L的累加求和. CountTask繼承自RecursiveTask, 能夠攜帶返回值.

每次分解大任務, 簡單的將任務劃分爲100個等規模的小任務, 並使用fork()提交子任務.

在子任務中經過THRESHOLD設置子任務分解的閾值, 若是當前須要求和的總數大於THRESHOLD, 則子任務須要再次分解,

若是子任務能夠直接執行, 則進行求和操做, 返回結果. 最終等待全部的子任務執行完畢, 對全部結果求和.

 1 public class CountTask extends RecursiveTask<Long>{
 2     // 任務分解的閾值
 3     private static final int THRESHOLD = 10000;
 4     private long start;
 5     private long end;
 6 
 7 
 8     public CountTask(long start, long end) {
 9         this.start = start;
10         this.end = end;
11     }
12 
13     public Long compute() {
14         long sum = 0;
15         boolean canCompute = (end - start) < THRESHOLD;
16         if (canCompute) {
17             for (long i = start; i <= end; i++) {
18                 sum += i;
19             }
20         } else {
21             // 分紅100個小任務
22             long step = (start + end) / 100;
23             ArrayList<CountTask> subTasks = new ArrayList<CountTask>();
24             long pos = start;
25             for (int i = 0; i < 100; i++) {
26                 long lastOne = pos + step;
27                 if (lastOne > end) {
28                     lastOne = end;
29                 }
30                 CountTask subTask = new CountTask(pos, lastOne);
31                 pos += step + 1;
32                 // 將子任務推向線程池
33                 subTasks.add(subTask);
34                 subTask.fork();
35             }
36 
37             for (CountTask task : subTasks) {
38                 // 對結果進行join
39                 sum += task.join();
40             }
41         }
42         return sum;
43     }
44 
45     public static void main(String[] args) throws ExecutionException, InterruptedException {
46         ForkJoinPool pool = new ForkJoinPool();
47         // 累加求和 0 -> 20000000L
48         CountTask task = new CountTask(0, 20000000L);
49         ForkJoinTask<Long> result = pool.submit(task);
50         System.out.println("sum result : " + result.get());
51     }
52 }