TokioによるAsync Rust: async/await、タスク、チャネル

1. 概要

Rustにはasync/await構文が組み込まれていますが、Futureを実行するランタイムは付属していません。Tokioがこのための最も一般的なランタイムです。スケジューラ、タイマー、非同期I/O、同期プリミティブを提供します。

本記事は最初のasync関数から始まり、タスクのspawn、join!による並行実行、select!によるタイムアウト、タスク間の状態共有、channel経由でのデータ送信、ブロッキングコードの処理といった一般的な操作をカバーしています。最後にテストの書き方と頻出エラーについて説明します。

2. セットアップ

Cargo.tomlにTokioを追加してください：

[dependencies]
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

fullフィーチャはマルチスレッドランタイム、time、sync、net、macrosなどすべてのモジュールを有効にします。学習時には便利です。

本番プロジェクトでは、コンパイル時間を削減するために実際に使用するフィーチャだけを有効にするべきです。例えば["rt-multi-thread", "macros", "time", "sync"]。

test-utilフィーチャはテストで時間を制御できます。これはセクション8で使用します。

3. 最初のAsync関数

3.1. #[tokio::main]と.await

use std::time::Duration;
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // データベース呼び出しをシミュレート
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

説明：

• async fn fetch_orderはStringを直接返しません。impl Future<Output = String>を返します。
• tokio::time::sleepはsleepの非同期版です。待機中、スレッドはランタイムに返され、他の処理を実行できます。
• #[tokio::main]はマクロです。マルチスレッドランタイムを作成し、block_onでその中でmainを実行します。

Order #42

3.2. Futureは遅延評価

これはJavaScriptのPromiseとの大きな違いです。Rustのfutureは.awaitまたはpollされるまで何もしません。

async fn load_customer(id: u32) -> String {
    println!("顧客 {id} を読み込み中");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // 関数内のコードはまだ実行されていません
    println!("awaitの前");
    let customer = fut.await;
    println!("{customer}");
}

awaitの前
顧客 7 を読み込み中
Customer 7

「顧客 7 を読み込み中」という行は.awaitを呼んだ後にのみ表示されます。.awaitを忘れると、関数は実行されません。コンパイラはunused implementer of Future that must be usedという警告を出すので、この警告を無視しないでください。

4. spawn task

.awaitはfutureを順序立てて実行します。同時実行させたい場合は、tokio::spawnを使ってタスクを作成します。タスクはTokioが管理する軽い「green thread」です。

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("タスクがpanic");
        println!("{order}");
    }
    println!("合計時間: ~{} ms", start.elapsed().as_millis());
}

Order #1
Order #2
Order #3
合計時間: ~300 ms

順序立てて実行する場合、合計時間は100 + 200 + 300 = 600 msになります。spawnすると、3つのタスクが並行実行されるため、合計は約300 msです。

重要なポイント：

• tokio::spawnはJoinHandle<T>を返します。handleを.awaitするとResult<T, JoinError>が得られます。タスクがpanic或いはキャンセルされた場合はErrになります。
• 渡すfutureはSend + 'staticを満たす必要があります。タスクは別のスレッドで実行され、現在のスコープより長く存在する可能性があるため、async moveが必要です。
• タスクはspawnされた直後に実行開始します。handle を.awaitするまで待つ必要はありません。

注意： JoinHandleをdropしてもタスクはキャンセルされません。タスクは引き続きdetachedモードで実行されます。キャンセルするにはhandle.abort()を呼びます。

タスク数が動的に変わる場合はtokio::task::JoinSetを使えます。join_set.join_next().awaitで完了順にタスクの結果を取得できます。

5. 同じタスク内での並行実行: join!、select!、timeout

5.1. join!とtry_join!

常にspawnが必要というわけではありません。tokio::join!は複数のfutureを同じタスク内で同時にpollし、すべての完了を待ちます：

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

Order #1 - Customer 7 (200 ms)

合計時間は最も遅いfuture相当、約200 msです。futureは'staticである必要がないため、ローカル変数を借用できます。

関数がResultを返す場合はtokio::try_join!を使います。このマクロは最初のエラーで止まり、残りのfutureをdropします：

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

ここでfetch_order_checkedとfetch_customer_checkedはResult<String, AppError>を返すasync関数です。

5.2. timeout

use tokio::time::{timeout, Duration};

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("タイムアウト"),
}

タイムアウト

fetch_orderは200 msかかり、50 msの制限を超えます。その場合、timeoutはErr(Elapsed)を返し、内部のfutureはdrop（キャンセル）されます。

5.3. select!

select!は最初に完了する分岐を待ち、他の分岐をキャンセルします：

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("処理完了 {order}"),
        _ = shutdown_rx => println!("シャットダウン信号を受け取り、処理を停止します"),
    }
}

シャットダウン信号を受け取り、処理を停止します

シャットダウン信号は100 ms後に到着し、fetch_order（200 ms）より早いため、第2分岐が勝ちます。

注意： Rustでキャンセルするとは、最も近い.awaitポイントでfutureがdropされることを意味します。キャンセルされた分岐がアトミックでない操作の途中、例えばソケットから半分のメッセージを読み込んでいる場合、データが失われる可能性があります。ループ内のselect!にfutureを入れる前に、各関数のドキュメントの「cancel safety」セクションを確認してください。

6. タスク間でのデータ共有

