Rustのチャネル（mpsc）

並行処理のページでは、スレッド同士が Arc<Mutex<T>> でデータにロックをかけて共有していました。それでもうまくいきますが、もっと簡単なことが多い別の方法もあります。データを共有する代わりに、スレッド同士でメッセージを送り合うのです。

チャネル（channel）は、両端のあるパイプのようなものだと考えてください。あるスレッドが送信側（sender）の端から値を入れ、別のスレッドが受信側（receiver）の端から同じ順番で値を取り出します。誰も何も共有しません。それぞれの値が、一方のスレッドからもう一方へムーブするだけです。Go のコミュニティには、Rust も借りている次のような言葉があります。「メモリを共有することで通信するな。通信することでメモリを共有せよ。」

はじめてのチャネル

標準ライブラリのチャネルは std::sync::mpsc にあります。この名前は multiple producer, single consumer（複数の生産者、1つの消費者）の略です。送信側はいくつあってもよく、受信側は1つだけです。

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

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

    thread::spawn(move || {
        let message = String::from("ワーカーからこんにちは");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main が受け取った: {received}");
}

実行結果:

main が受け取った: ワーカーからこんにちは

▶ 実際に試してみよう

一歩ずつ見ていきましょう。

• mpsc::channel() は Sender と Receiver のペアを返します。昔からの慣習で、tx（transmit、送信）と rx（receive、受信）と名付けます。
• move で tx を新しいスレッドに渡します。並行処理のページでスレッドにデータをムーブしたのと同じやり方です。
• tx.send(値) は値をチャネルに入れます。受信側がもういなくなっていると失敗するので、Result を返します。
• rx.recv() は値が届くまで待ちます。Err を返すのは、すべての送信側がいなくなり、しかもチャネルが空のときだけです。

join() が必要なかったことに気づきましたか？ recv() がすでにメッセージを待ってくれるので、main が早く終わりすぎることはありません。

たくさんの値を受け取る

Receiver はイテレータとして使えます。for ループで回すと、代わりに recv() を呼び続けてくれます。そして、すべての送信側が破棄されると、ループは自然に終わります。

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

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

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // スレッドが終わるここで tx が破棄される
    });

    for value in rx {
        println!("{value} を受け取った");
    }
    println!("チャネルが閉じました。すべて完了です");
}

実行結果:

10 を受け取った
20 を受け取った
30 を受け取った
チャネルが閉じました。すべて完了です

▶ 実際に試してみよう

ここでは送信側が1つしかないので、値は送られた順番のとおりに届きます。「送信側がいなくなるとループが止まる」というルールは大切です。次の節のために覚えておいてください。

複数の生産者：送信側を複製する

複数のスレッドから送れるようにするには、tx.clone() で送信側のコピーをそれぞれのスレッドに渡します。どのコピーも同じチャネルにつながっています。

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

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

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("ワーカー{id}から報告です")).unwrap();
        });
    }
    drop(tx); // 元の送信側を破棄する。そうしないと下のループが終わらない

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

実行結果:

ワーカー1から報告です
ワーカー2から報告です
ワーカー3から報告です

▶ 実際に試してみよう

注目してほしい点が2つあります。

• drop(tx) が大切です。 それぞれのスレッドは終わるときに自分のコピーを破棄しますが、main はまだ元の送信側を持っています。送信側が1つでも残っている限り、受信側は待ち続けるので、プログラムは永遠に止まったままになってしまいます。元の送信側を破棄することで、「main はもう何も送りません」と伝えるのです。
• スレッドはどの順番で終わるかわかりません。 メッセージは、ワーカー2、ワーカー1、ワーカー3 の順に届くこともあります。ここでは、表示がいつも同じになるように、集めてから並べ替えています。sort() を消して何度か実行すると、順番が変わるのがわかります。

永遠に待たない：try_recv と recv_timeout

recv() は何かが届くまでブロックします。待ちたくないときもありますよね。そんなときは、仲間のメソッドが2つあります。

• try_recv() はすぐに戻ります。結果は値か、Empty（まだ何もない）か、Disconnected（送信側がすべていなくなった）のどれかです。
• recv_timeout(時間) は待ちますが、制限時間までです。時間内に何も届かなければ Timeout になります。

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

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

    // まだ何も送られていないので、try_recv はすぐに戻る。
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // 値が待っているので、待たずにそれを返す。
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // 送信側はいなくなり、チャネルは空。
    println!("{:?}", rx.try_recv());
}

