Java高並發之設計模式.

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

 

單例

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

一般分為懶漢式, 餓漢式.

懶漢式: 方法上加synchronized

public static synchronized Singleton getInstance() {  
         if (single == null) {    
             single = new Singleton();  
         }    
        return single;  
}  

這種方式, 由於每次獲取示例都要獲取鎖, 不推薦使用, 性能較差

懶漢式: 使用雙檢鎖 + volatile

    private volatile Singleton singleton = null;
    public static Singleton getInstance() {
        if (singleton == null) {
            synchronized (Singleton.class) {
                if (singleton == null) {
                    singleton = new Singleton();
                }
            }
        }
        return singleton;
    }

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

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

至於為什么要volatile關鍵字, 主要涉及到jdk指令重排, 詳見之前的博文: Java內存模型與指令重排

懶漢式: 使用靜態內部類

public class Singleton {    
    private static class LazyHolder {    
       private static final Singleton INSTANCE = new Singleton();    
    }    
    private Singleton (){}    
    public static final Singleton getInstance() {    
       return LazyHolder.INSTANCE;    
    }    
}   

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

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

餓漢式

public class Singleton1 {  
    private Singleton1() {}  
    private static final Singleton1 single = new Singleton1();  
    public static Singleton1 getInstance() {  
        return single;  
    }  
} 

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

Future模式

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

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

因此可以讓調用者立刻返回一個憑證, 該方法放到另外線程執行,

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

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

通過FutureTask實現

注意其中兩個耗時操作.

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

public class FutureDemo1 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        FutureTask<String> future = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return new RealData().costTime();  8             }
        });
        ExecutorService service = Executors.newCachedThreadPool();
        service.submit(future);

        System.out.println("RealData方法調用完畢");
        // 模擬主函數中其他耗時操作
 doOtherThing(); 16         // 獲取RealData方法的結果
        System.out.println(future.get());
    }

    private static void doOtherThing() throws InterruptedException {
        Thread.sleep(2000L); 22     }
}

class RealData {

    public String costTime() {
        try {
            // 模擬RealData耗時操作
            Thread.sleep(1000L); 31             return "result";
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "exception";
    }

}

通過Future實現

與上述FutureTask不同的是, RealData需要實現Callable接口.

public class FutureDemo2 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newCachedThreadPool();
        Future<String> future = service.submit(new RealData2());

        System.out.println("RealData2方法調用完畢");
        // 模擬主函數中其他耗時操作
        doOtherThing();
        // 獲取RealData2方法的結果
        System.out.println(future.get());
    }

    private static void doOtherThing() throws InterruptedException {
        Thread.sleep(2000L);
    }
}

class RealData2 implements Callable<String>{ 20 
    public String costTime() {
        try {
            // 模擬RealData耗時操作
            Thread.sleep(1000L);
            return "result";
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "exception";
    }

    @Override
    public String call() throws Exception {
        return costTime();
    }
}

另外Future本身還提供了一些額外的簡單控制功能, 其API如下

    // 取消任務
    boolean cancel(boolean mayInterruptIfRunning);
    // 是否已經取消
    boolean isCancelled();
    // 是否已經完成
    boolean isDone();
    // 取得返回對象
    V get() throws InterruptedException, ExecutionException;
    // 取得返回對象, 並可以設置超時時間
    V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;

生產消費者模式

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

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

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

生產者和消費者之間則通過共享內存緩沖區進行通信, 其結構圖如下

PCData為我們需要處理的元數據模型, 生產者構建PCData, 並放入緩沖隊列.

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

生產者核心代碼

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

消費者核心代碼

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

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

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

一般使用BlockingQueue作為數據緩沖隊列, 他是通過鎖和阻塞來實現數據之間的同步, 

如果對緩沖隊列有性能要求, 則可以使用基於CAS無鎖設計的ConcurrentLinkedQueue.

分而治之

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

它可以將一個大任務拆解為若干個小任務並行執行, 提高系統吞吐量.

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

Master-Worker模式

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

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

將結果返回給Master進行歸納與總結.

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

Master代碼

public class MasterDemo {
    // 盛裝任務的集合
    private ConcurrentLinkedQueue<TaskDemo> workQueue = new ConcurrentLinkedQueue<TaskDemo>();
    // 所有worker
    private HashMap<String, Thread> workers = new HashMap<>();
    // 每一個worker並行執行任務的結果
    private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();

