JCConf 2020

2020-11-28

上星期也去參加 JCConf ，本以為今年會停辦，沒想到會正常舉行，現場也滿多人的
受疫情影響今年國外講師都是透過視訊連線，主題跟去年差不多都是環繞 Microservices
對我來說今年的亮點應該是上午場的 RSocket
現場有示範 RSocket 在 Spring 上的應用
回家後查了 RSocket 發現不是只有 Java 版本，還有 JS, Go 和 C++ 的版本

它可以有四種交互方式:

request / response (stream of 1) 與傳統的 http 一樣一去一回
request / stream (finite stream of many) 神奇，回傳一個 stream 給你操作
fire-and-forget (no response) 單向，不需要回應
channel (bi-directional streams) 雙向 streams 與 websocket 一樣一直來來回回

值得花點時間研究一下

今年的識別證

展開全文 >>

Java Unit Test 使用 Mockito + Powermock (3)

2020-08-29

離上一篇文已經間隔兩個月，今天終於有些動力來補充 Unit Test
今天要聊的是 ArgumentCaptor

在許多的情況下，我們只 assert 最終的結果 (Output)
但在整個測試的過程中有參數經過不同 method 傳遞，一不小心或粗心可能在某一層變更到數值
導致錯誤的輸入 (Input) 往下傳遞，而這個錯誤有可能不會影響最終的結果
就像一顆未爆彈潛伏在程式內，很難在短時間內發現…但未來可能在某個版本上爆發

所以我們要避免掉這個問題，會在比較重要的節點上
先驗證此時傳入到 method 的參數與我們預期的相符合

ArgumentCaptor 可以幫助我們取得執行過程中所傳遞的參數
它的用法很簡單，只需要定義與輸入參數相同類型
之後在 verify(...).callMethod(captor.capture()) 的時候，放入目標 method 中
就可以取得當下傳入的參數值，接下來就可以針對輸入內容進行 assert

ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);

...

Mockito.verify(...).callMethod(captor.capture());
Argument actual = captor.getValue();
assertThat(expected, actual)

在此使用之前做的 WebController 進行修改
先在 WebDao 增加 updateData 的 method 傳入 Data

public class WebDao {
    
    ...
    
    public Optional<Data> updateData(Data data) throws DatabaseException {
        throw new DatabaseException();
    }
}

之後我們就可以在測試期間，使用 ArgumentCaptor 來取得當時傳入的資料

WebController 也要增加一個 update() 讓我們進行更新資料的操作

public class WebController {

    ...
    
    public String update(HttpServletRequest request){
        int id = Integer.parseInt(request.getParameter("id"));
        String col1 = request.getParameter("col1");
        String col2 = request.getParameter("col2");
        Optional<Data> data;
        try {
            data = webDao.updateData(new Data(id, col1, col2));
        } catch (DatabaseException e) {
            return e.getMessage();
        }
        return GSON.toJson(data.get());
    }
}

最後在 WebControllerTest 新增一個測試，用於測試資料的更新結果

public class WebControllerTest {
    @Test
    public void Should_GetChangeData_When_UpdateDataValue() throws DatabaseException {
        final int ID = 1;
        final String ORI_COL1 = "col1";
        final String ORI_COL2 = "col2";
        
        final String NEW_COL1 = "new col1";
        final String NEW_COL2 = "new col2";
        
        // given
        Optional<Data> newData = Optional.of(new Data(ID, NEW_COL1, NEW_COL2));
        ArgumentCaptor<Data> dataCaptor = ArgumentCaptor.forClass(Data.class);
        
        when(this.request.getParameter("id")).thenReturn(String.valueOf(ID));
        when(this.request.getParameter("col1")).thenReturn(ORI_COL1);
        when(this.request.getParameter("col2")).thenReturn(ORI_COL2);
        when(this.webDao.updateData(any())).thenReturn(newData);
        
        // when
        String response = this.webController.update(this.request);
        
        // then
        try{
            Data result = this.GSON.fromJson(response, Data.class);
            
            verify(this.webDao, times(1)).updateData(dataCaptor.capture());
            
            Data inputData = dataCaptor.getValue();
            assertThat(ID, is(inputData.getId()));
            assertThat(ORI_COL1, is(inputData.getCol1()));
            assertThat(ORI_COL2, is(inputData.getCol2()));
            
            assertThat(result.getId(), is(ID));
            assertThat(result.getCol1(), is(NEW_COL1));
            assertThat(result.getCol2(), is(NEW_COL2));
        } catch (JsonSyntaxException e){
            fail("parser response data fail.");
        }
    }
}

第 13 行，我們定義了 ArgumentCaptor 用來裝 Data 類型的資料
第 27 行， verify updateData 的時候，使用 dataCaptor.capture() 取得輸入的 Data
第 29~32 行，驗證輸入的參數與我們預期的相同

後話

隨著時間推移，程式開發到一個階段，程式碼也逐漸積累量到一定的規模
有時候修改一個小地方，可能會導致其他地方也受到影響
此時如果沒有強健的測試預防開發者手殘，等到 release 後…
真的不敢想像…

最近工作上常常有機會寫到許多測試
瞭解到測試的重要性，這也是我這幾個月寫測試的感悟…

展開全文 >>

Java Unit Test 使用 Mockito + Powermock (2)

2020-06-14

一段時間沒有更新了
趁今天有空來補充一下 Uint Test 在實際開發中會需要使用到 Powermock 的場景

假設今天需要 mock 的對象只在 method 中使用
一般的 mock 做法是無法從外部引入
像這個例子，我們想要測試 Calculator ，這裡只有一個簡單的加法操作
在某個測試案例中，我們不想依賴 calc() 外部傳入的 param1 和 param2 對 Input 進行初始化
或不想回傳實際運算結果，而是強制指定 Output 的回傳內容

public class Calculator {

    public int calc(int param1, int param2) {
        Input input = new Input(param1, param2);
        Output output = new Output();

        addition(input, output);

        return output.getResult();
    }

