Tokio — Rust用の非同期ランタイム

Rustの非同期ランタイムである Tokio のエキスパートです。Rust非同期エコシステムの大部分を支える Tokio により、Tokioのタスクスケジューラー、非同期I/Oプリミティブ、チャネル、タイマー、同期ユーティリティを使用して、高性能なネットワークアプリケーション、並行サービス、I/Oバウンドシステムをビルドする開発者をサポートします。最小限のメモリオーバーヘッドで数百万の並行接続を処理できます。

コア機能

非同期タスク

use tokio::time::{sleep, Duration};
use tokio::task;

#[tokio::main]
async fn main() {
    // Spawn concurrent tasks
    let handle1 = task::spawn(async {
        sleep(Duration::from_secs(1)).await;
        "Task 1 done"
    });

    let handle2 = task::spawn(async {
        sleep(Duration::from_millis(500)).await;
        "Task 2 done"
    });

    // Await both
    let (r1, r2) = tokio::join!(handle1, handle2);
    println!("{}, {}", r1.unwrap(), r2.unwrap());

    // Select first to complete
    tokio::select! {
        val = async { sleep(Duration::from_secs(1)).await; "slow" } => {
            println!("Got: {val}");
        }
        val = async { sleep(Duration::from_millis(100)).await; "fast" } => {
            println!("Got: {val}");
        }
    }

    // Spawn blocking work (CPU-intensive) on dedicated thread pool
    let result = task::spawn_blocking(|| {
        heavy_computation()                // Won't block async runtime
    }).await.unwrap();
}

チャネル

use tokio::sync::{mpsc, broadcast, oneshot, watch};

// mpsc: Multiple producers, single consumer
let (tx, mut rx) = mpsc::channel::<String>(100);  // Buffer size 100
let tx2 = tx.clone();

task::spawn(async move { tx.send("hello".into()).await.unwrap(); });
task::spawn(async move { tx2.send("world".into()).await.unwrap(); });

while let Some(msg) = rx.recv().await {
    println!("Got: {msg}");
}

// broadcast: Multiple producers, multiple consumers
let (tx, _) = broadcast::channel::<String>(100);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();
tx.send("event".into()).unwrap();
// Both rx1 and rx2 receive "event"

// oneshot: Single value, single use (request-response)
let (tx, rx) = oneshot::channel::<String>();
tx.send("response".into()).unwrap();
let val = rx.await.unwrap();

// watch: Single value that can be updated and observed
let (tx, mut rx) = watch::channel("initial".to_string());
tx.send("updated".into()).unwrap();
rx.changed().await.unwrap();

TCPサーバー

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("0.0.0.0:8080").await?;

    loop {
        let (mut socket, addr) = listener.accept().await?;
        println!("New connection from {addr}");

        tokio::spawn(async move {
            let mut buf = vec![0u8; 1024];
            loop {
                let n = match socket.read(&mut buf).await {
                    Ok(0) => return,       // Connection closed
                    Ok(n) => n,
                    Err(e) => { eprintln!("Read error: {e}"); return; }
                };
                if socket.write_all(&buf[..n]).await.is_err() {
                    return;                // Write error
                }
            }
        });
    }
}

同期プリミティブ

use tokio::sync::{Mutex, RwLock, Semaphore};
use std::sync::Arc;

// Async Mutex (yields while waiting, doesn't block thread)
let data = Arc::new(Mutex::new(vec![1, 2, 3]));
let data_clone = data.clone();
tokio::spawn(async move {
    let mut lock = data_clone.lock().await;
    lock.push(4);
});

// RwLock: Multiple readers OR single writer
let cache = Arc::new(RwLock::new(HashMap::new()));
let read = cache.read().await;             // Non-exclusive
let mut write = cache.write().await;       // Exclusive

// Semaphore: Limit concurrency
let semaphore = Arc::new(Semaphore::new(10));  // Max 10 concurrent
let permit = semaphore.acquire().await.unwrap();
// ... do work ...
drop(permit);                              // Release

インストール

[dependencies]
tokio = { version = "1", features = ["full"] }
# または選別して: features = ["rt-multi-thread", "macros", "net", "io-util", "time", "sync"]

ベストプラクティス

1. ランタイムをブロックしない — CPU集約的または同期I/Oには spawn_blocking を使用します。非同期スレッドをブロックすると、他のタスクが飢餓状態になります
2. 通信にはチャネルを使用 — ワークキューには mpsc、イベントには broadcast、設定更新には watch を使用します
3. レースには Select を使用 — tokio::select! は最初に完了したフューチャーを選びます。タイムアウトとキャンセルに最適です
4. 非同期Mutex vs 標準Mutex — .await の間ロックを保持する場合は tokio::sync::Mutex を使用します。短い同期のみのロックには標準 Mutex を使用します
5. バックプレッシャーにはセマフォを使用 — 並行データベースクエリ、HTTPリクエスト、ファイル操作を制限します
6. グレースフルシャットダウン — tokio::signal::ctrl_c() とキャンセルトークンを使用して、終了前に作業をドレインします
7. ランタイム設定 — デフォルトは #[tokio::main] を使用します。カスタムスレッド数とスタックサイズは runtime::Builder を使用します
8. Tracingの統合 — Tokio with tracing クレートを使用します。スパンは非同期タスク境界を越えて自動的に伝播します