あっぽログ
← 記事一覧に戻る
RustTokio非同期処理

Rustのtokio::selectで複数の非同期処理を制御する：タイムアウトとキャンセルを実装する

はじめに

Rustの非同期プログラミングでは、複数の非同期処理を「同時に待ち、どれか一つが完了したら次へ進む」という場面がよくあります。たとえば「APIリクエストを送りつつ、3秒たったらタイムアウトする」といったケースです。

Tokioが提供する select! マクロはまさにこの用途のためにあります。本記事では tokio::select! の基本的な使い方から、タイムアウト・キャンセル処理の実践的なパターンまでを解説します。

環境のセットアップ

Cargo.toml に以下の依存関係を追加してください。

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

select!マクロの基本

select! マクロは複数の async ブランチを同時に待機し、最初に完了したブランチだけを実行します。残りのブランチはキャンセルされます。

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

#[tokio::main]
async fn main() {
    let task_a = sleep(Duration::from_secs(1));
    let task_b = sleep(Duration::from_secs(2));

    tokio::select! {
        _ = task_a => {
            println!("task_a が先に完了しました");
        }
        _ = task_b => {
            println!("task_b が先に完了しました");
        }
    }
}

この例では task_a（1秒）が先に完了するので、"task_a が先に完了しました" が出力されます。

ブランチの構文

各ブランチは以下の形式で記述します。

パターン = Futureの式 => { 実行する処理 }

_ を使うと戻り値を無視できます。戻り値が必要な場合は変数名を書きます。

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

async fn fetch_fast() -> &'static str {
    sleep(Duration::from_millis(100)).await;
    "高速レスポンス"
}

async fn fetch_slow() -> &'static str {
    sleep(Duration::from_secs(5)).await;
    "低速レスポンス"
}

#[tokio::main]
async fn main() {
    tokio::select! {
        result = fetch_fast() => {
            println!("結果: {}", result);
        }
        result = fetch_slow() => {
            println!("結果: {}", result);
        }
    }
}
// 出力: 結果: 高速レスポンス

タイムアウトを実装する

select! の典型的なユースケースがタイムアウトです。tokio::time::sleep を組み合わせると簡単に実装できます。

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

async fn heavy_task() -> String {
    // 重い処理（例：外部APIへのリクエスト）
    sleep(Duration::from_secs(10)).await;
    "処理完了".to_string()
}

#[tokio::main]
async fn main() {
    tokio::select! {
        result = heavy_task() => {
            println!("成功: {}", result);
        }
        _ = sleep(Duration::from_secs(3)) => {
            println!("タイムアウト: 3秒以内に完了しませんでした");
        }
    }
}
// 出力: タイムアウト: 3秒以内に完了しませんでした

なお、Tokioには tokio::time::timeout という便利な関数も用意されています。シンプルなタイムアウトであればこちらも活用できます。

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

#[tokio::main]
async fn main() {
    let result = timeout(Duration::from_secs(3), heavy_task()).await;

    match result {
        Ok(value) => println!("成功: {}", value),
        Err(_) => println!("タイムアウトしました"),
    }
}

キャンセル処理を実装する

select! を使うと、チャネルからのシグナルでタスクをキャンセルする仕組みも作れます。tokio::sync::oneshot を使った例を見てみましょう。

use tokio::sync::oneshot;
use tokio::time::{sleep, Duration};

async fn long_running_task() {
    println!("タスク開始...");
    for i in 1..=10 {
        sleep(Duration::from_secs(1)).await;
        println!("処理中: {} 秒経過", i);
    }
    println!("タスク完了");
}

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

    // 別タスクで3秒後にキャンセル信号を送る
    tokio::spawn(async move {
        sleep(Duration::from_secs(3)).await;
        let _ = cancel_tx.send(());
        println!("キャンセル信号を送信しました");
    });

    tokio::select! {
        _ = long_running_task() => {
            println!("タスクが正常に完了しました");
        }
        _ = cancel_rx => {
            println!("タスクがキャンセルされました");
        }
    }
}

oneshot チャネルは1度だけ値を送れる通信路です。cancel_tx.send(()) を呼ぶと cancel_rx が完了し、select! がキャンセルブランチを選びます。

ループの中でselect!を使う

サーバーのようにイベントを繰り返し処理したい場合、loop の中で select! を使うパターンが一般的です。

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

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

    // メッセージ送信タスク
    tokio::spawn(async move {
        for i in 1..=3 {
            sleep(Duration::from_millis(500)).await;
            tx.send(format!("メッセージ {}", i)).await.unwrap();
        }
    });

    loop {
        tokio::select! {
            Some(msg) = rx.recv() => {
                println!("受信: {}", msg);
            }
            _ = sleep(Duration::from_secs(2)) => {
                println!("2秒間メッセージがありませんでした。終了します。");
                break;
            }
        }
    }
}

このパターンでは「メッセージが来れば処理し、一定時間来なければタイムアウトして終了」という挙動を実現できます。

select!を使う際の注意点

Futureは再利用しない

select! はブランチが選ばれなかった Future をドロップ（破棄）します。同じ Future を複数回 select! に渡すには tokio::pin! で固定する必要があります。

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

#[tokio::main]
async fn main() {
    // pin!でFutureを固定することで再利用可能にする
    tokio::pin!(let my_sleep = sleep(Duration::from_secs(5)));

    for _ in 0..3 {
        tokio::select! {
            _ = &mut my_sleep => {
                println!("スリープ完了");
                break;
            }
            _ = sleep(Duration::from_secs(1)) => {
                println!("1秒経過...");
            }
        }
    }
}

まとめ

tokio::select! は複数の非同期処理を効率よく制御するための強力なツールです。

用途実装パターン
タイムアウトselect! + sleep
シンプルなタイムアウトtokio::time::timeout
キャンセルselect! + oneshot チャネル
イベントループloop + select! + mpsc

非同期処理は最初は難しく感じますが、select! のパターンを覚えると実用的なアプリケーションが書けるようになります。ぜひ手元で動かして試してみてください。

← 記事一覧に戻る