JA
PortuguêsFrançaisEspañolTiếng Việt简体中文한국어日本語EnglishDeutsch
ホーム / プログラミング言語 / Rust

RustのMutexとchannelを使った安全な並行処理

Cris The Dev · 2026/10/28 · 10 分で読めます · 47

1. 概要

Rustは自らの能力を「fearless concurrency」と呼んでいます。Compilerはプログラムが実行される前にビルド時にdata raceをブロックします。この記事では、標準ライブラリの2つの主要ツールを紹介します：stateを共有するためのMutex（Arcと共に）と、スレッド間でメッセージを送受信するためのmpsc channelです。この記事を読んだ後、各タイプの使い方、そのテストの書き方、deadlockやロック保持時間の延長などの一般的な落とし穴を回避する方法を理解するでしょう。

すべての例は標準ライブラリのみを使用し、外部の依存関係は必要ありません。cargo new orders-demoでプロジェクトを作成し、コードをsrc/main.rsに貼り付けることができます。

2. 問題：スレッド間でのデータ共有

複数のスレッドが注文を同時に処理し、総注文数を数えたいとします。素朴な書き方は次のようになります：

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

このコードはコンパイルできません：

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

Compilerは2つの理由を挙げています：

  • thread::spawnはclosureが'staticであることを要求します。子スレッドはmainより長く生存する可能性があるため、ローカル変数を借用することはできません。
  • moveを追加しても、各スレッドはtotal_ordersの独自のコピーを受け取ります。実際には誰も共通の変数に累積しません。

これを解決するには、2つのことが必要です：Arcを使ったownership共有とMutexを使ったアクセスの同期です。

3. Arc<Mutex<T>>でのstate共有

3.1. 複数のスレッドから注文数を数える

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

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("合計注文数: {}", *total_orders.lock().unwrap());
}
合計注文数: 4000

各部分の説明：

  • Arc::new(Mutex::new(0u32))：Arc（Atomic Reference Counted）は複数のスレッドが同じ値を所有することを許可します。Mutexは、任意の時点で1つのスレッドのみが内部の値にアクセスできることを保証します。
  • Arc::clone(&total_orders)は参照カウンタを増加させるだけで、データはコピーしません。各スレッドはmoveを通じて独自のArcを受け取ります。
  • counter.lock()はlockを取得するまで現在のスレッドをブロックします。関数はResult<MutexGuard<u32>, PoisonError<…>>を返します。
  • MutexGuardは&mut u32にderefするため、*... += 1は値を直接変更します。Guardはステートメントの最後でdropされ、lockはそこで自動的に解放されます。

Rustで「unlockを忘れる」ことはできません。また、データがMutexの内部にあるため、lockをスキップして値を読むこともできません。

注意： ArcをRcに置き換えることはできません。RcはSendを実装していないため、thread::spawnに渡すとcompilerはすぐにエラーを報告します。同様に、MutexをRefCellに置き換えることはできません。RefCellはSyncではないからです。

3.2. MutexGuardのスコープ

Lockはguardがdropされるまで保持されます。guardを変数に割り当てると、lockはその変数のスコープの終わりまで生存します：

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

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // guardはここでdropされ、lockは解放されます

    if reserved {
        // PaymentService呼び出しの遅延をシミュレート、lockを保持しません
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("顧客 {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}
顧客 0: true
顧客 1: true
顧客 2: false

行の順序は実行ごとに異なる場合があります。結果は常に2つのtrueと1つのfalseです。

ブロック{ ... }はlockの保持時間を最小限に制限します。遅い部分（支払い処理の呼び出し）はlockが解放された後に実行されるため、他のスレッドは待つ必要がありません。途中でより早く解放する必要がある場合は、明示的にdrop(stock)を呼び出すことができます。

3.3. Mutexのポイズニング

Lockを保持しているときにスレッドがpanicすると、RustはMutexをpoisonedとしてマークします。理由は、データが不完全な状態にある可能性があるからです：

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

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentServiceの途中エラー");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("残高: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutexはポイズニングされています、現在の残高: {}", *v);
        }
    }
}
thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentServiceの途中エラー
Mutexはポイズニングされています、現在の残高: 70

したがって、lock()の後の.unwrap()は実際には「別のスレッドがlockを保持中にpanicした場合はpanicせよ」という意味です。ほとんどのアプリケーションでは、これは合理的な選択です。データを回復する方法がわかっている場合、into_inner()はguardを返すので、チェックして修復することができます。

4. Channelを使ったメッセージの送受信

Mutexは「複数のスレッドが同じデータを修正する」という質問に答えます。Channelはその逆方向に進みます：各データはある時点では1つの所有者のみを持ち、ownershipはchannelを通じて転送されます。多くの問題では、このアプローチはより単純でエラーが少ないです。

4.1. 複数のproducer、単一のconsumer

std::sync::mpscはmultiple producer, single consumerの略です：

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

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

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

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("顧客-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // 重要：元のsenderをドロップする

    let mut revenue = 0;
    for order in rx {
        println!("注文 #{} を受け取りました（{}）", order.id, order.customer);
        revenue += order.amount;
    }
    println!("売上: {revenue}");
}
注文 #100 を受け取りました（顧客-1）
注文 #300 を受け取りました（顧客-3）
注文 #200 を受け取りました（顧客-2）
売上: 900000

