Tokio 的基本用法

Tokio 是一個異步 I/O 框架，它提供了一種高效的方式來編寫異步代碼。它使用 Rust 語言的 Futures 庫來管理異步任務(wù)，并使用 Reactor 模式來處理 I/O 事件。

本系列 Tokio 篇將由淺入深的從基礎(chǔ)到實戰(zhàn)，以一個完整的 Rust 語言子系列講述網(wǎng)絡(luò)編程。

為什么要使用 Tokio？

在 Rust 中，使用異步編程可以提高程序的性能和響應(yīng)速度，但是異步編程往往需要編寫大量的樣板代碼和復(fù)雜的控制流程。Tokio 提供了一種簡單的方式來編寫異步代碼，它使用 Futures 庫來管理異步任務(wù)，并提供了一組工具來處理異步 I/O 事件。

如何使用 Tokio？

使用 Tokio 編寫異步代碼需要掌握以下幾個概念：

• ? Future：表示一個異步任務(wù)，可以理解為一個異步函數(shù)的返回值；
• ? Task：表示一個異步任務(wù)的執(zhí)行上下文，可以理解為一個異步函數(shù)的執(zhí)行環(huán)境；
• ? Reactor：表示一個 I/O 事件的處理器，可以理解為一個事件循環(huán)；
• ? Runtime：表示一個異步任務(wù)的執(zhí)行環(huán)境，可以理解為一個異步函數(shù)的運行時環(huán)境。

下面我們將使用 Tokio 編寫一個最基礎(chǔ)的服務(wù)器和客戶端程序，以便了解 Tokio 的基本用法。

編寫服務(wù)器

我們將編寫一個簡單的 PingPong 服務(wù)器，它接收客戶端的 Ping 請求，并返回 Pong 響應(yīng)。首先，我們需要創(chuàng)建一個異步任務(wù)來處理客戶端的請求。我們可以使用 Tokio 提供的async關(guān)鍵字來定義一個異步函數(shù)：

async fn handle_client(mut stream: TcpStream) - > Result< (), Box< dyn Error >> {
    // ...
}

這個異步函數(shù)接收一個TcpStream對象，表示一個客戶端連接。我們可以在函數(shù)內(nèi)部處理客戶端的請求，并返回一個Result對象表示異步任務(wù)的執(zhí)行結(jié)果。在處理客戶端請求之前，我們需要先向客戶端發(fā)送一個歡迎消息：

async fn handle_client(mut stream: TcpStream) - > Result< (), Box< dyn Error >> {
    println!("new client connected");

    let mut buf = [0; 1024];
    stream.write_all(b"Welcome to the PingPong server!n").await?;

    // ...
}

在發(fā)送歡迎消息之后，我們需要不斷地從客戶端讀取數(shù)據(jù)，并返回 Pong 響應(yīng)。我們可以使用一個無限循環(huán)來實現(xiàn)這個功能：

async fn handle_client(mut stream: TcpStream) - > Result< (), Box< dyn Error >> {
    println!("new client connected");

    let mut buf = [0; 1024];
    stream.write_all(b"Welcome to the PingPong server!n").await?;

    loop {
        let n = stream.read(&mut buf).await?;
        if n == 0 {
            break;
        }
        stream.write_all(b"Pongn").await?;
    }

    println!("client disconnected");
    Ok(())
}

在循環(huán)中，我們使用stream.read()方法從客戶端讀取數(shù)據(jù)，并使用stream.write_all()方法向客戶端發(fā)送 Pong 響應(yīng)。如果客戶端關(guān)閉了連接，我們就退出循環(huán)并返回Ok(())表示異步任務(wù)執(zhí)行成功。

現(xiàn)在我們已經(jīng)編寫了一個異步任務(wù)來處理客戶端請求，接下來我們需要創(chuàng)建一個 Reactor 來處理 I/O 事件。我們可以使用 Tokio 提供的TcpListener對象來監(jiān)聽客戶端連接：

#[tokio::main]
async fn main() - > Result< (), Box< dyn Error >> {
    let addr = "127.0.0.1:8080";
    let listener = TcpListener::bind(addr).await?;
    println!("listening on {}", addr);

    loop {
        let (stream, _) = listener.accept().await?;
        tokio::spawn(async move {
            if let Err(e) = handle_client(stream).await {
                eprintln!("error: {}", e);
            }
        });
    }
}

在main函數(shù)中，我們首先創(chuàng)建一個TcpListener對象來監(jiān)聽客戶端連接。然后我們使用一個無限循環(huán)來等待客戶端連接，并使用listener.accept()方法來接收客戶端連接。當(dāng)有新的客戶端連接時，我們就創(chuàng)建一個新的異步任務(wù)來處理客戶端請求，并使用tokio::spawn()方法將任務(wù)提交到 Reactor 中執(zhí)行。