    private void addition(Input input, Output output) {
        output.setResult(input.getParam1() + input.getParam2());
    }
}

但 Input 和 Output 不是 Calculator 的 field 無法進行 mock
目前只能透過 Powermock 的方式，跨越次元壁強制注入

在此之前，先補上 Input 和 Output 的結構

public class Input {
    private int param1;
    private int param2;

    public Input(int param1, int param2) {
        this.param1 = param1;
        this.param2 = param2;
    }

    public int getParam1() {
        return param1;
    }

    public int getParam2() {
        return param2;
    }
}

public class Output {
    private int result;

    public Output() {
        this.result = 0;
    }

    public void setResult(int result) {
        this.result = result;
    }

    public int getResult() {
        return result;
    }
}

這次改用 @RunWith(PowerMockRunner.class) 當測試的 Runner
因為指定了 @PrepareForTest(Calculator.class) ， Calculator 是我們要測試的 class
Powermock 需要對它動一些手腳，我們才可以做到跨過次元壁對 Input 和 Output 做設置

想知道為什麼要加這些 annotation 可以去 google ，網路上一堆，就不在此解釋
這裡的重點是示範如何根據測試需求 Powermock Input 和 Output

@RunWith(PowerMockRunner.class)
@PrepareForTest(Calculator.class)
public class CalculatorTest {

    private Calculator calculator;

    @Before
    public void setup() {
        this.calculator = new Calculator();
    }
    ...
}

開始第一個測試
Powermock Input ，第 7 ~ 10 行與一般 mock 一樣的流程去 mock Input
重點是第 12 行， whenNew(Input.class).withAnyArguments().thenReturn(input)
再產生 Input Object 的時候，不管傳入任何參數，都指定回傳我們要的 input

@Test
public void Should_GetResult300_When_PowerMockInputSuccess() throws Exception {
    final int PARAM1 = 100;
    final int PARAM2 = 200;

    // given
    Input input = PowerMockito.mock(Input.class);

    PowerMockito.when(input.getParam1()).thenReturn(PARAM1);
    PowerMockito.when(input.getParam2()).thenReturn(PARAM2);

    PowerMockito.whenNew(Input.class).withAnyArguments().thenReturn(input);

    // when
    int result = this.calculator.calc(anyInt(), anyInt());

    // then
    verify(input, times(1)).getParam1();
    verify(input, times(1)).getParam2();
    assertThat(result, is(PARAM1 + PARAM2));
}

這個做法就可以讓我們專注測試 Input 的變化，而不受外部參數影響
在實際開發情況下 Calculator 的 calc() 這個 method 可能外面會包了好幾層
要透過外部一層一層傳入參數到 Input 不知道要花多少時間做調整

第二個測試與第一個測試實際上是一樣的
只是 .when(input.getParam1()).thenReturn(PARAM1) 改為 .doAnswer(invocation -> PARAM1).when(input).getParam1()
doAnswer 這個在一般 mock 也很常用到，根據執行傳入參數來動態決定決定 return 的結果

@Test
public void Should_GetResult300_When_PowerMockInputDoAnswerSuccess() throws Exception {
    final int PARAM1 = 100;
    final int PARAM2 = 200;

    // given
    Input input = PowerMockito.mock(Input.class);

    PowerMockito.doAnswer(invocation -> PARAM1).when(input).getParam1();
    PowerMockito.doAnswer(invocation -> PARAM2).when(input).getParam2();

    PowerMockito.whenNew(Input.class).withAnyArguments().thenReturn(input);

    // when
    int result = this.calculator.calc(anyInt(), anyInt());

    // then
    verify(input, times(1)).getParam1();
    verify(input, times(1)).getParam2();
    assertThat(result, is(PARAM1 + PARAM2));
}

在目前的需求中使用測試一的寫法就夠了
那為什麼在這裡多提醒 doAnswer 的寫法呢？

因為實際開發中我們會用到 Powermock 的情況下， mock 的對象通常會是 static
像這種共用的 Object 和 method 傳入的來源特別雜
所以需要根據傳入的參數回傳不同的結果
此時 doAnswer 會比 thenReturn 和 doReturn 好用

第三個測試是強制設置 Output
用法與 Input 類似，第 10 行由 withAnyArguments 改為 withNoArguments
這裡演示沒有參數傳入的做法
第 11 行改用 thenAnswer ，這與 doAnswer 用法一樣，根據個人習慣或測試邏輯選一個用就好

這裡要注意第 14 行不可以傳入 anyInt() 不然會在 calc() 內部
Input input = new Input(param1, param2) 初始化時發生錯誤
除非使用測試一和測試二的寫法 Powermock Input

這是一般寫測試的新手比較難注意到的地方
因為 Powermock 的錯誤訊息並不會提示這個問題
只能夠自己小心

@Test
public void Should_GetResult1000_When_PowerMockOutputSuccess() throws Exception {
    final int PARAM1 = 100;
    final int PARAM2 = 200;
    final int OUTPUT_RESULT = 1000;

    // given
    Output output = PowerMockito.mock(Output.class);

    PowerMockito.whenNew(Output.class).withNoArguments().thenReturn(output);
    PowerMockito.when(output.getResult()).thenAnswer(invocation -> OUTPUT_RESULT);

    // when
    int result = this.calculator.calc(PARAM1, PARAM2);

    // then
    verify(output, times(1)).setResult(anyInt());
    verify(output, times(1)).getResult();
    assertThat(result, is(OUTPUT_RESULT));
}

執行結果

3 個測試都通過，與預計的結果相符合

Unit Test coverage 100%

今天 Unit Test 就談到這

實際開發中 Powermock 可以不使用就不要使用，因為它一但用不好
可能會影響到其他的測試案例….
但有時候卻是必要之惡，不用 Powermock 有些測試情況很難模擬和維護
最終造成測試 coverage 不夠完整

詳情可參考我的 github source code:
https://github.com/HengLin31/java-unit-test/blob/master/src/test/java/pers/henglin/test

展開全文 >>

Java Unit Test 使用 Mockito + Powermock (1)

2020-04-25

