Rustにおける並行プログラミング:スレッド、メッセージング、および共有状態
並行性とは、プログラムが複数の処理を「同時に」実行する能力のことです。Rustは、所有権システムと型システムを通じてコンパイル時にデータ競合を排除するため、効率的かつ安全な並行コードを記述することができます。
シングルスレッドのプログラムを「最初から最後まで1人でキッチンで作業する」ことに例えるなら、マルチスレッドは「複数のシェフが同時に作業する」ことに例えられます。ある人は野菜を切り、ある人は炒め物を作り、またある人は料理を盛り付けます。しかし、キッチンに人が多すぎると、すぐに混乱が生じやすくなります。2人が同時に同じ包丁に手を伸ばしたり、ある人が他の人が現在使っている食材を手に取ったりしてしまうかもしれません。Rustの並行処理モデルは、「ルールのある複数シェフのキッチン」のようなものです。各シェフには専用の道具があり、食材は専用の経路を通じて渡され、共有の調味料は一度に1人しか使用できません。
1. 学習内容
std::thread::spawnスレッドを作成し、joinそのスレッドが終了するのを待つmoveクロージャは所有権をスレッドに移譲するmpsc::channelチャネル—複数プロデューサー、単一コンシューマーのメッセージングMutex<T>排他ロック――共有データへのスレッドセーフなアクセスを確保するArc<T>アトミック参照カウント —Mutex<T>を複数のスレッド間で共有できるようにするSend/Sync特性の概念――Rustにおける並行処理の安全性の基盤
2. 共同文書編集システムの物語
(1) 課題:並行制御のないオンラインドキュメント
ルナの会社は、オンラインの共同文書編集ツールを開発しています。初期のバージョンでは、一度に1人しか編集できなかったため、チームメンバーから絶えず不満の声が上がっていました:
- アリスが提案書を作成している間、ボブも同じ箇所のテキストを編集していたため、二人の編集内容が互いに上書きされてしまった。
- キャロルはドキュメントの更新状況をリアルタイムで確認したいのですが、そのたびにページを更新しなければなりません。
- デイブが変更を提出したが、ルナは誰がそれを行ったのか全く見当がつかなかった。
- さらに悪いことに、2人の編集者が同時に同じコンテンツを編集したため、ドキュメントがクラッシュしてしまった。
- Lunaが「排他編集」機能を追加しようとすると、そのたびにコードがごちゃごちゃになり、グローバル変数やロックだらけになってしまう。
「エディターごとに個別のスレッドを立ち上げ、変更依頼をチャネル経由でやり取りし、ドキュメントの内容をミューテックスで保護できれば素晴らしいですね……」
(2) Rustの並行処理モデルへのアプローチ
use std::thread;
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::time::Duration;
fn main() {
// Create a document with shared content(Arc<Mutex<String>>)
let document = Arc::new(Mutex::new(String::from("# Collaborative Documents\n\n")));
// Create a channel,Used to transmit edit requests
let (tx, rx) = mpsc::channel::<String>();
// Receiving Thread:Processing editing requests on an ongoing basis
let doc_for_receiver = Arc::clone(&document);
let receiver = thread::spawn(move || {
for edit in rx {
let mut doc = doc_for_receiver.lock().unwrap();
doc.push_str(&edit);
doc.push('\n');
println!("[Receiver] Edits Applied");
}
});
// Simulating an Editor Thread
let tx1 = tx.clone();
thread::spawn(move || {
tx1.send("- Alice: Added the content for Chapter 1".to_string()).unwrap();
});
let tx2 = tx.clone();
thread::spawn(move || {
tx2.send("- Bob: The title of Chapter 2 has been revised.".to_string()).unwrap();
});
// The main thread also sends a message
tx.send("- Luna: Format Adjustments".to_string()).unwrap();
// Wait until all editors have finished sending their edits
thread::sleep(Duration::from_millis(100));
drop(tx); // Close the sender
receiver.join().unwrap();
// Final Document Content
let final_doc = document.lock().unwrap();
println!("\n=== Final Document ===");
println!("{}", *final_doc);
}
Rustの並行処理ソリューションでは、責任範囲が明確に分離されています。各エディタは個別のスレッドで実行され、編集リクエストは
mpsc::channelを介して渡され、ドキュメントの内容はMutexによって保護されています。Arcにより、Mutexを複数のスレッド間で共有することが可能になります。所有権システムにより、レースコンディションが発生しないことが保証されています。
3. 基本概念
(1) Rust並行処理フレームワーク
graph TB
A[Rust Concurrent Programming] --> B[Thread Management]
A --> C[Messaging]
A --> D[Shared Status]
A --> E[Safety Guarantee]
B --> B1["thread::spawn ||"]
B --> B2["join() The Wait Is Over"]
B --> B3["move Closures Transfer Ownership"]
C --> C1["mpsc::channel"]
C --> C2["Sender / Receiver"]
C --> C3["send / recv"]
D --> D1["Mutex<T> Mutex"]
D --> D2["lock() Acquire a lock"]
D --> D3["Arc<T> Atomic Reference Counting"]
E --> E1["Send: Ownership can be transferred across threads"]
E --> E2["Sync: References can be shared across threads"]
E --> E3["Eliminating Data Races at Compile Time"]
(2) 3つの並行処理モデルの比較
| 機能 | thread::spawn (スレッド) | mpsc::channel (チャネル) | Arc<Mutex<T>> (共有状態) |
|---|---|---|---|
| 基本概念 | 独立実行ユニット | メッセージパッシング | 共有メモリ+ロック |
| データ転送方法 | move: クロージャを介して所有権を譲渡する | send/recv: メッセージを送信する | lock: 内部値を取得する |
| ユースケース | 独立したタスクの並列実行 | プロデューサー・コンシューマーパターン | 複数のスレッドによる同一データへのアクセス |
| 利点 | マルチコアCPUを最大限に活用する | 送信側と受信側を分離する | あらゆる型のデータを直接共有する |
| デメリット | スレッド間の通信が複雑 | 頻繁な少量のデータ転送には不向き | デッドロックの発生やパフォーマンスのオーバーヘッドが生じる可能性がある |
| Rustの機能 | 所有権によってダングリングポインタを防止 | コンパイラが正しい使用を保証 | コンパイル時にデータ競合を防止 |
(3) 送信および同期の特性
| 特性 | 意味 | 自動的に実装される条件 | 自動的に実装されない型 |
|---|---|---|---|
| 送信 | 型の所有権はスレッド間で譲渡可能 | ほとんどの型で自動的に実装されている | Rc<T> (非アトミックな参照カウント) |
| 同期 | 型への参照はスレッド間で共有可能 | ほとんどの型で自動的に実装される | RefCell<T> (非アトミックな内部可変性) |
| T: 送信 + 同期 | スレッド間で安全に共有・受け渡しできる型 | Arc<T>, Mutex<T> |
生のポインタ *const T, *mut T |
(4) スレッド間通信方式の選択に関するクイックリファレンス
| 通信方式 | タイプ | データの流れの方向 | ロックが必要か | 適用可能なシナリオ |
|---|---|---|---|---|
| チャネル | mpsc::channel |
単方向(送信者→受信者) | いいえ | プロデューサー・コンシューマーパターン |
| チャンネル(マルチプロデューサー) | mpsc::channel + tx.clone() |
マルチ→シングル | いいえ | マルチスレッドの集計結果 |
| 共有状態 | Arc<Mutex<T>> |
双方向 | はい(排他ロック) | 同一データへのマルチスレッドによる読み書きアクセス |
| ステータスの共有 (読み取り/書き込み) | Arc<RwLock<T>> |
双方向 | はい (読み取り/書き込みロック) | 読み取りが主体で書き込みが少ないシナリオ |
| アトミック操作 | AtomicUsize ら |
双方向 | いいえ(ハードウェアレベルでは) | 単純なカウンタ/フラグ |
| バリア | Barrier |
同期ポイント | なし | マルチスレッド収束点 |
Rustはスレッドセーフ性を確保するために実行時のチェックに依存していません。その代わりに、2つのマーカートレイトである
SendとSyncを用いてコンパイル時のチェックを行います。ある型がSendでない場合、その型を別のスレッドに渡そうとすると、コンパイル時にエラーが発生します。
4. 並行プログラミングの例
(1) ▶ サンプル:thread::spawn + join — スレッドの作成と待機 (難易度 ⭐)
// ============================================
// Demo:thread::spawn Create a Thread、join The Wait Is Over
// Simulation: Several chefs are preparing different dishes at the same time in the kitchen.
// ============================================
use std::thread;
use std::time::Duration;
fn main() {
println!("=== Kitchen Renovation Begins ===");
// --- 1. Create three threads to perform different tasks ---
let chef1 = thread::spawn(|| {
for i in 1..=3 {
println!("[Chef A] Chopping vegetables... Cut #{}", i);
thread::sleep(Duration::from_millis(50));
}
"A Finished chopping the vegetables"
});
let chef2 = thread::spawn(|| {
for i in 1..=3 {
println!("[Chef B] Stir-frying... Step #{}", i);
thread::sleep(Duration::from_millis(40));
}
"B Finished cooking the stir-fry"
});
// The main thread is also running
for i in 1..=3 {
println!("[Head Chef] Plating in progress... Item #{}", i);
thread::sleep(Duration::from_millis(60));
}
// --- 2. join Wait for all threads to finish and retrieve the return values ---
let result1 = chef1.join().unwrap();
let result2 = chef2.join().unwrap();
println!("\n=== Kitchen Shutdown ===");
println!("Chef A: {}", result1);
println!("Chef B: {}", result2);
println!("All work has been completed!");
}
出力:
=== Kitchen Renovation Begins ===
[Chef A] Chopping vegetables... Cut #1
[Chef B] Stir-frying... Step #1
[Head Chef] Plating in progress... Item #1
[Chef A] Chopping vegetables... Cut #2
[Chef B] Stir-frying... Step #2
[Head Chef] Plating in progress... Item #2
[Chef A] Chopping vegetables... Cut #3
[Chef B] Stir-frying... Step #3
[Head Chef] Plating in progress... Item #3
=== Kitchen Shutdown ===
Chef A: A Finished chopping the vegetables
Chef B: B Finished cooking the stir-fry
All work has been completed!
thread::spawnクロージャを受け取り、新しいOSスレッドでそれを実行します。join()対象のスレッドが終了するまで現在のスレッドをブロックし、Result<T>を返します。スレッドがパニックを起こした場合、join()はErrを返します。スレッドの実行順序はオペレーティングシステムのスケジューラによって決定され、実行ごとに異なる場合があります。
(2) ▶ サンプル:mpsc::channel — メッセージング (難易度 ⭐⭐)
// ============================================
// Demo: mpsc::channel Many-Producer, Single-Consumer Messaging
// Simulation: Multiple editors send modification requests to the document server
// ============================================
use std::thread;
use std::sync::mpsc;
use std::time::Duration;
#[derive(Debug)]
enum EditAction {
Insert { user: String, text: String },
Delete { user: String, line: u32 },
Format { user: String, style: String },
}
fn main() {
println!("=== Collaborative Document Editor ===");
// Create a Channel:Sender Can be cloned,Receiver It is the only one
let (tx, rx) = mpsc::channel::<EditAction>();
// --- 1. Start the receive thread (Document Server) ---
let receiver = thread::spawn(move || {
for action in rx {
match &action {
EditAction::Insert { user, text } => {
println!("[Server] {} Inserted: {}", user, text);
}
EditAction::Delete { user, line } => {
println!("[Server] {} Deleted line {}", user, line);
}
EditAction::Format { user, style } => {
println!("[Server] {} Formatting has been applied: {}", user, style);
}
}
// Simulated Processing Time
thread::sleep(Duration::from_millis(20));
}
println!("[Server] Passage Closed,Unsubscribe");
});
// --- 2. Create the first editor thread (Alice) ---
let tx1 = tx.clone();
let editor1 = thread::spawn(move || {
tx1.send(EditAction::Insert {
user: "Alice".to_string(),
text: "Chapter 1: Rust Introduction".to_string(),
}).unwrap();
thread::sleep(Duration::from_millis(10));
tx1.send(EditAction::Format {
user: "Alice".to_string(),
style: "Bold the title".to_string(),
}).unwrap();
});
// --- 3. Create the second editor thread (Bob) ---
let tx2 = tx.clone();
let editor2 = thread::spawn(move || {
tx2.send(EditAction::Insert {
user: "Bob".to_string(),
text: "Rust is a systems programming language".to_string(),
}).unwrap();
thread::sleep(Duration::from_millis(10));
tx2.send(EditAction::Delete {
user: "Bob".to_string(),
line: 1,
}).unwrap();
});
// --- 4. The main thread also sends a message ---
tx.send(EditAction::Insert {
user: "System".to_string(),
text: "Auto-Save Documents".to_string(),
}).unwrap();
// Wait for the editor thread to finish
editor1.join().unwrap();
editor2.join().unwrap();
// Close the sender - After all Senders are dropped, the Receiver for loop will end automatically
// tx is dropped here (because tx is the last sender on the main thread)
println!("All editors have completed their work");
receiver.join().unwrap();
println!("=== End of editing session ===");
}
// Note: tx is automatically dropped when leaving the scope, channel closes
// If you need to explicitly turn it off,Can be used drop(tx)
出力:
=== Collaborative Document Editor ===
[Server] System Inserted: Auto-Save Documents
[Server] Alice Inserted: Chapter 1: Rust Introduction
[Server] Bob Inserted: Rust is a systems programming language
[Server] Alice Formatting has been applied: Bold the title
[Server] Bob Deleted line 1
All editors have completed their work
[Server] Passage Closed,Unsubscribe
=== End of editing session ===
mpscは「複数のプロデューサー、単一のコンシューマー」を意味します。Senderはcloneを通じて複数のスレッドにコピーできますが、Receiverは 1 つしか存在できません。すべてのSenderが破棄されると、チャネルは自動的に閉じられ、Receiverのイテレータは終了します。sendはResultを返します。受信側が閉じられている場合は、Errを返します。
(3) ▶ サンプル:Arc<Mutex<T>>—共有状態へのスレッドセーフなアクセス (難易度 ⭐⭐⭐)
// ============================================
// Demo: Arc<Mutex<T>> Safely Sharing Data Among Multiple Threads
// Simulation: Multiple workers modifying a shared counter simultaneously
// ============================================
use std::thread;
use std::sync::{Arc, Mutex};
use std::time::Duration;
fn main() {
println!("=== Shared Counter Demo ===");
// Use Arc<Mutex<i32>> to wrap shared data
let counter = Arc::new(Mutex::new(0i32));
let mut handles = vec![];
// --- Start 5 threads,Each thread increments the counter by 10 ---
for id in 0..5 {
let counter_clone = Arc::clone(&counter);
let handle = thread::spawn(move || {
for _ in 0..10 {
// lock() Acquire a mutex lock——If the lock is held by another thread,The current thread will block while waiting
let mut num = counter_clone.lock().unwrap();
*num += 1;
println!("[Workers{}] Current Count: {}", id, *num);
// Locks are automatically released when they go out of scope
}
println!("[Workers{}] Work Completed", id);
});
handles.push(handle);
}
// --- Wait for all threads to finish ---
for handle in handles {
handle.join().unwrap();
}
// --- Read the final results ---
let final_count = counter.lock().unwrap();
println!("\n=== Final Results ===");
println!("Final counter value: {}", *final_count);
println!("Expected value: {} (5 threads x 10 times)", 5 * 10);
}
出力:
=== Shared Counter Demo ===
[Workers0] Current Count: 1
[Workers0] Current Count: 2
[Workers1] Current Count: 3
[Workers1] Current Count: 4
[Workers0] Current Count: 5
... (The intermediate output varies depending on thread scheduling)
[Workers4] Current Count: 50
[Workers4] Work Completed
=== Final Results ===
Final counter value: 50
Expected value: 50 (5 threads x 10 times)
Mutex<T>は排他制御を提供します。つまり、内部データにアクセスできるスレッドは一度に 1 つだけです。lock()はMutexGuard<T>を返します。MutexGuard<T>はDerefおよびDropを実装しており、スコープを離れるとロックは自動的に解放されます。Arc<T>(アトミック参照カウント)は、Rc<T>のマルチスレッド版であり、アトミック操作を使用して参照カウントのスレッドセーフ性を確保します。Arc::cloneは、データをコピーすることなく参照カウントをインクリメントします。
(4) ▶ サンプル:スレッド間の連携――プロデューサー・コンシューマーパターン(難易度 ⭐⭐⭐)
// ============================================
// Producer-Consumer: Multiple producers + Single consumer
// ============================================
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
let tx2 = tx.clone();
thread::spawn(move || {
let items = vec!["Apple", "Banana", "Orange"];
for item in items {
tx.send(format!("Producer1: {}", item)).unwrap();
thread::sleep(Duration::from_millis(100));
}
println!("Producer1 Done");
});
thread::spawn(move || {
let items = vec!["Watermelon", "Grapes"];
for item in items {
tx2.send(format!("Producer2: {}", item)).unwrap();
thread::sleep(Duration::from_millis(150));
}
println!("Producer2 Done");
});
println!("=== Consumer Receipt ===");
for msg in rx {
println!(" Received: {}", msg);
}
println!("All producers have finished, consumer section ended");
}
出力(順序は異なる場合があります):
=== Consumer Receipt ===
Received: Producer1: Apple
Received: Producer2: Watermelon
Received: Producer1: Banana
Received: Producer1: Orange
Received: Producer2: Grapes
Producer1 Done
Producer2 Done
All producers have finished, consumer section ended
tx.clone()は、複数の送信者を作成してマルチプロデューサーパターンを実装します。すべての送信者(txおよびtx2)が破棄されると、rxイテレータは自動的に終了します。手動で「終了」シグナルを送信する必要はありません。
(5) ▶ サンプル:題 5:総合演習—並列データ処理(難易度 ⭐⭐⭐)
// ============================================
// Parallel Computing: Multithreaded Sharded Summation
// ============================================
use std::sync::{Arc, Mutex};
use std::thread;
fn parallel_sum(data: &[i64], num_threads: usize) -> i64 {
let chunk_size = (data.len() + num_threads - 1) / num_threads;
let result = Arc::new(Mutex::new(0i64));
let mut handles = Vec::new();
for i in 0..num_threads {
let chunk = data[i * chunk_size..(i * chunk_size + chunk_size).min(data.len())].to_vec();
let result = Arc::clone(&result);
handles.push(thread::spawn(move || {
let partial: i64 = chunk.iter().sum();
*result.lock().unwrap() += partial;
partial
}));
}
let mut partials = Vec::new();
for handle in handles {
partials.push(handle.join().unwrap());
}
println!("The various thread sections and: {:?}", partials);
*result.lock().unwrap()
}
fn main() {
let data: Vec<i64> = (1..=1000).collect();
let sequential_sum: i64 = data.iter().sum();
println!("=== Serial Summation ===");
println!("1 to 1000 sum: {}", sequential_sum);
println!("\n=== Parallel Summation (4 Thread) ===");
let parallel_result = parallel_sum(&data, 4);
println!("Parallel Summation Results: {}", parallel_result);
assert_eq!(sequential_sum, parallel_result);
let data2: Vec<i64> = (1..=10_000_000).collect();
let start = std::time::Instant::now();
let _ = data2.iter().sum::<i64>();
let seq_time = start.elapsed();
let start = std::time::Instant::now();
let _ = parallel_sum(&data2, 8);
let par_time = start.elapsed();
println!("\n=== 10M Data Performance Comparison ===");
println!("Serial: {:?}", seq_time);
println!("Parallel (8Thread): {:?}", par_time);
}
出力:
=== Serial Summation ===
1 to 1000 sum: 500500
=== Parallel Summation (4 Thread) ===
The various thread sections and: [78126, 218874, 109374, 94126]
Parallel Summation Results: 500500
=== 10M Data Performance Comparison ===
Serial: [Time]
Parallel (8Thread): [Time]
parallel_sumデータを分割し、各スレッドが部分和を計算した後、Arc<Mutex<i64>>を通じてそれらを合計して最終結果を得る。chunk_sizeすべてのデータが確実に反映されるよう、切り上げを行う。assert_eq!並列処理の結果が逐次処理の結果と一致することを確認する。大規模なデータセットの場合、マルチスレッド化によりパフォーマンスを大幅に向上させることができる。
❓ よくある質問
thread::spawn 作成されたスレッドとメインスレッドの関係はどのようなものですか?mpsc::channelのsendとrecvはブロックしますか?sendは通常ブロックしません(バッファに空きがあれば直ちに返ります)。一方、recvはメッセージを受信するまでブロックします。Mutex<T> と Arc<Mutex<T>> は必ず一緒に使用する必要がありますか?Mutex を使用するとデッドロックが発生する可能性がありますか?Send と Sync の違いは何ですか?Send は「スレッド間で所有権を譲渡できる」ことを意味し、Sync は「スレッド間で参照を共有できる」ことを意味します。📖 まとめ
thread::spawnオペレーティングシステムのスレッドを作成し、実行本体としてクロージャを受け取ります。join()は、そのスレッドが終了するのを待ち、結果を取得します。moveクロージャ ダングリング参照を回避するために、値の所有権を新しいスレッドに移譲する。これは、並行処理におけるRustの所有権システムの直接的な応用例である。mpsc::channelは、マルチプロデューサー・シングルコンシューマー型のチャネルを提供します。sendがメッセージを送信し、recvがメッセージを受信します。すべての送信者が切断されると、チャネルは自動的に閉じられます。Mutex<T>は排他制御を行います。lock()はロックを取得します(ブロックします)。MutexGuardはスコープ外になると自動的にロックを解放します。Arc<T>はRc<T>のスレッドセーフ版であり、アトミック操作を用いて参照カウントを管理し、複数のスレッド間でMutex<T>を共有するためのブリッジとして機能します。Send/Syncは、Rust の並行処理安全性を保証するマーカー・トレイトです。Sendはスレッド間の所有権の譲渡を許可し、Syncはスレッド間の参照の共有を許可します。コンパイラはコンパイル時にチェックを行います。
📝 練習問題
- 難易度 ⭐:1..=10、11..=20、21..=30 の各範囲の和を計算するために、3つのスレッドを作成するプログラムを作成してください。メインスレッドは
joinを使用してすべてのスレッドの処理完了を待ち、3つの部分和を合算して最終結果を出力します。 - 難易度 ⭐⭐:
mpsc::channelを使用して「タスクディスパッチャ」を実装してください。プロデューサースレッドを 1 つ(1 から 10 までの番号が付いた 10 個のタスクを生成する)、コンシューマースレッドを 2 つ(各コンシューマーはチャネルからタスク番号を受け取り、「[コンシューマー X] タスク #N を処理中」と出力する)作成してください。すべてのタスクが確実に処理されるようにしてください。 - 難易度 ⭐⭐⭐:「共有銀行口座」システムを実装してください。
Arc<Mutex<f64>>を共有残高として使用します。預金操作をシミュレートするために 4 つのスレッドを作成してください(各スレッドは 10 元から 100 元の間のランダムな金額を預金します)。メインスレッドは、すべての入金スレッドが終了するのを待ってから、最終残高を読み取ります。追加要件:レースコンディションが発生していないことを確認するため、各入金前後の残高の変化を出力してください(最終残高 = 初期残高 + すべての入金の合計)。



