一個(gè)常見的說法是，線程可以做到 async/await 所能做的一切，且更簡單。那么，為什么大家選擇 async/await 呢？

Rust 是一種低級語言，它不會隱藏協(xié)程的復(fù)雜性。這與像 Go 這樣的語言相反，在 Go 中，異步是默認(rèn)發(fā)生的，程序員甚至不需要考慮它。

聰明的程序員試圖避免復(fù)雜性。因此，他們看到 async/await 中的額外復(fù)雜性，并質(zhì)疑為什么需要它。當(dāng)考慮到存在一個(gè)合理的替代方案——操作系統(tǒng)線程時(shí)，這個(gè)問題尤其相關(guān)。

讓我們通過 async 來進(jìn)行一次思維之旅吧。

背景概覽

通常，代碼是線性的；一件事情在另一件事情之后運(yùn)行。它看起來像這樣：

fn main() {
    foo();
    bar();
    baz();
}

很簡單，對吧？然而，有時(shí)你會想同時(shí)運(yùn)行很多事情。這方面的典型例子是 web 服務(wù)器?？紤]以下用線性代碼編寫的：

fn main() -> io::Result<()> {
    let socket = TcpListener::bind("0.0.0.0:80")?;
    loop {
        let (client, _) = socket.accept()?;
        handle_client(client)?;
    }
}

想象一下，如果 handle_client 需要幾毫秒，并且兩個(gè)客戶端同時(shí)嘗試連接到你的 web 服務(wù)器。你會遇到一個(gè)嚴(yán)重的問題！

• 客戶端 #1 連接到 web 服務(wù)器，并被 accept() 函數(shù)接受。它開始運(yùn)行 handle_client()。

• 客戶端 #2 連接到 web 服務(wù)器。然而，由于 accept() 當(dāng)前沒有運(yùn)行，我們必須等待 handle_client() 完成客戶端 #1 的運(yùn)行。

• 幾毫秒后，我們回到 accept()?？蛻舳?#2 可以連接。

現(xiàn)在想象一下，如果有兩百萬個(gè)同時(shí)客戶端。在隊(duì)列的末尾，你必須等待幾分鐘，web 服務(wù)器才能幫助你。它很快就會變得不可擴(kuò)展。

顯然，初期的 web 試圖解決這個(gè)問題。最初的解決方案是引入線程。通過將一些寄存器的值和程序的棧保存到內(nèi)存中，操作系統(tǒng)可以停止一個(gè)程序，用另一個(gè)程序替換它，然后再后繼續(xù)運(yùn)行那個(gè)程序。本質(zhì)上，它允許多個(gè)例程（或“線程”，或“進(jìn)程”）在同一個(gè) CPU 上運(yùn)行。

使用線程，我們可以將上述代碼重寫如下：

fn main() -> io::Result<()> {
    let socket = TcpListener::bind("0.0.0.0:80")?;
    loop {
        let (client, _) = socket.accept()?;
        thread::spawn(move || handle_client(client));
    }
}

現(xiàn)在，客戶端由一個(gè)與處理新連接等待不同的線程處理。太棒了！通過允許并發(fā)線程訪問，這避免了問題。

• 客戶端 #1 被服務(wù)器接受。服務(wù)器生成一個(gè)調(diào)用 handle_client 的線程。

• 客戶端 #2 嘗試連接到服務(wù)器。

• 最終，handle_client 在某處阻塞。操作系統(tǒng)保存處理客戶端 #1 的線程，并將主線程帶回來。

• 主線程接受客戶端 #2。它生成一個(gè)單獨(dú)的線程來處理客戶端 #2。在只有幾微秒的延遲后，客戶端 #1 和客戶端 #2 并行運(yùn)行。

線程在考慮到生產(chǎn)級 web 服務(wù)器擁有幾十個(gè) CPU 核心時(shí)特別好用。不僅僅是操作系統(tǒng)可以給人一種所有這些線程同時(shí)運(yùn)行的錯(cuò)覺；實(shí)際上，操作系統(tǒng)可以讓它們真正同時(shí)運(yùn)行。

最終，出于我稍后將詳細(xì)說明的原因，程序員希望將這種并發(fā)性從操作系統(tǒng)空間帶到用戶空間。用戶空間并發(fā)性有許多不同的模型。有事件驅(qū)動編程、actor 和協(xié)程。Rust 選擇的是 async/await。

簡單來說，你將程序編譯成一個(gè)狀態(tài)機(jī)的集合，這些狀態(tài)機(jī)可以獨(dú)立于彼此運(yùn)行。Rust 本身提供了一種創(chuàng)建狀態(tài)機(jī)的機(jī)制；async 和 await 的機(jī)制。使用 smol 編寫的上述程序?qū)⑷缦滤荆?/p>