目前開發的產品很重視 Unit Test ，需要根據使用情境和 source code 本身寫測試
且 Unit Test coverage 都可達到 95% ~ 100%

在寫一段時間的測試後，也有了紀錄一下測試心得的想法
當然測試的方式和組合也百百種，不可能一下子就講齊
這裡會使用簡單的測試案例來說明怎麼寫測試

第一個測試案例是 web controller
因為網路的興起，大部分寫 Java 都會使用 spring 搭配 MVC 架構來進行開發
對於測試來說，最麻煩的就是需要與資料庫或底層資料互動
依據資料庫回傳的結果進行不同的行為或流程
幸運的是，資料庫通常都會將其抽為 dao 層
在進行測試時，我們只需要針對 dao 的 input 和 output 進行 mock 做不同行為綁定

由於 mock 可以設定特定的行為，設定當 dao 輸入哪些參數時， dao 要回傳什麼資料
讓整個過程中不會使用到 dao 真實行為，而是使用 mock 設計好的行為
這樣可以隨意操控 dao ，讓它回傳或者拋出測試時所需的資料或異常
進而達成我們針對 web controller 不同情境下測試的目的

建立一個 web dao 來使用 id 取得資料，因為沒有真正實作細節，所以直接拋出自定義的 database exception

public class WebDao {

    public Optional<Data> findDataById(int id) throws DatabaseException {
        throw new DatabaseException();
    }
}

exception 預設訊息為無法連線資料庫

public class DatabaseException extends Exception {
    private static final String UNABLE_CONN_DB_EXCEPTION = "Unable to connect to database.";
    
    public DatabaseException() {
        super(UNABLE_CONN_DB_EXCEPTION);
    }
}

data bean 格式

public class Data {
    private int id;
    private String col1;
    private String col2;
    ...
}

web controller 中定義 web dao ，因為沒有要真的使用 web dao
所以沒有引入 spring 針對 web dao 綁定 @autowired 做注入

當前只有一個 read(...) 根據 request 傳入 data id 後透過 dao 取得結果

public class WebController {
    private final Gson GSON = new Gson();
    private WebDao webDao;
    
    public String read(HttpServletRequest request){
        int id = Integer.parseInt(request.getParameter("id"));
        Optional<Data> data;
        try {
            data = webDao.findDataById(id);
        } catch (DatabaseException e) {
            return e.getMessage();
        }
        if(data.isEmpty()){
            return "not find data by id: " + id;
        }
        return GSON.toJson(data.get());
    }
}

簡單分析這個 read(...) 可以知道執行過程中可能有 3 種情況發生

正常從 dao 取得 data 後回傳結果 ( 執行到 16 行 )
從 dao 取到的 data 結果為空，回傳錯誤訊息 ( 執行到 14 行 )
資料庫拋出異常，回傳錯誤訊息 ( 執行到 11 行 )

而我們需要針對這 3 種不同的情境進行測試，確保每個情境最終結果與我們預測的一樣

開始寫測試
Mockito 有幾種掛載方式，如果只是簡單的測試，我個人是習慣直接使用＠RunWith 設定 runner
這裡需要注意的是 setup() 這裡指定要 mock 的對象主要有兩個， request 和 webDao
request 模擬 read(...) 的入參，因為我們不希望真的產生 HttpServletRequest
webDao 模擬實際的 dao ，因為我們不會實際對資料庫做連線
最後 instance webController 以便進行後續的測試操作

此時有一個問題， webController 要使用 webDao 需要將它注入到 webController
但是我們並沒有對外開放 set method 用於注入 webDao

這種情況在實際開發過程中常常會遇到，我們不可能為了測試方便特別修改 source code
此時需要一些特別的手段突破注入的次元壁，而 Powermock 可以透過侵入式的方式強行進行注入，進而解決我們的問題

這裡採用 Powermock 的 Whitebox.setInternalState(Object object, Class<?> fieldType, Object value)
第一個參數是注入目標對象，後面兩個參數是要注入的類型與其資料

@RunWith(MockitoJUnitRunner.class)
public class WebControllerTest {
    private final Gson GSON = new Gson();

    private HttpServletRequest request;
    private WebDao webDao;
    private WebController webController;
    
    @Before
    public void setup() {
        this.request = mock(HttpServletRequest.class);
        this.webDao = mock(WebDao.class);
        
        this.webController = new WebController();
        // use powermock to mock private field
        Whitebox.setInternalState(this.webController, WebDao.class, this.webDao);
    }
    ...
}

關於 Powermock 的使用會在之後的文章慢慢補齊，此案例只用到 Whitebox
現階段主要還是會以情境搭需求使用，需要用到 Powermock 時才會說明

第一個測試案例
正常取得資料的情況

測試名稱採用 Should_ExpectedBehavior_When_StateUnderTest 的命名方式
這是網路上建議測試時採用的命名規則之一，有興趣的可以 google 一下其他測試命名規則

這個命名規則的好處是測試名稱即情境，在多人合作下可以清楚知道同事主要是針對哪個情況去測試，不用深入去看 source code

@Test
public void Should_ReturnData_When_GetDataById() throws DatabaseException {
    final int ID = 1;
    final String COL1 = "col1";
    final String COL2 = "col2";
    
    // given
    Optional<Data> data = Optional.of(new Data(ID, COL1, COL2));
        
    when(this.request.getParameter("id")).thenReturn("1");
    when(this.webDao.findDataById(anyInt())).thenReturn(data);
    
    // when
    String response = this.webController.read(this.request);
    
    // then
    try{
        Data result = this.GSON.fromJson(response, Data.class);
        
        verify(this.webDao, times(1)).findDataById(anyInt());
        assertThat(result.getId(), is(ID));
        assertThat(result.getCol1(), is(COL1));
        assertThat(result.getCol2(), is(COL2));
    } catch (JsonSyntaxException e){
        fail("parser response data fail.");
    }
}