6.1. Arc<Mutex<T>>

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("残在: {}", inventory.lock().unwrap()["laptop"]);
}

残在: 5

ここではstd::sync::Mutexを使用し、tokio::sync::Mutexではありません。critical sectionが短く内部に.awaitがない場合、これが推奨される選択肢です。標準のmutexの方が高速です。

.awaitポイントを超えてlockを保持する場合、例えばI/O待機中に共有接続を保持する場合にのみtokio::sync::Mutexが必要です。

6.2. Channel mpsc

メモリを共有する代わりに、channel経由でタスク間のメッセージを渡すことができます：

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Order>(32);

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // ループを終了させるために元のsenderを閉じます

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("受信 {order:?}");
        total += order.amount;
    }
    println!("合計: {total}");
}

受信 Order { id: 1, amount: 100 }
受信 Order { id: 3, amount: 300 }
受信 Order { id: 2, amount: 200 }
合計: 600

受信する順序は実行ごとに異なる場合があります。パラメータ32はバッファ容量です。バッファが満杯になると、sendは待機し、自然なバックプレッシャー機構を作成します。recv()はすべての senderがdropされたときにNoneを返します。drop(tx)を忘れると、プログラムはwhileループで無限に待機します。

mpscの他に、Tokioにはまた以下のものがあります：

• oneshot: 単一の値を送信する。
• broadcast: 複数のconsumer、各consumerがすべてのメッセージを受信する。
• watch: consumerが最新の値のみを関心を持つ。

7. ブロッキングコードの処理

Tokioは協調的スケジューリングを使用します。タスクは.awaitポイントでのみスレッドをyieldします。重いCPU操作や同期I/Oはworker threadを占有し、他のタスクをstarveさせます。

fn hash_password(password: &str) -> String {
    // CPUコストの高い操作をシミュレート、例えばbcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("spawn_blockingのタスクがpanic");
    println!("{hashed}");
}

hashed:s3cret

spawn_blockingはclosureを専用のブロッキング用スレッドプールに移し、runtime のworker threadは他のリクエストを処理するために自由にできます。

注意： async関数の中でstd::thread::sleep、std::fs::read、または同期HTTPクライアントを直接呼び出さないでください。async版（tokio::time::sleep、tokio::fs）を使用するか、spawn_blockingでラップしてください。大規模な並列計算にはRayonの方が適していることが多いです。

8. 非同期コードのテスト

マクロ#[tokio::test]は各テスト用にランタイムを作成します。デフォルトではシングルスレッドランタイムです。

use std::time::Duration;
use tokio::time::{sleep, timeout};

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

cargo test

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true（test-utilフィーチャが必要）はTokioの時間を止めます。すべてのタスクがタイマー待機中の場合、runtime は自動的に次のタイムポイントにジャンプします。そのおかげで、timeout関連のテストはほぼ即座に実行され、安定した結果が得られ、CI機のパフォーマンスに依存しません。

複数のスレッドの動作をテストする必要がある場合は、#[tokio::test(flavor = "multi_thread", worker_threads = 2)]を使用します。

9. よくあるエラー

.awaitを越えてstd::sync::MutexGuardを保持する。 このguardはSendではないため、futureもSendではなくなります。その場合、tokio::spawnはコンパイルエラーfuture cannot be sent between threads safelyを報告します。修正方法はguardのスコープを.awaitの前に{ ... }ブロックで縮小するか、tokio::sync::Mutexに切り替えることです。

ネストされたランタイムを作成する。 既にTokioで実行中のコード内でRuntime::new().block_on(...)を呼び出すと、Cannot start a runtime from within a runtimeというpanicが発生します。async文脈では単に.awaitを使用するだけです。

必要のない場合にasyncを使用する。 CLIツールがいくつかのファイルを読み込む場合、または純粋な計算のジョブはTokioから利益を得られません。追加のコンパイル時間と複雑さを払うだけです。asyncは多くの同時I/O操作がある場合に有効です：Webサーバー、クローラー、プロキシ、複数サービスを呼び出すクライアント。

無制限にspawnする。 タスクは安価ですが無料ではありません。100万要素のリストの各要素につき1つのタスクをspawnすると、downstreamのリソースを枯渇させる可能性があります。例えばデータベースの接続プール。tokio::sync::Semaphoreで並行レベルを制限するか、futurescrateのbuffer_unorderedを使用してください。

JoinErrorを無視する。 handleを.awaitしてlet _ =を使用した場合、タスク内のpanicは隠れます。エラーをログするか上流に伝播させるべきです。

10. まとめ

本記事はTokioを使ったAsync Rustの重要な部分を網羅しました：

• Futureは遅延評価で、.awaitされた時にのみ実行されます。
• tokio::spawnは並行実行するタスクを作成します。渡すfutureはSend + 'staticである必要があります。
• join!とtry_join!は同じタスク内で複数のfutureを実行します。select!とtimeoutは競争とキャンセルを処理します。
• 共有データはArc<Mutex<T>>で共有するか、mpscやoneshotなどのchannelで送信できます。
• ブロッキングコードはspawn_blockingにラップする必要があります。
• #[tokio::test(start_paused = true)]はタイミング関連のテストを高速で安定させます。

これらをマスターしたら、tokio::netでネットワークI/O、またはAxumなどのフレームワークを使ったTokioベースのWebサービス構築に進められます。