在Java中使用異步編程

作者：Yasin x 2021-03-22 08:45:30

異步是一種程序的運(yùn)行方式，各種編程語(yǔ)言語(yǔ)言或多或少都對(duì)它有所支持。異步對(duì)于Java后端程序員來說并不是一種特別熟悉的概念，而安卓或者前端的同學(xué)可能會(huì)對(duì)異步這個(gè)概念會(huì)更熟悉一些。

[[388836]]

本文轉(zhuǎn)載自微信公眾號(hào)「編了個(gè)程」，作者Yasin x。轉(zhuǎn)載本文請(qǐng)聯(lián)系編了個(gè)程公眾號(hào)。

最近在學(xué)習(xí)協(xié)程，打算輸出幾篇文章來介紹一下協(xié)程。而協(xié)程與異步有很大的關(guān)聯(lián)，所以想先介紹一下異步。

程序同步和異步

同步是最簡(jiǎn)單也是最符合我們?nèi)祟愃季S方式的編程方式，所謂同步，就是程序會(huì)按照代碼一行行執(zhí)行，執(zhí)行完一句再執(zhí)行下一句。

同步代碼看起來是這樣：

stepA(); 
stepB(); 
stepC(); 
...

stepA執(zhí)行完后，開始執(zhí)行stepB，stepB執(zhí)行完后，執(zhí)行stepC。

而有時(shí)候我們會(huì)有這樣的需求：在后臺(tái)執(zhí)行一段程序。具體到我們這個(gè)案例來說，就是執(zhí)行完stepA后，要開始執(zhí)行stepB，但不用等stepB執(zhí)行完，現(xiàn)在可以立即執(zhí)行stepC。

于是異步編程就出來了。在Java語(yǔ)言里，我們可以創(chuàng)建一個(gè)新的線程(或者使用線程池)去執(zhí)行異步任務(wù)：

stepA(); 
new Thread(() -> stepB()).start(); 
stepC();

這樣，stepB就在另一個(gè)線程里面“異步”執(zhí)行了，而stepC還是繼續(xù)在當(dāng)前線程里執(zhí)行。

異步有什么好處呢?

有一個(gè)顯而易見的好處：讓程序“響應(yīng)更快”。比如上述的case，如果stepB()任務(wù)比較耗時(shí)，比如發(fā)郵件操作。那使用同步的方式，程序需要等待卡在這里stepB完成才能往下走。而如果使用異步的方式，可以讓stepB“后臺(tái)”執(zhí)行，不影響當(dāng)前程序往下執(zhí)行。

這在UI程序中尤為重要，畢竟界面的響應(yīng)時(shí)間對(duì)用戶的體驗(yàn)很大。所以涉及到UI的語(yǔ)言、框架是最先研究和嘗試異步技術(shù)的。比如RxJava起源于安卓，Kotlin、Dart、JavaScript等語(yǔ)言也在UI程序中用得比較多。

而同樣的，對(duì)于IO密集型的程序，使用異步也能夠明顯提升性能，大家熟悉的nginx、redis、netty等，其底層都是利用的操作系統(tǒng)的系統(tǒng)調(diào)用(比如Linux的epoll)來實(shí)現(xiàn)異步，達(dá)到高性能的表現(xiàn)。

使用異步

在Java中使用異步一般是用多線程來實(shí)現(xiàn)的。

正如我們上文提到的，我們可以啟動(dòng)一個(gè)新的線程去“后臺(tái)”執(zhí)行一個(gè)異步任務(wù)。當(dāng)然，我們也可以把它扔進(jìn)線程池里。

// 新建線程執(zhí)行異步任務(wù) 
new Thread(() -> stepB()).start();

但如果我們要使用異步的返回結(jié)果怎么辦呢?比如常見的場(chǎng)景是請(qǐng)求另一個(gè)微服務(wù)的接口。

JDK 1.5提供了Callable和Future接口，用于實(shí)現(xiàn)“有返回值”的多線程任務(wù)。使用的時(shí)候一般是配合線程池使用：

public static void main(String[] args) throws Exception { 
    ExecutorService executor = Executors.newSingleThreadExecutor(); 
    Future<String> future = executor.submit(() -> { 
        // 模擬IO需要一秒 
        Thread.sleep(1000); 
        return "hello"; 
    }); 
    System.out.println("submitted"); 
    // 這里會(huì)阻塞直到future.get返回值或者超時(shí) 
    System.out.println(future.get(2, TimeUnit.SECONDS)); 
    executor.shutdown(); 
}