第 10 行和 11 行，針對 read(...) 執行過程中會用到行為做設置
這裡設定當觸發 request.getParameter("id") 這個行為需要回傳字串”1”
當觸發 webDao.findDataById(anyInt()) 這個行為時回傳第 8 行我們設置的假資料
這裡可能會有疑問是為什麼 id 回傳是設置字串”1”而不是 anyString()
這是因為我們在執行 read(...) 過程中會針對 id 轉為 integer ，避免傳入非整數的字串造成問題
而 webDao.findDataById(anyInt()) 則沒這個問題，我們不在意傳入當 id 的 int 數值是什麼
最終都會傳回我們自己設置的假資料

第 20 行驗證 webDao.findDataById(...) 有被呼叫過一次，確保目標 method 有被正常執行到
第 21 ~ 23 行，檢查回傳的資料結構與數值與我們預計的相符合
第 25 行可以針對資料結構異常時，告知我們測試失敗，否則當轉換失敗會無法執行到 20 ~ 23 行做資料驗證，導致測試案例成功的異常問題

第二個測試案例
測試資料庫拋出異常時，回傳對映的異常訊息與我們預計的是否相符合

@Test
public void Should_Fail_When_UnableConnDatabase() throws DatabaseException {
    final String EXPECT_EXCEPTION_MSG = "Unable to connect to database.";
    
    // given
    DatabaseException exception = new DatabaseException();
    
    when(this.request.getParameter("id")).thenReturn("1");
    when(this.webDao.findDataById(anyInt())).thenThrow(exception);
    
    // when
    String response = this.webController.read(this.request);
    
    // then
    verify(this.webDao, times(1)).findDataById(anyInt());
    assertThat(response, is(EXPECT_EXCEPTION_MSG));
}

第 9 行拋出我們第 6 行設置的異常
第 12 行 response 的資料會是異常訊息
第 16 行驗證錯誤訊息與我們預計的相符合

第三個測試案例
測試當使用 id 查詢沒有取得資料時，回傳對映的異常訊息與我們預計的是否相符合

@Test
public void Should_Fail_When_NotFindData() throws DatabaseException {
    final String EXPECT_EXCEPTION_MSG = "not find data by id";
    
    // given
    Optional<Data> data = Optional.empty();
    
    when(this.request.getParameter("id")).thenReturn("1");
    when(this.webDao.findDataById(anyInt())).thenReturn(data);
    
    // when
    String response = this.webController.read(this.request);
    
    // then
    assertThat(response, anything(EXPECT_EXCEPTION_MSG));
}

第 9 行回傳資料提換為我們第 6 行設置的空資料
第 12 行 response 的資料會是找不到 data 的異常訊息
第 15 行驗證錯誤訊息與我們預計的相符合

執行結果

3 個測試都通過，與預計的結果相符合

web controller unit test coverage 100%

關於 Java Unit Test 的測試案例解說，暫時告一段落
有空再補齊其他測試案例

詳情可參考我的 github source code:
https://github.com/HengLin31/java-unit-test/blob/master/src/test/java/pers/henglin/test

展開全文 >>

關於 Java CompletableFuture 的用法

2020-04-19

已經好久沒有進行更新
最近都在熟悉新工作，與過往不同的 domain knowledge ，融入貫通的過程中學習到滿多關於硬體的知識，目前已經進入狀況

剛好最近都在使用 CompleteableFuture 處理 thread 非同步的事件
來寫篇文章記錄一下用法，方便之後查詢使用

CompletableFuture 個人覺得很像 JS 的 Promise
JS Promise 串接async callback function
而 CompletableFuture 用於串連不同的 thread 接點，某種程度將一連串執行流程中 thread 之間的前後相依性解耦
某個 thread 結果要串給另一個 thread 當作輸入條件，使用 CompletableFuture 可輕鬆綁定
所以在設計時只需定義好每個事件個自的 input 和 output ，不需要將關聯寫死

首先是最簡單的 runAsync 和 supplyAsync 的基礎用法，差別在於是否需要 return result
它們都可以搭配 thread pool 使用 ( 在實際應用場景都會根據實際需求自建 thread pool 使用，若沒設置則預設是使用 JVM 本身的 thread pool )

// basic
CompletableFuture<Void>  runAsync(Runnable runnable)
CompletableFuture<T>     supplyAsync(Supplier<T> supplier)

// use your own thread pool
CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)
CompletableFuture<T>     supplyAsync(Supplier<T> supplier, Executor executor)

提供自建的 thread pool 給 CompletableFuture 使用

1
2
3

ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> System.out.println("runAsync"), threadPool);
CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> "supplyAsync", threadPool);

CompletableFuture 也允許在事件執行完成時 callback 取得結果，方便進行下一步的處理，而且在執行過程中發生 exception ，也可以攔截 exception 後執行特定的 action

// basic
CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)
// catch exception
CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)

// use your own thread pool
CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)

將剛剛的 supplyAsync 例子改寫為強制拋出 exception ，可在 completeAsync 或 exceptionally 處理異常，非常的方便

ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
    throw new CompletionException(new Exception("throw exception"));
}, threadPool);

supplyAsync.whenCompleteAsync((result, ex) -> {
    System.out.println("result: " + result);
    System.out.println("exception: " + ex);
}, threadPool).exceptionally(ex -> {
    System.out.println("exceptionally: " + ex.getMessage());
    return ex.getMessage();
}).join();

輸出

1
2
3

result: null
exception: java.util.concurrent.CompletionException: java.lang.Exception: throw exception
exceptionally: java.lang.Exception: throw exception

除此之外 CompletableFuture 也提供 handle 來處理 exception ，差別在於 handle 是允許有返回值，用於串接過程中處理 exception
handle 也用於做進一步的資料轉換

// basic
CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)

// use your own thread pool
CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

再次改寫 supplyAsync ，改為使用 handle 處理 exception ，此例子將 exception 轉換成字串當作一般的結果往後傳遞

ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
    throw new CompletionException(new Exception("throw exception"));
}, threadPool);

String ans = supplyAsync.handle((result, ex) -> (null != ex) ? ex.getMessage() : result).join();
System.out.println("ans: " + ans);