説明：

  • mpsc::channel()は(Sender, Receiver)のペアを返します。各producerはtx.clone()を受け取ります。
  • tx.send(order)はorderをchannelにmoveします。この行の後、送信スレッドはorderを使用できなくなり、使おうとするとcompilerはエラーを報告します。これにより、2つのスレッドが同じOrderを修正することはありません。
  • for order in rxは、すべての Senderがdropされるまでループします。drop(tx)行がない場合、元のsenderはmain内で生存し続け、ループは永遠に待ちます。

受信の順序はスレッドのスケジューリングに依存します。総売上は常に900000です。

4.2. バッファ制限付きchannelとbackpressure

mpsc::channel()には無制限のバッファがあります。producerがconsumerより速い場合、メモリは徐々に膨張します。sync_channel(n)はバッファをn要素に制限し、バッファが満杯のときにsendはブロックします：

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

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("注文 {id} を送信しました");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("注文 {id} を処理しています");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}
注文 1 を送信しました
注文 2 を送信しました
注文 1 を処理しています
注文 3 を送信しました
注文 2 を処理しています
...

Producerは2つの注文を送信してからconsumerが取り出すのを待つ必要があります。これは自然なbackpressureメカニズムで、データ処理パイプラインに非常に有用です。sync_channel(0)では、各sendは誰かがrecvするまで待機する必要があります（rendezvous）。

5. Arcが不要な場合：scoped thread

スレッドが関数から戻る前に確実に終了する場合、thread::scopeはArcなしでデータを直接借用することを許可します：

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // scope内のすべてのスレッドがここでjoinされました

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}
[101, 102, 103, 104]

scopeにより、compilerはスレッドがprocessedとordersより長く生存しないことを知っています。したがって、&の借用は有効です。scope後、into_inner()はlockなしでMutexからデータを取得します。

6. 並行コードのテスト作成

テストを容易にするため、ロジックを純粋な関数に分離する必要があります。以下をsrc/lib.rsに配置します：

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

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

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

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}
cargo test
running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

テストempty_payments_returns_zeroは境界ケースをチェックします：drop(tx)を忘れた場合、このテストは0を返す代わりに永遠にハングアップします。これはエラーを検出する明確な兆候です。

count_ordersには小さなポイントがあります：値は戻る前に変数resultに割り当てられます。理由は、関数の最後の式の一時的なguardがtotalより長く生存する可能性があり、compilerはborrow errorを報告します。

7. 一般的なエラー

ロック順序が逆になるによるデッドロック。 スレッドAはordersをロックしてからcustomersをロックし、一方スレッドBはcustomersをロックしてからordersをロックします。両者は永遠に互いを待ちます。Rustはdata raceを防ぐだけで、デッドロックは防ぎません。固定されたロック順序を規約するか、2つのデータをMutex<(A, B)>にまとめる必要があります。

一時的なguardが予想より長く生存します。 次のコードを見てください：

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // lockはここでもまだ保持されています！
}

while let（およびmatch）の式の一時的なものは、ループの本体の終わりまで生存します。他のスレッドがqueueにプッシュしたい場合、processが実行されている間、ブロックされます。修正は分割することです：let next = queue.lock().unwrap().pop();してから処理します。

同じスレッドでlock()を再度呼び出します。 std::sync::Mutexはreentrantではありません。同じスレッドでlock()を2回呼び出すとdeadlockまたはpanicが発生します。

I/Oの待機中またはrecv()の間lockを保持します。 guardを保持している間のblocking操作はすべて、Mutexをボトルネックに変えます。必要なデータをコピーまたはcloneして外に出し、lockを解放してから遅い作業を行う必要があります。

注意： async code（Tokio）では、.awaitを超えてstd::sync::MutexGuardを保持しないでください。tokio::sync::MutexまたはTokioのchannelを使用してください。さらに、カウンタのような単純な整数の場合、AtomicU32はMutex<u32>よりはるかに高速です。

どれを選ぶか？

  • 複数のスレッドが同じstateを長期間読み書きする必要がある場合はMutexを使用します。例えば、cacheまたは在庫管理。読み取りが書き込みより明らかに多い場合は、RwLockを検討してください。
  • データがある場所から別の場所へと流れる場合はchannelを使用します。例えば、注文キュー、パイプライン、またはworker pool。Channelは各値が1つの所有者のみを持つことを保証するため、ロジックは推論しやすくなります。

8. 結論

Rustは一般的な並行処理エラーをコンパイルエラーに変えます。Arcがないとデータを共有できず、Mutexがないとデータを変更できず、channelを通じて送信された値を再利用することはできません。Arc<Mutex<T>>は共有stateに適切で、mpsc channelはスレッド間のownership転送に適切です。sync_channelはbackpressureを追加し、thread::scopeはスレッドのライフタイムが明確な場合にArcを削除できます。Compilerが保護できないのはdeadlock、guardが長く生存すること、およびsenderがdropされないことです。これらの場所を慎重にレビューし、テストを慎重に書く必要があります。

rustconcurrencymutexchannelthread
著者のアプリGlowmi · Lantern NightA cozy Mid-Autumn match-3 puzzleiOS · Android

同じトピックの記事

Rust · 2026/10/27 · 9 分で読めます · 53

RustのAxumでREST APIを構築する

Axum 0.8でCRUD REST APIを構築します。routing、extractor、shared state、IntoResponseを使ったエラーハンドリング、ネットワークポートを開かないhandlerのテストを扱います。

Rust · 2026/10/26 · 10 分で読めます · 49

Rustのスマートポインタ: Box、Rc、Arc

Rust での Box、Rc、Arc の使い分け: ヒープ割り当て、単一スレッド内での所有権共有と複数スレッド間での共有、Weak、RefCell、Mutex の組み合わせ