#[apply(smol_macros::main!)]
async fn main(ex: &smol::Executor) -> io::Result<()> {
    let socket = TcpListener::bind("0.0.0.0:80").await?;
    loop {
        let (client, _) = socket.accept().await?;
        ex.spawn(async move {
            handle_client(client).await;
        }).detach();
    }
}

主函數(shù)前面有 async 關(guān)鍵字。這意味著它不是一個(gè)傳統(tǒng)函數(shù)，而是一個(gè)返回狀態(tài)機(jī)的函數(shù)。大致上，函數(shù)的內(nèi)容對應(yīng)于該狀態(tài)機(jī)。

await 包括另一個(gè)狀態(tài)機(jī)作為當(dāng)前運(yùn)行狀態(tài)機(jī)的一部分。對于 accept()，這意味著狀態(tài)機(jī)將把它作為一個(gè)步驟包含在內(nèi)。

最終，一個(gè)內(nèi)部函數(shù)將會產(chǎn)生結(jié)果，或者放棄控制。例如，當(dāng) accept() 等待新連接時(shí)。在這一點(diǎn)上，整個(gè)狀態(tài)機(jī)將把執(zhí)行權(quán)交給更高級別的執(zhí)行器。對我們來說，那是 smol::Executor。

一旦執(zhí)行被產(chǎn)生，執(zhí)行器將用另一個(gè)正在并發(fā)運(yùn)行的狀態(tài)機(jī)替換當(dāng)前狀態(tài)機(jī)，該狀態(tài)機(jī)是通過 spawn 函數(shù)生成的。

我們將一個(gè)異步塊傳遞給 spawn 函數(shù)。這個(gè)塊代表一個(gè)完全新的狀態(tài)機(jī)，獨(dú)立于由 main 函數(shù)創(chuàng)建的狀態(tài)機(jī)。這個(gè)狀態(tài)機(jī)所做的一切都是運(yùn)行 handle_client 函數(shù)。

一旦 main 產(chǎn)生結(jié)果，就選擇一個(gè)客戶端來代替它運(yùn)行。一旦那個(gè)客戶端產(chǎn)生結(jié)果，循環(huán)就會重復(fù)。

你現(xiàn)在可以處理數(shù)百萬的并發(fā)客戶端。

當(dāng)然，像這樣的用戶空間并發(fā)性引入了復(fù)雜性的提升。當(dāng)你使用線程時(shí)，你不必處理執(zhí)行器、任務(wù)和狀態(tài)機(jī)等。

如果你是一個(gè)理智的人，你可能會問：“我們?yōu)槭裁葱枰鏊羞@些事情？線程工作得很好；對于 99% 的程序，我們不需要涉及任何用戶空間并發(fā)性。引入新復(fù)雜性是技術(shù)債務(wù)，技術(shù)債務(wù)會花費(fèi)我們的時(shí)間和金錢。”

“那么，我們?yōu)槭裁床皇褂镁€程呢？”

超時(shí)問題

也許 Rust 最大的優(yōu)勢之一是可組合性。它提供了一組可以嵌套、構(gòu)建、組合和擴(kuò)展的抽象。

我記得讓我堅(jiān)持使用 Rust 的是 Iterator trait。它可以讓我將某個(gè)東西變成 Iterator，應(yīng)用一些不同的組合器，然后將結(jié)果 Iterator 傳遞給任何接受 Iterator 的函數(shù)，這讓我大開眼界。

它繼續(xù)給我留下深刻印象。假設(shè)你想從另一個(gè)線程接收一列表整數(shù)，只取那些立即可用的整數(shù)，丟棄任何不是偶數(shù)的整數(shù)，給它們?nèi)考右?，然后將它們推到一個(gè)新列表上。

在某些其他語言中，這將是五十行代碼和一個(gè)輔助函數(shù)。在 Rust 中，可以用五行完成：

let (send, recv) = mpsc::channel();
my_list.extend(
    recv.try_iter()
        .filter(|x| x & 1 == 0)
        .map(|x| x + 1)
);

async/await 最好的事情是，它允許你將這種可組合性應(yīng)用于 I/O 限制函數(shù)。假設(shè)你有一個(gè)新的客戶端要求；你想在上面的函數(shù)中添加一個(gè)超時(shí)。假設(shè)我們的 handle_client 函數(shù)看起來像這樣：

async fn handle_client(client: TcpStream) -> io::Result<()> {
    let mut data = vec![];
    client.read_to_end(&mut data).await?;

    let response = do_something_with_data(data).await?;
    client.write_all(&response).await?;
    Ok(())
}

