併發編程高級篇一----JDK多任務執行框架,Concurrent.util工具類以及鎖的高級深化

1.Jdk的多任務執行框架

JDK提供了一套線程框架Executor來幫助開發者有效的進行線程控制，Executors扮演線程工廠的角色，其建立線程的方法以下

1. newFixedThreadPool() 返回固定數量的線程池，該方法的線程數始終不變。若線程空閒則當即執行 不然暫緩到隊列中
2. newSingleThreadPool() 建立一個線程池，若線程空閒則當即執行 不然暫緩到隊列中
3. newCachedThreadPool() 返回一個可根據實際狀況調整線程個數的線程池 ，若線程空閒則當即執行 不然暫緩到隊列中
4. newScheduledThreadPool() 返回一個ScheduledExecutorService對象，但該線程能夠執行線程的數量(schedule 排定/安排/目錄)

ScheduledExecutorService scheduler = 
        Executors.newScheduledThreadPool(10);
        
        //command 就是一個Thread
        ScheduledFuture<?> scheduledTask =
         scheduler.scheduleWithFixedDelay(command,5,1, TimeUnit.SECONDS);

若Executors工廠類沒法知足咱們的需求，能夠本身去建立自定義的線程池。自定義線程池的構造方法以下

public ThreadPoolExecutor(int corePoolSize,//核心線程數
                          int maximumPoolSize,//最大線程數
                          long keepAliveTime,//線程的空閒時間
                          TimeUnit unit,//給定單元粒度的時間段 
                          BlockingQueue<Runnable> workQueue,//有界、無界隊列
                          RejectedExecutoionHandler handler//任務拒絕策略
                          ){.....}

使用什麼隊列對該構造方法來講比較關鍵

1. 使用有界隊列 若有任務須要執行 若是實際線程數<corePoolSize，則建立線程

若是實際線程數>corePoolSize，則將任務添加到緩存隊列
    若是緩存隊列已滿 ，總線程<maximumPoolSize, 則建立線程
                     總線程>maximumPoolSize, 則執行拒絕策略

1. 使用無界隊列LinkedBlockingQueue 除非系統資源耗盡 不然不會出現入隊失敗

若有任務須要執行 若是實際線程數<corePoolSize，則建立線程
                   若是實際線程數>corePoolSize，則將任務添加到緩存隊列，
                                               直到資源耗盡

BlockingQueue<Runnable> queue =
                    //new LinkedBlockingQueue<Runnable>();//無界隊列
                    new ArrayBlockingQueue<Runnable>(10);//有界隊列
            ExecutorService executor  = new ThreadPoolExecutor(
                    5,         //core
                    10,     //max
                    120L,     //120s
                    TimeUnit.SECONDS,
                    queue);

JDK的拒絕策略

1. AbortPolicy:直接拋出異常 系統正常工做
2. CallerRunsPolicy：只要線程池未被關閉 嘗試運行被丟棄的任務
3. DiscardOldestPolicy:丟失最老的請求 嘗試提交當前任務
4. DiscardPolicy：丟棄沒法處理的任務不給予處理

JDK提供的拒絕策略不友好，能夠自定義拒絕策略，實現RejectedExecutionHandler接口（添加日誌等等）

public class MyRejected implements RejectedExecutionHandler {
    
        public MyRejected(){}

        @Override
        public void rejectedExecution(Runnable r, 
        ThreadPoolExecutor executor) {
            System.out.println("自定義處理");
            System.out.println("當前被拒絕的任務爲"+r.toString());
        }
    }

---

2.Concurrent.util工具類詳解

CyclicBarrier
假設每個線程表明一個運動員，當運動員都準備好了，才能一塊兒出發。

CyclicBarrier barrier = new CyclicBarrier(3);

CountDownLatch
常常用於監聽某些初始化操做，當初始化執行完畢之後，通知主線程繼續工做

final CountDownLatch countDownLatch = new CountDownLatch(2);

Callable 和Future使用
Futrue模式費用適合在處理耗時很長的業務邏輯進行使用，能夠有效的減少系統的影響，
提升系統的吞吐量