輸出

1	ans: java.lang.Exception: throw exception

若是沒有處理 exception 的需求則可以使用 thenApply 進行資料的轉換，類似於 stream map 的效果

// basic
CompletableFuture<U> thenApply(Function<? super T, ? extends U> fn)
CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn)

// use your own thread pool
CompletableFuture<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor)

還有對映 flatMap 的 thenCompose

CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn)

CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor)

不需要返回值的 thenAccept

// basic
CompletableFuture<Void> thenAccept(Consumer<? super T> action)
CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)

// use your own thread pool
CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

使用 thenCombine 還可以將兩個獨立的 CompletableFuture 執行結果進行整合，非常的強大

// basic
CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn)
CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn)

// use your own thread pool
CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T, ? super U, ? extends V> fn, Executor executor)

現在將 supplyAsync1 和 supplyAsync2 結果整合

ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletableFuture<String> supplyAsync1 = CompletableFuture.supplyAsync(() -> "supplyAsync 1", threadPool);
CompletableFuture<String> supplyAsync2 = CompletableFuture.supplyAsync(() -> "supplyAsync 2", threadPool);

String ans = supplyAsync1.thenCombine(supplyAsync2, (result1, result2) -> result1 + ", " + result2).join();
System.out.println("and: " + ans);

輸出

1	and: supplyAsync 1, supplyAsync 2

不需要返回值的 thenAcceptBoth

// basic
CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,  Runnable action)

// use your own thread pool
CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)

使用方式與 thenCombine 相同

ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletableFuture<String> supplyAsync1 = CompletableFuture.supplyAsync(() -> "supplyAsync 1", threadPool);
CompletableFuture<String> supplyAsync2 = CompletableFuture.supplyAsync(() -> "supplyAsync 2", threadPool);

supplyAsync1.thenAcceptBothAsync(supplyAsync2, (result1, result2) -> System.out.println(result1 + ", " + result2), threadPool).join();

輸出

1	supplyAsync 1, supplyAsync 2

前面提到的都是兩個獨立 CompletableFuture 的例子，若大於兩個的時候會使用 allOf
allOf 可以等待全部的 CompletableFuture 執行完成後執行

1	CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

等待 supplyAsync1, supplyAsync2, supplyAsync3 執行結束後取得最終結果

ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletableFuture<String> supplyAsync1 = CompletableFuture.supplyAsync(() -> "supplyAsync 1", threadPool);
CompletableFuture<String> supplyAsync2 = CompletableFuture.supplyAsync(() -> "supplyAsync 2", threadPool);
CompletableFuture<String> supplyAsync3 = CompletableFuture.supplyAsync(() -> "supplyAsync 3", threadPool);

CompletableFuture.allOf(supplyAsync1, supplyAsync2, supplyAsync3).thenRun(() -> {
    try {
        StringBuffer ans = new StringBuffer();
        ans.append(supplyAsync1.get()).append(", ")
            .append(supplyAsync2.get()).append(", ")
            .append(supplyAsync3.get());
        System.out.println("ans: " + ans.toString());
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }
}).join();

輸出

1	ans: supplyAsync 1, supplyAsync 2, supplyAsync 3

在 allOf 後面串接一個 thenRun
它的用法很簡單，使用方式與 thenAccept 一樣，差別在於不需要傳入上一個 CompletableFuture 的執行結果

// basic
CompletableFuture<Void> thenRun(Runnable action)
CompletableFuture<Void> thenRunAsync(Runnable action)

// use your own thread pool
CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

上述提到的都是全部完成的例子，反過來說我只想要其中一個有完成就往下執行
CompletableFuture 也可以做到這點

// basic
CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn)
CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn)

// use your own thread pool
CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor)

沒有返回值的 acceptEither

// basic
CompletableFuture<Void>  acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
CompletableFuture<Void>  acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)

// use your own thread pool
CompletableFuture<Void>  acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)

用於多個 CompletableFuture 的 anyOf ，在眾多 CompletableFuture 只需一個完成就可往下執行

1	CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

展開全文 >>

Java NIO ByteBuffer 基本操作圖

2020-01-13

剛剛在刪舊圖片時發現以前為了學 ByteBuffer 時自己畫的操作圖

這張圖簡單顯示 ByteBuffer 的寫入與讀取時 position point 和 limit point 的位移
allocate 宣告 buffer capacity ，配給 JVM heap ，屬於 GC 可管範圍，需要 copy data ，速度慢於 allocateDirect
put(bytes[4]) position 一路從 0 移動到 4 ，0 ~ 3 依序放入內容值， 4 為下一個寫入位置
flip() 為準備讀取 buffer 資料前的動作，為了讀取剛剛寫入 0 ~ 3 的內容，則改變 point 指向位置 limit = position 接著 position = 0
get(bytes[4]) position 一路從 0 移動到 4 ，0 ~ 3 依序取出內容值
clear() 清除不是真的清除資料，也只是改變 point 指向位置， position = 0 且 limit = capacity ，因為下次有資料寫入會直接覆蓋 buffer 內的舊值

使用 NIO 時會經常使用 ByteBuffer 來做讀取和寫入資料，是一種基於 channel 和 buffer 的 I/O 方式
還有一種是 ByteBuffer.allocateDirect(capability) 配給 OS local memory ，不屬於 GC 可控制範圍，不需要 copy 所以速度較快
換句話說，它可以使用 native 直接配給 java heap 外的 memory ，之後通過 java heap 裡面的 DirectByteBuffer 針對這一塊 memory refernce 進行操作
可在某些使用情況下可提高性能，它避免了 java heap 和 native heap 中來回 copy data

展開全文 >>

對於 Java Fork Join framework 的理解

2020-01-07

老實說我在工作中沒直接用過 Java 的 Fork Join 只有以前在作自主練習時用過，知道這個東西
如果沒有特別需要，也沒有必要為了使用而使用，一切還是要回歸需求面
因最近複習 multi-thread 的相關知識，才有這個機會再次使用
它可 parallel 處理 divide and conquer 的問題
適合可獨立拆分的問題，且拆出的子問題之間不要存在相依性
感覺很像 MapperReduce 的運作

