高并發Java（10）：JDK8對并發的新支持

1. LongAdder

和AtomicLong類似的使用方式，但是性能比AtomicLong更好。

LongAdder與AtomicLong都是使用了原子操作來提高性能。但是LongAdder在AtomicLong的基礎上進行了熱點分離，熱點分離類似于有鎖操作中的減小鎖粒度，將一個鎖分離成若干個鎖來提高性能。在無鎖中，也可以用類似的方式來增加CAS的成功率，從而提高性能。

LongAdder原理圖：

AtomicLong的實現方式是內部有個value 變量，當多線程并發自增，自減時，均通過CAS 指令從機器指令級別操作保證并發的原子性。唯一會制約AtomicLong高效的原因是高并發，高并發意味著CAS的失敗幾率更高， 重試次數更多，越多線程重試，CAS失敗幾率又越高，變成惡性循環，AtomicLong效率降低。

而LongAdder將把一個value拆分成若干cell，把所有cell加起來，就是value。所以對LongAdder進行加減操作，只需要對不同的cell來操作，不同的線程對不同的cell進行CAS操作，CAS的成功率當然高了（試想一下3+2+1=6，一個線程3+1，另一個線程2+1，最后是8，LongAdder沒有乘法除法的API）。

可是在并發數不是很高的情況，拆分成若干的cell，還需要維護cell和求和，效率不如AtomicLong的實現。LongAdder用了巧妙的辦法來解決了這個問題。

初始情況，LongAdder與AtomicLong是相同的，只有在CAS失敗時，才會將value拆分成cell，每失敗一次，都會增加cell的數量，這樣在低并發時，同樣高效，在高并發時，這種“自適應”的處理方式，達到一定cell數量后，CAS將不會失敗，效率大大提高。

LongAdder是一種以空間換時間的策略。

2. CompletableFuture

實現CompletionStage接口（40余個方法），大多數方法多數應用在函數式編程中。并且支持流式調用

CompletableFuture是Java 8中對Future的增強版

簡單實現：

import java.util.concurrent.CompletableFuture;

public class AskThread implements Runnable {
    CompletableFuture<Integer> re = null;

public AskThread(CompletableFuture<Integer> re) {
    this.re = re;
}

@Override
public void run() {
    int myRe = 0;
    try {
        myRe = re.get() * re.get();
    } catch (Exception e) {
    }
    System.out.println(myRe);
}

public static void main(String[] args) throws InterruptedException {
    final CompletableFuture<Integer> future = new CompletableFuture<Integer>();
    new Thread(new AskThread(future)).start();
    // 模擬長時間的計算過程
    Thread.sleep(1000);
    // 告知完成結果
    future.complete(60);
}

}</code></pre> 
  