実行結果:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ 実際に試してみよう

受信する前にスレッドを join() しているので、値はすでに待っている状態になり、出力はいつも同じになります。

try_recv() は、ほかにもやることがあるループで便利です。たとえば、1フレームに1回新しいメッセージを確認するゲームループなどです。recv_timeout() は「ワーカーに時間がかかりすぎたらあきらめる」という場面に向いています。

sync_channel で上限のあるチャネルを作る

ふつうの channel() にはサイズの上限がありません。送信側が受信側よりずっと速いと、メッセージがメモリにどんどんたまっていきます。mpsc::sync_channel(n) は、最大 n 個までしかメッセージを保持しないチャネルを作ります。いっぱいになると、受信側が1つ取り出すまで send() が待ちます。これをバックプレッシャー（backpressure）と呼びます。遅い消費者が、速い生産者を自動的に減速させるのです。

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

fn main() {
    // 一度に入るメッセージは2つまで。
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // バッファがいっぱいのあいだはブロックする
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("合計 = {total}");
}

実行結果:

合計 = 15

▶ 実際に試してみよう

生産者が main より2つを超えて先に進むことはありません。sync_channel(0) にするとバッファがまったくなくなり、どの send() も受信側が値を受け取れる状態になるまで待ちます。荷物を手渡しするようなものです。

小さなワーカープール

では、ここまでの内容を組み合わせてみましょう。ワーカープールとは、共有されたキューから仕事を取り出し、結果を送り返す、いくつかのスレッドの集まりです。チャネルは2つ使います。1つは仕事用、もう1つは結果用です。

1つだけ注意点があります。Receiver は複製できません（消費者は1つだけだからです）。そこで、ワーカーたちは仕事の受信側を Arc<Mutex<_>> に入れて共有し、順番に取り出します。

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // 受信側は複製できないので、ワーカーたちは Arc<Mutex<_>> に入れた1つを共有する。
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // もう仕事はない
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // もう仕事はないとワーカーたちに伝える

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} の2乗は {square}");
    }
}

実行結果:

1 の2乗は 1
2 の2乗は 4
3 の2乗は 9
4 の2乗は 16
5 の2乗は 25
6 の2乗は 36

▶ 実際に試してみよう

きれいに終了する仕組みは次のとおりです。

1. drop(job_tx) で仕事用のチャネルが閉じます。すると各ワーカーの recv() が Err を返し、ワーカーはループを抜けます。
2. ワーカーが終わると、そのワーカーが持っていた result_tx のコピーが破棄されます。元の送信側はすでに破棄してあるので、3つのワーカーがすべて終わった時点で結果用のチャネルが閉じます。
3. すると result_rx.iter() が止まります。ワーカーはどの順番で終わるかわからないので、結果は並べ替えています。

細かい点ですが、job_rx.lock().unwrap().recv() をわざと1つの let 文で書いています。ロックはその文の終わりで解放されるので、このワーカーが数を2乗しているあいだに、ほかのワーカーが次の仕事を取りに行けるのです。

チャネルと Arc<Mutex<T>> の使い分け

どちらも安全です。向いている問題の形が違うだけです。

チャネルを使うとき	Arc<Mutex<T>> を使うとき
仕事が一方向に流れる。生産者が何かを作り、消費者がそれを処理する	たくさんのスレッドが同じ値を読んで更新する必要がある（カウンター、キャッシュなど）
仕事、イベント、ログの行などのキューがほしい	データが大きく、コピーしたりムーブしたりするのが無駄になる
それぞれの値の所有者が、一度に1人だけ	スレッドが見たいのは更新の流れではなく、今の状態

目安になるのはこうです。プログラムを「このスレッドたちが、あのスレッドに何かを伝える」と説明できるなら、チャネルを使いましょう。「このスレッドたちがみんなでこれを編集する」なら、Mutex を使いましょう。実際のプログラムでは、さきほどのワーカープールのように、両方を使うこともよくあります。

よくある間違い