ForkJoinPool:
與 ThreadPoolExecutor 一樣都是繼承自 AbstractExecutorService
fork(): 將分割的 task 放入 workQueue ，task 交由 ForkJoinPool 處理
join(): blocking 等待 task 完成
與 ThreadPoolExecutor 不同的是 ForkJoinPool 內的所有 workThread 都有自己的 workQueue
而 ThreadPoolExecutor 是大家共用一個 workQueue
這裡要注意的是 workThread 的 workQueue 是 Deque
workThread 會從 workQueue 頭部取 task
當 workThread 內的 workQueue 沒有 task 可執行，會從其他 workThread 的 workQueue 尾端取 task

目前看到比較常用的三種 ForkJoinTask
RecursiveTask: 有返回值
RecursiveAction: 無返回值
CountedCompleter: 完成 task 後將觸發其他 task

這裡練習使用 RecursiveTask 寫一個 merge sort

public class MergeSort extends RecursiveAction {
    private final int THRESHOLD;
    private int[] arr;
    
    public MergeSort(final int[] arr, final int threshold){
        this.arr = arr;
        this.THRESHOLD = threshold;
    }
    
    @Override
    protected void compute() {
        if(arr.length <= THRESHOLD){
            Arrays.sort(arr);
            return;
        }
        
        int midIndex = arr.length / 2;
        int[] leftArr = Arrays.copyOfRange(arr, 0, midIndex);
        int[] rightArr = Arrays.copyOfRange(arr, midIndex, arr.length);
        
        MergeSort left = new MergeSort(leftArr, THRESHOLD);
        MergeSort right = new MergeSort(rightArr, THRESHOLD);
        
        left.fork();
        right.fork();
        
        left.join();
        right.join();
        
        System.out.println(Arrays.toString(left.getArr()));
        System.out.println(Arrays.toString(right.getArr()));
        
        arr = mergeArr(left.getArr(), right.getArr());
    }
    
    public int[] getArr(){
        return arr;
    }
    
    private int[] mergeArr(final int[] leftArr, final int[] rightArr){
        int[] resultArr = new int[leftArr.length + rightArr.length];
        
        int resultIndex = 0;
        int leftIndex = 0;
        int rightIndex = 0;
        
        while(leftIndex < leftArr.length && rightIndex < rightArr.length){
            if(leftArr[leftIndex] <= rightArr[rightIndex]){
                resultArr[resultIndex] = leftArr[leftIndex];
                leftIndex++;
            }else{
                resultArr[resultIndex] = rightArr[rightIndex];
                rightIndex++;
            }
            resultIndex++;
        }
        
        while(leftIndex < leftArr.length){
            resultArr[resultIndex] = leftArr[leftIndex];
            leftIndex++;
            resultIndex++;
        }
        
        while(rightIndex < rightArr.length){
            resultArr[resultIndex] = rightArr[rightIndex];
            rightIndex++;
            resultIndex++;
        }
        
        return resultArr;
    }
}

執行 merge sort

public static void main(String[] args){
    int[] arr = new Random().ints(25, 1, 100).toArray();
    System.out.println("ori: " + Arrays.toString(arr) + "\n ---");
    
    ForkJoinPool threadPool = new ForkJoinPool(5);
    MergeSort mergeSort = new MergeSort(arr, 5);
    threadPool.invoke(mergeSort);
    System.out.println("---\nresult: " + Arrays.toString(mergeSort.getArr()));

}

執行結果:

這篇聊得比較少，也只練習了 RecursiveTask
目前稍微熟悉一下就好，等未來有需要使用時再深入了解吧

展開全文 >>

Java BlockingQueue 適合應用在哪些場景

2020-01-06

BlockingQueue 適用於緩衝如經典的 producer / consumer 場景
大多用在多個 thread 之間倆倆需要共享資料時，一個放另一個取，非常方便，通常我用它來取代 wait() 和 notify() 做 thread 通信
上一篇 ThreadPoolExecutor 第5個參數就是 BlockingQueue 用來存放等待從 thread pool 取得 thread 的 task
可根據不同的應用場景可搭配適合的 BlockingQueue
這個主題偏向於應用面的思考，針對不同場景選擇使用 BlockingQueue

BlockingQueue:
ArrayBlockingQueue

適用場景:
在非高併發任務，且可預估出固定處理速率時
如: 一般的緩衝區

特色:
queue 容量固定個數，屬於有 bounded 的類型，滿或空的時候會 blocking
放入與取出使用同一個 lock ，所以不利於高併發下使用

BlockingQueue:
PriorityBlockingQueue

適用場景:
當需要優先處理特定任務時，需讓高優先權任務盡快被取出處理，再接著處理次要的任務
如: 在有階級機制下的搶票系統，可根據階級設定每個人的優先權，讓高優先權的人優先取得票

特色:
queue 容量為 Integer.MAX_VALUE – 8 ， Java doc 上說明為 unbounded
根據 priority 排序，優先級高的會被優先取出處理
比較特別的是它 add(), offer(), put() 操作都一樣，都是 offer() 操作

BlockingQueue:
LinkedBlockingQueue

適用場景:
需要高併發情況下使用，但需要評估取出時的消耗率，避免高併發下取出速度趕不上放入速度造成 OutOfMemory
如: 即時訊息處理

特色:
queue 容量可調整， Java doc 上說明為 optionally-bounded ，預設值為 Integer.MAX_VALUE，但一般來說會根據需求設值
用於高併發，使用讀寫分離，放入使用 putLock 而取出使用 takeLock
在上一篇 ThreadPoolExecutor 有稍微提到
且 newFixedThreadPool, newSingleThreadExecutor 都使使用 LinkedBlockingQueue 作為 workQueue

BlockingQueue:
LinkedBlockingDeque

適用場景:
有雙向需求
如: real-time 處理任務，在某些危險的情況下需要警急處理終止命令