    public MasterDemo(WorkerDemo worker, int workerCount) {
        // 每個worker對象都需要持有queue的引用, 用於領任務與提交結果
        worker.setResultMap(resultMap);
        worker.setWorkQueue(workQueue);
        for (int i = 0; i < workerCount; i++) {
            workers.put("子節點: " + i, new Thread(worker));
        }
    }

    // 提交任務
    public void submit(TaskDemo task) {
        workQueue.add(task);
    }

    // 啟動所有的子任務
    public void execute(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            entry.getValue().start();
        }
    }

    // 判斷所有的任務是否執行結束
    public boolean isComplete() {
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            if (entry.getValue().getState() != Thread.State.TERMINATED) {
                return false;
            }
        }

        return true;
    }

    // 獲取最終匯總的結果
    public int getResult() {
        int result = 0;
        for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
            result += Integer.parseInt(entry.getValue().toString());
        }

        return result;
    }

}

Worker代碼

public class WorkerDemo implements Runnable{

    private ConcurrentLinkedQueue<TaskDemo> workQueue;
    private ConcurrentHashMap<String, Object> resultMap;

    @Override
    public void run() {
        while (true) {
            TaskDemo input = this.workQueue.poll();
            // 所有任務已經執行完畢
            if (input == null) {
                break;
            }
            // 模擬對task進行處理, 返回結果
            int result = input.getPrice();
            this.resultMap.put(input.getId() + "", result);
            System.out.println("任務執行完畢, 當前線程: " + Thread.currentThread().getName());
        }
    }

    public ConcurrentLinkedQueue<TaskDemo> getWorkQueue() {
        return workQueue;
    }

    public void setWorkQueue(ConcurrentLinkedQueue<TaskDemo> workQueue) {
        this.workQueue = workQueue;
    }

    public ConcurrentHashMap<String, Object> getResultMap() {
        return resultMap;
    }

    public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
        this.resultMap = resultMap;
    }
}

public class TaskDemo {

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

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getPrice() {
        return price;
    }

    public void setPrice(int price) {
        this.price = price;
    }
}

主函數測試

        MasterDemo master = new MasterDemo(new WorkerDemo(), 10);
        for (int i = 0; i < 100; i++) {
            TaskDemo task = new TaskDemo();
            task.setId(i);
            task.setName("任務" + i);
            task.setPrice(new Random().nextInt(10000));
            master.submit(task);
        }

        master.execute();

        while (true) {
            if (master.isComplete()) {
                System.out.println("執行的結果為: " + master.getResult());
                break;
            }
        }

ForkJoin線程池

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

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

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

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

假設我們的場景需要計算從0到20000000L的累加求和. CountTask繼承自RecursiveTask, 可以攜帶返回值.

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

在子任務中通過THRESHOLD設置子任務分解的閾值, 如果當前需要求和的總數大於THRESHOLD, 則子任務需要再次分解,

如果子任務可以直接執行, 則進行求和操作, 返回結果. 最終等待所有的子任務執行完畢, 對所有結果求和.

public class CountTask extends RecursiveTask<Long>{
    // 任務分解的閾值
    private static final int THRESHOLD = 10000;
    private long start;
    private long end;


    public CountTask(long start, long end) {
        this.start = start;
        this.end = end;
    }

    public Long compute() {
        long sum = 0;
        boolean canCompute = (end - start) < THRESHOLD;
        if (canCompute) {
            for (long i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 分成100個小任務
            long step = (start + end) / 100;
            ArrayList<CountTask> subTasks = new ArrayList<CountTask>();
            long pos = start;
            for (int i = 0; i < 100; i++) {
                long lastOne = pos + step;
                if (lastOne > end) {
                    lastOne = end;
                }
                CountTask subTask = new CountTask(pos, lastOne);
                pos += step + 1;
                // 將子任務推向線程池
                subTasks.add(subTask);
                subTask.fork();
            }

            for (CountTask task : subTasks) {
                // 對結果進行join
                sum += task.join();
            }
        }
        return sum;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        // 累加求和 0 -> 20000000L
        CountTask task = new CountTask(0, 20000000L);
        ForkJoinTask<Long> result = pool.submit(task);
        System.out.println("sum result : " + result.get());
    }
}

ForkJoin線程池使用一個無鎖的棧來管理空閑線程, 如果一個工作線程暫時取不到可用的任務, 則可能被掛起.

掛起的線程將被壓入由線程池維護的棧中, 待將來有任務可用時, 再從棧中喚醒這些線程.