送ったあとで値を使う。 send() は、関数に値を渡すときと同じように所有権を受け取ります。送ったあとの値は、受信側のものです。

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

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

    thread::spawn(move || {
        let message = String::from("hi");
        tx.send(message).unwrap();
        println!("sent {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:25
   |
 8 |         let message = String::from("hi");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("sent {message}");
   |                         ^^^^^^^ value borrowed here after move

エラーは「ムーブされた値 message を借用しています」と言っています（エラーメッセージの位置がずれないように、この例の文字列は英語のままにしています）。これは Rust があなたを守ってくれているのです。もし送ったあとも送信側が message を変更できたら、2つのスレッドが同じデータに触れることになってしまいます。送る前に表示するか、.clone() を送りましょう。

ループの中で、送信側をスレッドにムーブしてしまう。 最初のスレッドが tx を持っていってしまうので、2つ目のスレッドに渡すものが残りません。

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

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

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

このメッセージは「ループの前の回で、値がクロージャにムーブされています」と言っています。直し方は、複数の生産者の例と同じように、ループの先頭に let tx = tx.clone(); を書くことです。

いつまでも終わらないプログラム。 これはコンパイルエラーにならないので、気づきにくい間違いです。どこかの送信側（よくあるのは main にある元の tx）がまだ生きているのに受信側をループで回す（for x in rx）と、ループは永遠に待ち続けます。プログラムが「ただ止まったまま」になったら、すべての送信側が破棄されているか確認しましょう。

次のステップ

std::sync::mpsc でたいていのことはできますが、人気のある代わりの選択肢にも2つ出会うはずです。

• crossbeam-channel（英語）：複数の消費者に対応したチャネル（受信側を複製できる）と、複数のチャネルを一度に待てる select! マクロを提供します。
• tokio::sync::mpsc は、非同期のコードのためのチャネルです。async / .await を使っているなら（async/await を参照）、標準ライブラリのものではなくこちらを使いましょう。recv().await はスレッド全体をブロックしないからです。

まとめ

• mpsc::channel() は (Sender, Receiver) のペアを返します。send() で値を入れ、recv() で次の値を待ちます。
• 送信側を複製すれば、生産者をいくつでも持てます。受信側はいつも1つだけです。
• 受信側をループで回すと、すべての送信側が破棄されたときに終わります。元の送信側を drop() するのを忘れないようにしましょう。
• try_recv() と recv_timeout() を使うと、永遠に待たずに確認できます。sync_channel(n) はキューを n 個のメッセージに制限します。
• 複数のスレッドからのメッセージは、どんな順番でも届きます。安定した出力が必要なら、並べ替えたり合計したりしましょう。
• データがスレッドからスレッドへ流れるならチャネルを、スレッドたちが1つの値を共有して編集するなら Arc<Mutex<T>> を使いましょう。

理解度チェック

理解度チェック

1. "mpsc" は何の略ですか？

Many processes, shared cache（多数のプロセス、共有キャッシュ） Message passing, safe concurrency（メッセージパッシング、安全な並行性） Multiple producer, single consumer（複数の生産者、1つの消費者） Mutex protected shared channel（Mutex で守られた共有チャネル）

2. 3つのスレッドから同じチャネルに送れるようにするには、どうしますか？

それぞれのスレッドに tx.clone() を渡す 受信側を Rc で包む mpsc::channel() を3回呼んで、まとめる できない。チャネルの送信側は1つだけ

3. for value in rx のループがいつまでも終わりません。いちばん考えられる原因は何ですか？

チャネルがいっぱいになっている 送信側（よくあるのは main にある元の tx）が破棄されていない recv() を呼ぶのが早すぎた 値が Copy ではなかった

4. sync_channel(2) にすでにメッセージが2つ入っているとき、send() はどうなりますか？

いちばん古いメッセージを捨てる すぐにエラーを返す パニックする 受信側がメッセージを1つ取り出すまで待つ

スコア: 0 / 4

さらに詳しく

• メッセージ受け渡しを使ってスレッド間でデータを転送する — The Rust Programming Language 日本語版：チャネルの公式の入門です。さらに多くの例が載っています。
• std::sync::mpsc — 標準ライブラリのドキュメント（英語）：Sender、SyncSender、Receiver のすべてのメソッドが載っています。
• チャネル — Rust By Example 日本語版：複数の生産者スレッドを使った、短くて実行できる例です。
• シングルスレッドサーバをマルチスレッド化する — The Rust Programming Language 日本語版：チャネルと Arc<Mutex<Receiver>> を使って、本格的なスレッドプールを一歩ずつ作っていきます。
• tokio::sync::mpsc — docs.rs（英語）：tokio で使う非同期のチャネルと、標準ライブラリのチャネルとの違いがわかります。