如果使用Future，我們?cè)谡{(diào)用future.get()方法的時(shí)候，會(huì)阻塞直到異步任務(wù)返回結(jié)果或者拋異常或者超時(shí)。試想一下我們有這個(gè)需求：任務(wù)B1需要任務(wù)B的結(jié)果，任務(wù)C1需要任務(wù)C的結(jié)果，但它們彼此是獨(dú)立的。如果使用Future我們得這樣做：

stepA(); 
Future futureB = executor.submit(() -> stepB()); 
Future futureC = executor.submit(() -> stepC()); 
 
stepB1(futureB.get()); 
// 這一步必須等stepB1執(zhí)行完 
stepC1(futureC.get());

所以使用future其實(shí)還是會(huì)在調(diào)用get方法的時(shí)候阻塞主流程。那有沒有什么辦法不阻塞呢?解決辦法是使用回調(diào)。

回調(diào)與回調(diào)地獄

所謂回調(diào)，在函數(shù)式編程語(yǔ)言中的說法就是，我傳一個(gè)函數(shù)進(jìn)去，等異步任務(wù)完成后，就執(zhí)行這個(gè)函數(shù)。Java雖然不是函數(shù)式編程語(yǔ)言，但Java8也支持函數(shù)式編程。

假設(shè)我們的需求僅僅是把一個(gè)異步任務(wù)產(chǎn)生的結(jié)果字符串打印出來，我們可以這樣寫：

public static void main(String[] args) throws Exception { 
    Consumer<String> callback = System.out::println; 
    new Thread(() -> { 
        // 模擬api調(diào)用，省略try-catch 
        Thread.sleep(1000); 
        // 假設(shè)這是調(diào)用第三方api返回的字符串 
        String s = "hello"; 
        callback.accept(s); 
    }).start(); 
    System.out.println("started"); 
}

甚至可以不用callback函數(shù)，直接在把程序代碼段放到異步任務(wù)里面：

public static void main(String[] args) throws Exception { 
    new Thread(() -> { 
        // 模擬api調(diào)用，省略try-catch 
        Thread.sleep(1000); 
        // 假設(shè)這是調(diào)用第三方api返回的字符串 
        String s = "hello"; 
        print(s); 
    }).start(); 
    System.out.println("started"); 
} 
 
private static void print(String str) { 
    System.out.println(str); 
}

那如果異步任務(wù)需要的回調(diào)太多呢?比如我們需要先異步請(qǐng)求接口A，拿到結(jié)果后再去異步請(qǐng)求接口B，拿到結(jié)果后再去異步請(qǐng)求接口C：

public static void main(String[] args) throws Exception { 
    new Thread(() -> { 
        String resultA = callAPI("input", "a"); 
        new Thread(() -> { 
            String resultB = callAPI(resultA, "b"); 
            new Thread(() -> { 
                String resultC = callAPI(resultB, "c"); 
                System.out.println(resultC); 
            }).start(); 
        }).start(); 
    }).start(); 
    System.out.println("started"); 
} 
 
private static String callAPI(String param, String mockRes) { 
    // 模擬api調(diào)用，省略try-catch 
    Thread.sleep(1000); 
    return mockRes; 
}

有沒有感覺這層層嵌套的代碼比較難看?這就是臭名昭著的“回調(diào)地獄”。

Java 8提供了一個(gè)叫CompletableFuture類來支持一些異步功能，包括回調(diào)。它支持「鏈?zhǔn)秸{(diào)用」，可以在一定程度上解決“回調(diào)地獄”的問題。上述代碼可以用CompletableFuture這樣寫:

public static void main(String[] args) throws Exception { 
    CompletableFuture.supplyAsync(() -> callAPI("input", "a")) 
        .thenApply(res -> callAPI(res, "b")) 
        .thenApply(res -> callAPI(res, "c")) 
        .thenAccept(System.out::println); 
    System.out.println("started"); 
    // 等異步任務(wù)輸出 
    Thread.sleep(20000); 
}

響應(yīng)式編程

是另一種異步解決方案。它的主要應(yīng)用場(chǎng)景是異步處理數(shù)據(jù)集合。對(duì)標(biāo)的是同步的Iterable。這里有一個(gè)對(duì)比圖：

比較典型的場(chǎng)景是UI產(chǎn)生的事件流(比如點(diǎn)擊事件等)。

響應(yīng)式編程的核心是“觀察者模式”。客戶端發(fā)送請(qǐng)求和，能夠立即得到一個(gè)Stream返回，客戶端訂閱這個(gè)Stream來接收通知。等服務(wù)端有數(shù)據(jù)時(shí)，就會(huì)往Stream上發(fā)布數(shù)據(jù)，客戶端就能夠收到數(shù)據(jù)了。

Spring 5也支持響應(yīng)式編程，并認(rèn)為它將是未來web編程的一大趨勢(shì)。響應(yīng)流 API java.util.concurrent.flow 已正式成為 Java 9 的一部分。但目前發(fā)展還比較緩慢，大家對(duì)這個(gè)東西的接受度一般，可能是因?yàn)榍袚Q成本比較高，且目前webmvc能夠滿足大多數(shù)需求吧。