Future最令人詬病的就是要等待，要自己去檢查任務是否完成了，在Future中，任務完成的時間是不可控的。而CompletableFuture的最大改進在于，任務完成的時間也開放了出來。
 
  
future.complete(60);
 
  
用來設置完成時間。
 
  
CompletableFuture的異步執行：
 
  
public static Integer calc(Integer para) {
        try {
            // 模擬一個長時間的執行
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
        return para * para;
    }


public static void main(String[] args) throws InterruptedException,
        ExecutionException {
    final CompletableFuture<Integer> future = CompletableFuture
            .supplyAsync(() -> calc(50));
    System.out.println(future.get());
}</code></pre> 

CompletableFuture的流式調用：
 
  
public static Integer calc(Integer para) {
        try {
            // 模擬一個長時間的執行
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
        return para * para;
    }

    public static void main(String[] args) throws InterruptedException,
            ExecutionException {
        CompletableFuture<Void> fu = CompletableFuture
                .supplyAsync(() -> calc(50))
                .thenApply((i) -> Integer.toString(i))
                .thenApply((str) -> "\"" + str + "\"")
                .thenAccept(System.out::println);
        fu.get();
    }
 
  
組合多個CompletableFuture：
 
  
public static Integer calc(Integer para) {
        return para / 2;
    }

    public static void main(String[] args) throws InterruptedException,
            ExecutionException {
        CompletableFuture<Void> fu = CompletableFuture
                .supplyAsync(() -> calc(50))
                .thenCompose(
                        (i) -> CompletableFuture.supplyAsync(() -> calc(i)))
                .thenApply((str) -> "\"" + str + "\"")
                .thenAccept(System.out::println);
        fu.get();
    }
 
  
這幾個例子更多是側重Java8的一些新特性，這里就簡單舉下例子來說明特性，就不深究了。
 
  
CompletableFuture跟性能上關系不大，更多的是為了支持函數式編程，在功能上的增強。當然開放了完成時間的設置是一大亮點。
 
  
3. StampedLock
 
  
在上一篇中剛剛提到了鎖分離，而鎖分離的重要的實現就是ReadWriteLock。而StampedLock則是ReadWriteLock的一個改進。StampedLock與ReadWriteLock的區別在于，StampedLock認為讀不應阻塞寫，StampedLock認為當讀寫互斥的時候，讀應該是重讀，而不是不讓寫線程寫。這樣的設計解決了讀多寫少時，使用ReadWriteLock會產生寫線程饑餓現象。
 
  
所以StampedLock是一種偏向于寫線程的改進。
 
  
StampedLock示例：
 
  
import java.util.concurrent.locks.StampedLock;

public class Point {
    private double x, y;
    private final StampedLock sl = new StampedLock();

    void move(double deltaX, double deltaY) { // an exclusively locked method
        long stamp = sl.writeLock();
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            sl.unlockWrite(stamp);
        }
    }

    double distanceFromOrigin() { // A read-only method
        long stamp = sl.tryOptimisticRead();
        double currentX = x, currentY = y;
        if (!sl.validate(stamp)) {
            stamp = sl.readLock();
            try {
                currentX = x;
                currentY = y;
            } finally {
                sl.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }
}
 
  
上述代碼模擬了寫線程和讀線程， StampedLock根據stamp來查看是否互斥，寫一次stamp變增加某個值
 
  
tryOptimisticRead()
 
  
就是剛剛所說的讀寫不互斥的情況。
 
  
每次讀線程要讀時，會先判斷
 
  
if (!sl.validate(stamp))
 
  
validate中會先查看是否有寫線程在寫，然后再判斷輸入的值和當前的 stamp是否相同，即判斷是否讀線程將讀到最新的數據。如果有寫線程在寫，或者 stamp數值不同，則返回失敗。
 
  
如果判斷失敗，當然可以重復的嘗試去讀，在示例代碼中，并沒有讓其重復嘗試讀，而采用的是將樂觀鎖退化成普通的讀鎖去讀，這種情況就是一種悲觀的讀法。
 
  
stamp = sl.readLock();
 
  
StampedLock的實現思想：
 
  
CLH自旋鎖：當鎖申請失敗時，不會立即將讀線程掛起，在鎖當中會維護一個等待線程隊列，所有申請鎖，但是沒有成功的線程都記錄在這個隊列中。每一個節點（一個節點代表一個線程），保存一個標記位（locked），用于判斷當前線程是否已經釋放鎖。當一個線程試圖獲得鎖時，取得當前等待隊列的尾部節點作為其前序節點。并使用類似如下代碼判斷前序節點是否已經成功釋放鎖
 
  
while (pred.locked) {   
}
 
  
這個循環就是不斷等前面那個結點釋放鎖，這樣的自旋使得當前線程不會被操作系統掛起，從而提高了性能。
 
  
當然不會進行無休止的自旋，會在若干次自旋后掛起線程。
 
  
本系列：
 
  
 
   
高并發Java（1）：前言
 
   
高并發Java（2）：多線程基礎
 
   
高并發Java（3）：Java內存模型和線程安全
 
   
高并發Java（4）：無鎖
 
   
高并發Java（5）：JDK并發包1
 
   
高并發Java（6）：JDK并發包2
 
   
高并發Java（7）：并發設計模式
 
   
高并發Java（8）：NIO和AIO
 
   
高并發Java（9）：鎖的優化和注意事項
 
   
高并發Java（10）：JDK8對并發的新支持
 
  
 
  
 
 
  
來自：http://www.importnew.com/21358.html
 
  
 
 
  
Save