特色:
有雙向需求，基本上與 LinkedBlockingQueue 相似
可以當作極端的 PriorityBlockingQueue ，但只有一般和最高兩種權限

BlockingQueue:
SynchronousQueue

場景:
目前想不到單獨使用的場請，所以需搭配 newCachedThreadPool 使用
newCachedThreadPool 適用於在特定情況或時間下會才一瞬間出現大量 task，但在大部分情況下處理量不高
如: 某些監控系統或感應器，只有在警報發生或瞬間觸發多個動作時需要處理大量 task ，其餘時間皆為正常的處理量

特色:
它沒有容量，所以 peek() 永遠為空，底層是 transfer 且等待為0秒，放與取只能手把手地快速交換
目前所知需搭配 newCachedThreadPool 使用，其底層實作為上一篇提到的 ThreadPoolExecutor
因 SynchronousQueue 沒容量，因此 newCachedThreadPool 可以根據 task 需求量在 thread pool 中創建多個 thread 直到設定的上限值
而不受 workQueue 影響

BlockingQueue:
DelayQueue

適用場景:
延後處理，或者有時間倒數的情況
如: 某些嚴格的登入系統因安全性考量，當每次登入後可能只有3分鐘時間可處理，等時間超過3分鐘則會自動登出

特色:
queue 容量為 Integer.MAX_VALUE ， Java doc 上說明為 unbounded
task 放入 queue 後需要過一段時間後才可被取出
假設目前有 t1, t2 兩個 task 需放入 queue
將 t1 設定5秒後處理後放入 queue ，接著 t2 設定2秒後處理後放入 queue
拿的時候會先 blocking 等到2秒後先取出 t2 ，若 t2 需處理較長的時間，假設需要10秒
此時 t1 還在 queue 中，所以可能在10秒後才從 queue 中取出 t1
反之 t2 若瞬間處理結束，則需 blocking 直到5秒後取出 t1

BlockingQueue:
LinkedTransferQueue

適用場景:
producer 壓榨 consumer ，producer 想要在指定時間內或需要 consumer 立即處理，此時 consumer 需要有快速消耗的能力
如: 追求高併發極限，只要有 task 就立即處理

特色:
確保 producer 的資料能立即被 consumer 處理
底層與 SynchronousQueue 一樣都是使用 transfer ，除非超時，否則會一直 blocking
一般的 BlockingQueue 是 producer 放完資料就結束，除非 queue 滿了才會 blocking
而 LinkedTransferQueue 需要等 consumer 處理才算結束，不然就 blocking

展開全文 >>

對於 Java ThreadPoolExecutor 的理解

2020-01-05

提到 thread 就會想到 thread pool ，因為在實際應用上可能需要處理成千上萬的 task
不可能一個 task 就開一個 thread 且 thread 過多光是頻繁的 context switch 就可能造成效能低落
頻繁的創建與銷毀 thread 也是不小的開銷
所以一般都是根據實際需求創建 thread pool 來控制 thread 個數
Executors 有提供好幾種 thread pool 如 newFixedThreadPool, newSingleThreadExecutor, newCachedThreadPool 等
它們的底層都是使用 ThreadPoolExecutor 搭配不同參數所實現的
在正常情況下我會依據問題的需求自己調整參數，因為 Executors 大多使用 LinkedBlockingQueue 其預設容量為 Integer.MAX_VALUE
在高併發下，最終可能累計過多 task 造成 OutOfMemory 的情況發生

ThreadPoolExecutor 有7個參數可供調整，只要熟悉每個參數的使用，就可以根據需求來設計 thread pool

public ThreadPoolExecutor(
            int corePoolSize,
            int maximumPoolSize,
            long keepAliveTime,
            TimeUnit unit,
            BlockingQueue<Runnable> workQueue,
            ThreadFactory threadFactory,
            RejectedExecutionHandler handler) {
                ...
}

corePoolSize: thread pool 建立後初始化的 thread 個數
maximumPoolSize: thread pool 內同時間執行 thread 的上限值，因為當 workQueue 滿了又有 task 進入時需要創建更多 thread 來處理，但不能無限增加 thread 數量，故使用此設定值
keepAliveTime: thread pool 沒用到的空閒 thread 的存活時間，因為當 loading 降低時，需要將 thread pool 個數需要降回 corePoolSize 減少資源的浪費
unit: keepAliveTime 時間的單位
workQueue: 任何實現 BlockingQueue 的 queue，放尚未被執行的 task
threadFactory: 沒什麼好說的，就生成 thread 用的 factory ，用預設值就好
handler(rejected): 當 workQueue滿了且 thread pool 內的 thread 已達 maximumPoolSize 的上限，無法處理新的 task 所以需進行額外的處理

畫個圖簡單說明 ThreadPoolExecutor 執行的情況，以下系列圖中的 t 為 task ，所以 t1 表示 task 1
實線方體為可用 thread ，虛線方體為未存在的 thread ，長方灰階為 work queue
這裡 corePoolSize = 2 所以有兩個實線方體， maximumPoolSize = 5 所以最多可生成5個 thread ， workQueue = 4 所以長方灰階最多可排4個 task

假設目前 t1, t2, t3 3個 task 執行 execute() 此時需決定是否取得 thread 執行 task ，或者放入 workQueue 等待

可用 thread 個數為2，所以 t1, t2 取得 thread 執行，而 t3 進入 workQueue 等待

此時 t4, t5, t6 3個 task 執行 execute() 後需決定是否取得 thread 執行 task ，或者放入 workQueue 等待

無可用 thread 所以 t4, t5, t6 放入 workQueue 等待

很快的 t7, t8, t9 也執行 execute()

無可用 thread 且 workQueue 已滿，所以生成3個 thread 讓 t3, t4, t5 取得 thread 後 t7, t8, t9 放入 workQueue

t10 執行 execute() 後發現 thread 個數已達 maximumPoolSize 且 workQueue 已滿，觸發 handler 處理
RejectedExecutionHandler 提供四種策略
AbortPolicy: 直接拋出 RejectedExecutionException 阻止正常執行
CallerRunsPolicy: 不會觸發 exception，回給 caller 執行
DiscardOldestPolicy: 移除 workQueue 內等待最久的 task ，並將新的 task 放入
DiscardPolicy: 直接拒絕 task 不觸發任何 exception