如果我們想添加一個(gè)三秒鐘的超時(shí)，我們可以組合兩個(gè)組合器來做到這一點(diǎn)：

race 函數(shù)同時(shí)運(yùn)行兩個(gè) future。

Timer future 等待一段時(shí)間后返回。

最終的代碼看起來像這樣：

async fn handle_client(client: TcpStream) -> io::Result<()> {
    // 處理實(shí)際連接的 future
    let driver = async move {
        let mut data = vec![];
        client.read_to_end(&mut data).await?;

        let response = do_something_with_data(data).await?;
        client.write_all(&response).await?;
        Ok(())
    };
    // 處理等待超時(shí)的 future
    let timeout = async {
        Timer::after(Duration::from_secs(3)).await;
        // 我們剛剛超時(shí)了！返回一個(gè)錯(cuò)誤。
        Err(io::ErrorKind::TimedOut.into())
    };
    // 并行運(yùn)行兩者
    driver.race(timeout).await
}

我發(fā)現(xiàn)這是一個(gè)非常簡單的過程。你所要做的就是將你的現(xiàn)有代碼包裝在一個(gè)異步塊中，然后將其與另一個(gè) future 競速。

這種方法的額外好處是，它適用于任何類型的流。在這里，我們使用 TcpStream。然而，我們可以很容易地將其替換為任何實(shí)現(xiàn) impl AsyncRead + AsyncWrite 的東西。async 可以輕松地適應(yīng)你需要的任何模式。

用線程實(shí)現(xiàn)

如果我們想在我們的線程示例中實(shí)現(xiàn)這一點(diǎn)呢？

fn handle_client(client: TcpStream) -> io::Result<()> {
    let mut data = vec![];
    client.read_to_end(&mut data)?;
    let response = do_something_with_data(data)?;
    client.write_all(&response)?;
    Ok(())
}

這并不容易。通常，你不能在阻塞代碼中中斷 read 或 write 系統(tǒng)調(diào)用，除非做一些災(zāi)難性的事情，比如關(guān)閉文件描述符（在 Rust 中無法做到）。

幸運(yùn)的是，TcpStream 有兩個(gè)函數(shù) set_read_timeout 和 set_write_timeout，可以用來分別設(shè)置讀寫超時(shí)。然而，我們不能天真地使用它。想象一個(gè)客戶端每 2.9 秒發(fā)送一個(gè)字節(jié)，只是為了重置超時(shí)。

所以我們需要在這里稍微防御性地編程。由于 Rust 組合器的強(qiáng)大，我們可以編寫自己的類型，包裝 TcpStream 來編程超時(shí)。

// `TcpStream` 的截止日期感知包裝器
struct DeadlineStream {
    tcp: TcpStream,
    deadline: Instant,
}

impl DeadlineStream {
    // 創(chuàng)建一個(gè)新的 `DeadlineStream`，經(jīng)過一段時(shí)間后過期
    fn new(tcp: TcpStream, timeout: Duration) -> Self {
        Self {
            tcp,
            deadline: Instant::now() + timeout,
        }
    }
}

impl io::Read for DeadlineStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        // 設(shè)置截止日期
        let time_left = self.deadline.saturating_duration_since(Instant::now());
        self.tcp.set_read_timeout(Some(time_left))?;
        // 從流中讀取
        self.tcp.read(buf)
    }
}

impl io::Write for DeadlineStream {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        // 設(shè)置截止日期
        let time_left = self.deadline.saturating_duration_since(Instant::now());
        self.tcp.set_write_timeout(Some(time_left))?;
        // 從流中讀取
        self.tcp.write(buf)
    }
}

// 創(chuàng)建包裝器
let client = DeadlineStream::new(client, Duration::from_secs(3));
let mut data = vec![];
client.read_to_end(&mut data)?;
let response = do_something_with_data(data)?;
client.write_all(&response)?;
Ok(())

一方面，可以認(rèn)為這是優(yōu)雅的。我們使用 Rust 的能力用一個(gè)相對簡單的組合器解決了問題。我相信它會運(yùn)行得很好。

另一方面，這絕對是 hacky。

我們鎖定了自己使用 TcpStream。Rust 中沒有特質(zhì)來抽象使用 set_read_timeout 和 set_write_timeout 類型。所以如果要使用任何類型的寫入器，需要額外的工作。

這涉及到設(shè)置超時(shí)的額外系統(tǒng)調(diào)用。

我認(rèn)為這種類型對于 web 服務(wù)器要求的實(shí)際邏輯來說，使用起來要笨重得多。

異步成功案例

