あっぽログ
← 記事一覧に戻る
Rustスレッド並行処理mpscTokio

RustのChannelで非同期メッセージパッシングを理解する：mpscで安全なスレッド間通信を実装する

チャンネルとは何か？

並行処理を書くとき、複数のスレッドが安全にデータをやり取りする手段が必要になります。Rustにはそのための仕組みとして**チャンネル（Channel）**が用意されています。

チャンネルはざっくり言うと「パイプ」のようなものです。一方の端からデータを送り（Sender）、もう一方の端で受け取る（Receiver）という構造になっています。

Rustの標準ライブラリには std::sync::mpsc モジュールが用意されており、mpsc は Multiple Producer, Single Consumer（複数の送信者、1つの受信者）の略です。

mpscチャンネルの基本的な使い方

まずはシンプルなサンプルから見ていきましょう。

use std::sync::mpsc;
use std::thread;

fn main() {
    // チャンネルを作成する
    // tx: 送信者（Sender）、rx: 受信者（Receiver）
    let (tx, rx) = mpsc::channel();

    // 別スレッドでメッセージを送信する
    thread::spawn(move || {
        let message = String::from("こんにちは、メインスレッド！");
        tx.send(message).unwrap();
        println!("メッセージを送信しました");
    });

    // メインスレッドでメッセージを受信する
    let received = rx.recv().unwrap();
    println!("受信したメッセージ: {}", received);
}

mpsc::channel() を呼ぶとタプル (Sender<T>, Receiver<T>) が返ってきます。tx を move でスレッドに渡し、rx でブロックしながら受信します。

recv() と try_recv() の違い

受信には2種類のメソッドがあります。

メソッド挙動
recv()メッセージが届くまでブロック（待機）する
try_recv()すぐに返る。届いていなければ Err を返す
use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        thread::sleep(Duration::from_millis(500));
        tx.send(42).unwrap();
    });

    loop {
        match rx.try_recv() {
            Ok(value) => {
                println!("受信: {}", value);
                break;
            }
            Err(_) => {
                println!("まだ届いていません…待機中");
                thread::sleep(Duration::from_millis(100));
            }
        }
    }
}

try_recv() を使うと受信待ちの間に別の処理を挟めるため、UIスレッドやイベントループとの相性が良いです。

複数の送信者から送る

mpsc の「Multiple Producer」の特性を活かして、複数のスレッドから1つの受信者へメッセージを送ることができます。Sender は clone() で複製できます。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for i in 0..3 {
        // txをクローンして各スレッドに渡す
        let tx_clone = tx.clone();
        thread::spawn(move || {
            let msg = format!("スレッド {} からのメッセージ", i);
            tx_clone.send(msg).unwrap();
        });
    }

    // 元のtxをドロップして、すべてのSenderが消えたらチャンネルが閉じる
    drop(tx);

    // イテレータとして受信する
    for received in rx {
        println!("受信: {}", received);
    }
}

rx はイテレータを実装しており、すべての Sender がドロップされるとループが自然に終了します。drop(tx) を忘れるとループが永遠に待ち続けるので注意しましょう。

Tokioの非同期チャンネルを使う

std::sync::mpsc は同期的なチャンネルです。非同期処理（async/await）の文脈では Tokio が提供する非同期チャンネルを使うのが一般的です。

# Cargo.toml
[dependencies]
tokio = { version = "1", features = ["full"] }

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // バッファサイズを指定する（bounded channel）
    let (tx, mut rx) = mpsc::channel::<String>(32);

    tokio::spawn(async move {
        for i in 0..5 {
            let msg = format!("非同期メッセージ {}", i);
            // .await が必要（バッファが満杯のとき待機する）
            tx.send(msg).await.unwrap();
        }
    });

    while let Some(message) = rx.recv().await {
        println!("受信: {}", message);
    }
}

Tokioのチャンネルの種類

Tokioにはいくつかの種類のチャンネルが用意されています。

チャンネル特徴
tokio::sync::mpsc複数送信者・1受信者（最も一般的）
tokio::sync::oneshot1度だけ送受信する（レスポンス返却に便利）
tokio::sync::broadcast1送信者・複数受信者（イベントブロードキャスト）
tokio::sync::watch最新値のみ保持・購読型

よくあるミスと注意点

Senderをムーブし忘れる

// NG: tx がスコープ外になる前にドロップされない
let (tx, rx) = mpsc::channel::<i32>();
thread::spawn(|| {
    tx.send(1).unwrap(); // コンパイルエラー：txがmoveされていない
});

move キーワードを必ず付けて、tx の所有権をスレッドに渡しましょう。

バッファサイズの設計

Tokioの mpsc::channel はバッファサイズを指定します。送信側が速く受信側が遅い場合、バッファが満杯になると send().await がブロックします。適切なバッファサイズを設計することがパフォーマンスに直結します。

まとめ

  • std::sync::mpsc は標準ライブラリの同期チャンネル。複数送信者・1受信者に対応
  • recv() はブロッキング、try_recv() はノンブロッキングで使い分ける
  • Sender は clone() で複数スレッドに渡せる
  • 非同期処理には Tokio の tokio::sync::mpsc を使う
  • Tokioには用途別に oneshot・broadcast・watch も用意されている

チャンネルを使うことで、共有メモリやMutexを使わずにスレッド間通信を安全に実装できます。Rustの「メッセージパッシングで共有するな、共有するならメッセージパッシングを使え」という設計哲学を体感できる強力な機能です。ぜひ実際に書いて動かしてみてください。

← 記事一覧に戻る