過一段時間 workQueue 已空且只剩2個 thread 被 t8, t9 取用

又過了一段時間，根據 keepAliveTime = 60, unit = sec 所以在3個未用的 thread 經過60秒後銷毀

已無任何 task 需要執行， thread pool 回到原始狀態

了解執行原理後，接下來就需要根據經驗來調整參數
要如何合理的配置 thread pool 的 thread 個數
這需要牽扯到要解的問題是遇到 CPU bound 還是 I/O bound
若是 CPU bound 可以設置 maximumPoolSize 設為趨近目前電腦或 VM 所配置的核心數，減少 context switch 的切換
Java 可用 Runtime.getRuntime().availableProcessors() 取得此數據
若是 I/O bound 就可以盡量設置高一點，因為在做頻繁 I/O 時 task 大多需要等待 I/O 執行完畢，卡在身上不如給其他人使用，提高 CPU 的利用率
以此為基礎多跑幾次實驗，最終可調出較佳的結果，這東西很迷沒有正確的數值只能慢慢調整

結論

為了效能需求 thread pool 最好使用 ThreadPoolExecutor 依據需求自行設定
請根據 CPU bound 或 I/O bound 進行調整 thread 個數

展開全文 >>

Java ReentrantLock 和 ReentrantReadWriteLock 的使用時機

2020-01-04

在實際工作中我大多使用 ReentrantReadWriteLock 因為遇到的使用情景都是讀取的需求遠大於寫入
所以除非某些場景需要使用 synchronized 為了更好的效能可能會用 ReentrantLock 取代
synchronized 是 JVM 層級的同步鎖，底層機制鎖死整個區塊，而 Lock 類的是使用 java.util.concurrent 下實現的 lib
而且可以綁定多個 condition 可以喚醒特定的 thread (thread.signal())，不像 synchronized 只能一次喚醒全部或者隨機一個，彈性上強上許多

如果用一句話解釋 ReentrantLock 和 ReentrantReadWriteLock 兩者的差別就是
ReentrantLock: One thread at a time.
ReentrantReadWriteLock: One writer thread at a time or multiple reader threads at a time.
它們分別 implement Lock 和 ReadWriteLock
ReadWriteLock其實就是基於 Lock 上作讀寫分離

public interface ReadWriteLock {
    Lock readLock();
    Lock writeLock();
}

先聊 ReentrantLock
在正常使用下只需針對存取變數block包覆

private static ReentrantLock lock = new ReentrantLock();

private static void accessResource() {
    lock.lock();
    try {
        //access the resource
    } finally {
        lock.unlock();
    }
}

ReentrantLock 如字面上的意思他是所以可以重入(會計算重入次數，離開時遞減次數)，但 lock 和 unlock 必須成對出現
但在一般情況下是不會使用兩層 lock ，主要是針對不同 method 之間有交互引用時，存取共享變數資料時好用

private static ReentrantLock lock = new ReentrantLock();

private static void accessResource() {
    lock.lock();
    lock.lock();
    try {
        //access the resource
    } finally {
        lock.unlock();
        lock.unlock();
    }
}

還有它初始化時可以設置公平鎖和非公平鎖

1
2
3

public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

FairSync: multi-thread 按照申請鎖的順序來獲取鎖，類似排隊，先來後到
NonfairSync: 獲取鎖的順序並不是按照申請鎖的順序，可能後申請的 thread 比先申請的優先權高，在高併發的情況下，可能造成優先級別反轉或者飢餓現象(長時間都拿不到cpu執行權限)

接著來談 ReentrantReadWriteLock
簡單的說它與 ReentrantLock 最大的不同在於 readLock 可以放多個讀取操作進入，而且 readLock 與 writeLock 是互斥的
為了更容易了解，就簡單畫個圖來說明，這裡的 t 表示 thread

當 t1 取得鎖在執行寫入操作時(writeLock)，後面有3個thread進入等待
此時 t1 執行完畢後，接著執行 readLock ，允許多個 read 操作進入(same read lock)，所以 t2 和 t4 也進入讀取變數
t5 進入等待區，t5 雖然也是讀取操作，但沒有與 t2 和 t4 一樣進入區間，因為在 t2 和 t4進入後 read 操作的 thread 等待個數為 0 所以下次需讓給 write lock
t2 和 t4 讀取結束，放 t3 進入，因為 t3 為寫操作，所以只能是自已進入作寫入(single thread)

Java中的ReentrantReadWriteLock使用一個 int 作為鎖的計數
sharedCount() 共享鎖(高位元16)，用於 readLock ，允許多個 read thread
exclusiveCount() 獨佔鎖(低位元16)，用於 writeLock ，只允許一個 write thread

有時候為了確保讀寫互斥安全性會做鎖降級或鎖升級
鎖降級: 在 writeLock 沒有釋放的時候，獲取 readLock，再釋放 writeLock
鎖升級: 在 readLock 沒有釋放的時候，獲取 writeLock，再釋放 readLock
這裡示範鎖降級，利用讀寫互斥在 writeLock.unlock() 前面卡 readLock.lock()，避免被其他 write thread 搶佔cpu更新變數，保證一次性的 atomic 操作

private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
private ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();

private volatile boolean isUpdate;

public void readWriteResource(){
    readLock.lock();
    if(isUpdate){
        readLock.unlock();
        writeLock.lock();
        //write the resource
        readLock.lock();
        writeLock.unlock();
    }
    //read the resource
    readLock.lock();
}

這裡需注意 isUpdate 需要設置 volatile 讓全部 thread 可見 flag 變化

結論

ReentrantLock 可作為 synchronized 的替代品，需使用公平鎖機制時使用
ReentrantReadWriteLock 用於讀取遠大於寫入的場景，

目前先寫到這，其他等以後想到其他再補充吧…

展開全文 >>