現(xiàn)在我們已經(jīng)完成了一個最基礎(chǔ)的 PingPong 服務(wù)器，可以使用cargo run命令來運行程序，并使用 telnet 命令來測試服務(wù)器：

$ cargo run
    Finished dev [unoptimized + debuginfo] target(s) in 0.08s
     Running `target/debug/pingpong`
listening on 127.0.0.1:8080

$ telnet localhost 8080
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
Welcome to the PingPong server!
ping
Pong
ping
Pong
^]
telnet > quit
Connection closed.

編寫客戶端

現(xiàn)在我們已經(jīng)編寫了一個最基礎(chǔ)的 PingPong 服務(wù)器，接下來我們將編寫一個客戶端程序來連接服務(wù)器并發(fā)送 Ping 請求。首先，我們需要創(chuàng)建一個異步任務(wù)來連接服務(wù)器：

async fn connect() - > Result< (), Box< dyn Error >> {
    let addr = "127.0.0.1:8080";
    let mut stream = TcpStream::connect(addr).await?;
    println!("connected to {}", addr);

    // ...
}

這個異步任務(wù)使用TcpStream::connect()方法來連接服務(wù)器，并返回一個Result對象表示連接結(jié)果。在連接成功之后，我們可以向服務(wù)器發(fā)送一個 Ping 請求：

async fn connect() - > Result< (), Box< dyn Error >> {
    let addr = "127.0.0.1:8080";
    let mut stream = TcpStream::connect(addr).await?;
    println!("connected to {}", addr);

    stream.write_all(b"Pingn").await?;
    let mut buf = [0; 1024];
    let n = stream.read(&mut buf).await?;
    let pong = std::str::from_utf8(&buf[..n])?;
    println!("{}", pong);

    Ok(())
}

在發(fā)送 Ping 請求之后，我們使用stream.read()方法從服務(wù)器讀取響應(yīng)，并使用std::str::from_utf8()方法將響應(yīng)轉(zhuǎn)換為字符串。最后，我們將響應(yīng)打印到控制臺上，并返回Ok(())表示異步任務(wù)執(zhí)行成功。

現(xiàn)在我們已經(jīng)編寫了一個異步任務(wù)來連接服務(wù)器并發(fā)送 Ping 請求，接下來我們需要在main函數(shù)中啟動這個任務(wù)：

#[tokio::main]
async fn main() - > Result< (), Box< dyn Error >> {
    connect().await?;
    Ok(())
}

現(xiàn)在我們已經(jīng)完成了一個最基礎(chǔ)的 PingPong 客戶端，可以使用cargo run命令來運行程序，并查看控制臺輸出：

$ cargo run
    Finished dev [unoptimized + debuginfo] target(s) in 0.08s
     Running `target/debug/pingpong`
connected to 127.0.0.1:8080
Pong

完整代碼

最后，我們將完整的服務(wù)器和客戶端代碼放在一起，以便讀者參考：

use std::error::Error;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

async fn handle_client(mut stream: TcpStream) - > Result< (), Box< dyn Error >> {
    println!("new client connected");

    let mut buf = [0; 1024];
    stream.write_all(b"Welcome to the PingPong server!n").await?;

    loop {
        let n = stream.read(&mut buf).await?;
        if n == 0 {
            break;
        }
        stream.write_all(b"Pongn").await?;
    }

    println!("client disconnected");
    Ok(())
}

#[tokio::main]
async fn main() - > Result< (), Box< dyn Error >> {
    let addr = "127.0.0.1:8080";
    let listener = TcpListener::bind(addr).await?;
    println!("listening on {}", addr);

    loop {
        let (stream, _) = listener.accept().await?;
        tokio::spawn(async move {
            if let Err(e) = handle_client(stream).await {
                eprintln!("error: {}", e);
            }
        });
    }
}

async fn connect() - > Result< (), Box< dyn Error >> {
    let addr = "127.0.0.1:8080";
    let mut stream = TcpStream::connect(addr).await?;
    println!("connected to {}", addr);

    stream.write_all(b"Pingn").await?;
    let mut buf = [0; 1024];
    let n = stream.read(&mut buf).await?;
    let pong = std::str::from_utf8(&buf[..n])?;
    println!("{}", pong);

    Ok(())
}

#[tokio::main]
async fn main() - > Result< (), Box< dyn Error >> {
    connect().await?;
    Ok(())
}

總結(jié)

通過本文的介紹，我們了解了 Tokio 的基本用法，并編寫了一個最基礎(chǔ)的 PingPong 服務(wù)器和客戶端程序。Tokio 提供了一種簡單的方式來編寫異步代碼，可以幫助我們提高程序的性能和響應(yīng)速度。在實際開發(fā)中，我們可以根據(jù)需要使用 Tokio 提供的各種工具來編寫更加復(fù)雜的異步程序。