這就是為什么 HTTP 生態(tài)系統(tǒng)采用 async/await 作為其主要運(yùn)行機(jī)制的原因，即使是客戶端也是如此。你可以取任何進(jìn)行 HTTP 調(diào)用的函數(shù)，并使其適應(yīng)你想要的任何用例。

tower 可能是我能想到的這種現(xiàn)象最好的例子，這也是讓我意識到 async/await 可以有多強(qiáng)大的東西。如果你將你的服務(wù)實(shí)現(xiàn)為一個(gè)異步函數(shù)，你會得到超時(shí)、速率限制、負(fù)載均衡、對沖和背壓處理。所有這些都是無負(fù)擔(dān)實(shí)現(xiàn)的。

不管你使用的是什么運(yùn)行時(shí)，或者你的服務(wù)實(shí)際上在做什么。你可以將它扔給 tower，使其更加健壯。

macroquad 是一個(gè)小型 Rust 游戲引擎，旨在使游戲開發(fā)盡可能簡單。它的主函數(shù)使用 async/await 來運(yùn)行其引擎。這是因?yàn)?nbsp;async/await 確實(shí)是在 Rust 中表達(dá)需要停下來等待其他事情的線性函數(shù)的最佳方式。

在實(shí)踐中，這可能非常強(qiáng)大。想象一下，同時(shí)輪詢你的游戲服務(wù)器和你的 GUI 框架的網(wǎng)絡(luò)連接，在同一線程上。可能性是無限的。

提升異步的形象

我認(rèn)為問題不在于有人認(rèn)為線程比異步更好。我認(rèn)為問題是異步的好處沒有被廣泛傳播。這導(dǎo)致一些人對異步的好處有誤解。

如果這是一個(gè)教育問題，我認(rèn)為值得看一下教育材料。這是 Rust Async Book 在比較 async/await 和操作系統(tǒng)線程時(shí)所說的：

操作系統(tǒng)線程不需要對編程模型做任何改變，這使得并發(fā)表達(dá)非常容易。然而，線程間的同步可能會很困難，性能開銷也很大。線程池可以緩解這些成本，但不足以支持大規(guī)模的 I/O 密集型工作負(fù)載。

—— Rust Async Book

我認(rèn)為這是整個(gè)異步社區(qū)的一個(gè)一貫問題。當(dāng)有人問“為什么我們想用這個(gè)而不是操作系統(tǒng)線程”時(shí)，人們傾向于揮揮手說“異步開銷更小。除此之外，其他都一樣?！?/p>

這就是 web 服務(wù)器作者轉(zhuǎn)向 async/await 的原因。這就是他們?nèi)绾谓鉀Q C10k 問題的。但這不會是其他人轉(zhuǎn)向 async/await 的原因。

c10k 問題：https://en.wikipedia.org/wiki/C10k_problem

性能提升是不穩(wěn)定的，可能會在錯(cuò)誤的情況下消失。有很多情況下，線程工作流程可以比等效的異步工作流程更快（主要是在 CPU 密集型任務(wù)的情況下）?？赡芤郧拔覀冞^分強(qiáng)調(diào)了異步 Rust 的短暫性能優(yōu)勢，但低估了它的語義優(yōu)勢。

在最壞的情況下，這會導(dǎo)致人們對 async/await 置之不理，認(rèn)為它是“你為小眾用例而求助的奇怪事物”。它應(yīng)該被視為一個(gè)強(qiáng)大的編程模型，讓你能夠簡潔地表達(dá)在同步 Rust 中無法表達(dá)的模式，而不需要數(shù)十個(gè)線程和通道。

有一種趨勢是試圖使異步 Rust “就像同步 Rust 一樣”，這種方式鼓勵(lì)了負(fù)面比較。當(dāng)我說到“趨勢”時(shí)，我的意思是這是 Rust 項(xiàng)目的明確路線圖，即“編寫異步 Rust 代碼應(yīng)該像編寫同步代碼一樣容易，除了偶爾的 async 和 await 關(guān)鍵字?！?/p>

我拒絕這種框架，因?yàn)樗静豢赡堋＿@就像試圖在一個(gè)滑雪坡上舉辦披薩派對。我們不應(yīng)該試圖將我們的模型強(qiáng)行塞入不友好的慣用法，以迎合拒絕采用另一種模式的程序員。我們應(yīng)該努力突出 Rust 的 async/await 生態(tài)系統(tǒng)的優(yōu)勢；它的可組合性和它的能力。我們應(yīng)該努力使 async/await 成為程序員達(dá)到并發(fā)性時(shí)的默認(rèn)選擇。我們不應(yīng)該試圖使同步 Rust 和異步 Rust 相同，我們應(yīng)該接受差異。