public class UseFuture implements Callable<String>{
        private String para;
        
        public UseFuture(String para){
            this.para = para;
        }
        
        /**
         * 這裏是真實的業務邏輯，其執行可能很慢
         */
        @Override
        public String call() throws Exception {
            //模擬執行耗時
            Thread.sleep(5000);
            String result = this.para + "處理完成";
            return result;
        }
        
        //主控制函數
        public static void main(String[] args) throws Exception {
            String queryStr = "query";
            //構造FutureTask，而且傳入須要真正進行業務邏輯處理的類,
            //該類必定是實現了Callable接口的類
            FutureTask<String> future = 
            new FutureTask<String>(new UseFuture(queryStr));
            
            FutureTask<String> future2 = 
            new FutureTask<String>(new UseFuture(queryStr));
            //建立一個固定線程的線程池且線程數爲1,
            ExecutorService executor = Executors.newFixedThreadPool(2);
            //這裏提交任務future,則開啓線程執行RealData的call()方法執行
            //submit和execute的區別： 
            //第一點是submit能夠傳入實現Callable接口的實例對象，
            // 第二點是submit方法有返回值
            
            Future f1 = executor.submit(future);        
            //單獨啓動一個線程去執行的
            Future f2 = executor.submit(future2);
            System.out.println("請求完畢");
            
            try {
                //這裏能夠作額外的數據操做，也就是主程序執行其餘業務邏輯
                System.out.println("處理實際的業務邏輯...");
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            //調用獲取數據方法,若是call()方法沒有執行完成,則依然會進行等待
            System.out.println("數據：" + future.get());
            System.out.println("數據：" + future2.get());
            
            executor.shutdown();
        }
    
    }

Semaphore信號量
能夠控制系統的流量，拿到線程的信號量則訪問不然等待
經過acquire和release來獲取和釋放線程

final Semaphore semp = new Semaphore(5);

---

3.鎖的高級深化

Lock and Condition
使用synchronized關鍵字能夠實現線程間的同步互斥工做

使用Lock對象也能夠實現同步互斥

若是多個線程之間須要實現協做 使用Object的wait和nofity,notifyAll

在使用Lock的時候可使用一個新的等待/通知的類Condition 只針對一個具體的鎖

public class UseCondition {

        private Lock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
        
        public void method1(){
            try {
                lock.lock();
                System.out.println("當前線程：" + 
    Thread.currentThread().getName() + "進入等待狀態..");
                Thread.sleep(3000);
                System.out.println("當前線程：" + 
    Thread.currentThread().getName() + "釋放鎖..");
                condition.await();    // Object wait 
                System.out.println("當前線程：" + 
    Thread.currentThread().getName() +"繼續執行...");
                condition.signal();        //Object notify
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

ReentrantLock重入鎖

private Lock lock = new ReentrantLock(boolean isFair);//是否爲公平鎖

ReentrantReadWriteLock讀寫鎖
核心是實現讀寫分離 在都多寫少的狀況下 性能高於重入鎖

public class UseReentrantReadWriteLock {

    private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    private ReadLock readLock = rwLock.readLock();
    private WriteLock writeLock = rwLock.writeLock();
    
    public void read(){
        try {
            readLock.lock();
            System.out.println("當前線程:" + 
Thread.currentThread().getName() + "進入...");
            Thread.sleep(3000);
            System.out.println("當前線程:" + 
Thread.currentThread().getName() + "退出...");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readLock.unlock();
        }
    }
    
    public void write(){
        try {
            writeLock.lock();
            System.out.println("當前線程:" + 
Thread.currentThread().getName() + "進入...");
            Thread.sleep(3000);
            System.out.println("當前線程:" + 
Thread.currentThread().getName() + "退出...");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            writeLock.unlock();
        }
    }
}

鎖的優化

1. 避免死鎖
2. 減小鎖的持有時間
3. 減小鎖的粒度
4. 鎖的分離
5. 儘可能使用無鎖的操做 好比原子類操做