404 Not Found

404 Not Found


nginx

Rustにおける並行プログラミング:スレッド、メッセージング、および共有状態

並行性とは、プログラムが複数の処理を「同時に」実行する能力のことです。Rustは、所有権システムと型システムを通じてコンパイル時にデータ競合を排除するため、効率的かつ安全な並行コードを記述することができます。

シングルスレッドのプログラムを「最初から最後まで1人でキッチンで作業する」ことに例えるなら、マルチスレッドは「複数のシェフが同時に作業する」ことに例えられます。ある人は野菜を切り、ある人は炒め物を作り、またある人は料理を盛り付けます。しかし、キッチンに人が多すぎると、すぐに混乱が生じやすくなります。2人が同時に同じ包丁に手を伸ばしたり、ある人が他の人が現在使っている食材を手に取ったりしてしまうかもしれません。Rustの並行処理モデルは、「ルールのある複数シェフのキッチン」のようなものです。各シェフには専用の道具があり、食材は専用の経路を通じて渡され、共有の調味料は一度に1人しか使用できません。


1. 学習内容


2. 共同文書編集システムの物語

(1) 課題:並行制御のないオンラインドキュメント

ルナの会社は、オンラインの共同文書編集ツールを開発しています。初期のバージョンでは、一度に1人しか編集できなかったため、チームメンバーから絶えず不満の声が上がっていました:

「エディターごとに個別のスレッドを立ち上げ、変更依頼をチャネル経由でやり取りし、ドキュメントの内容をミューテックスで保護できれば素晴らしいですね……」

(2) Rustの並行処理モデルへのアプローチ

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並行処理フレームワーク

100%
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&lt;T&gt; Mutex"]
    D --> D2["lock() Acquire a lock"]
    D --> D3["Arc&lt;T&gt; 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つのマーカートレイトであるSendSyncを用いてコンパイル時のチェックを行います。ある型がSendでない場合、その型を別のスレッドに渡そうとすると、コンパイル時にエラーが発生します。


4. 並行プログラミングの例

(1) ▶ サンプル:thread::spawn + join — スレッドの作成と待機 (難易度 ⭐)

RUST
// ============================================
// 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!");
}

出力:

TEXT
=== 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 — メッセージング (難易度 ⭐⭐)

RUST
// ============================================
// 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)

出力:

TEXT
=== 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 は「複数のプロデューサー、単一のコンシューマー」を意味します。Senderclone を通じて複数のスレッドにコピーできますが、Receiver は 1 つしか存在できません。すべての Sender が破棄されると、チャネルは自動的に閉じられ、Receiver のイテレータは終了します。sendResultを返します。受信側が閉じられている場合は、Errを返します。


(3) ▶ サンプル:Arc<Mutex<T>>—共有状態へのスレッドセーフなアクセス (難易度 ⭐⭐⭐)

RUST
// ============================================
// 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);
}

出力:

TEXT
=== 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) ▶ サンプル:スレッド間の連携――プロデューサー・コンシューマーパターン(難易度 ⭐⭐⭐)

RUST
// ============================================
// 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");
}

出力(順序は異なる場合があります):

TEXT
=== 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:総合演習—並列データ処理(難易度 ⭐⭐⭐)

RUST
// ============================================
// 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);
}

出力:

TEXT
=== 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! 並列処理の結果が逐次処理の結果と一致することを確認する。大規模なデータセットの場合、マルチスレッド化によりパフォーマンスを大幅に向上させることができる。


❓ よくある質問

Q thread::spawn 作成されたスレッドとメインスレッドの関係はどのようなものですか?
A すべてのスレッドは並行して実行されます。メインスレッドが終了すると、プログラム全体が終了します。
Q mpsc::channelsendrecvはブロックしますか?
A sendは通常ブロックしません(バッファに空きがあれば直ちに返ります)。一方、recvはメッセージを受信するまでブロックします。
Q Mutex<T>Arc<Mutex<T>> は必ず一緒に使用する必要がありますか?
A 必ずしもそうではありません。
Q Mutex を使用するとデッドロックが発生する可能性がありますか?
A Rust はデッドロックを防止しません。コード内でデッドロックを回避する必要があります。
Q SendSync の違いは何ですか?
A Send は「スレッド間で所有権を譲渡できる」ことを意味し、Sync は「スレッド間で参照を共有できる」ことを意味します。

📖 まとめ


📝 練習問題

  1. 難易度 ⭐:1..=10、11..=20、21..=30 の各範囲の和を計算するために、3つのスレッドを作成するプログラムを作成してください。メインスレッドは join を使用してすべてのスレッドの処理完了を待ち、3つの部分和を合算して最終結果を出力します。
  2. 難易度 ⭐⭐: mpsc::channel を使用して「タスクディスパッチャ」を実装してください。プロデューサースレッドを 1 つ(1 から 10 までの番号が付いた 10 個のタスクを生成する)、コンシューマースレッドを 2 つ(各コンシューマーはチャネルからタスク番号を受け取り、「[コンシューマー X] タスク #N を処理中」と出力する)作成してください。すべてのタスクが確実に処理されるようにしてください。
  3. 難易度 ⭐⭐⭐:「共有銀行口座」システムを実装してください。Arc<Mutex<f64>> を共有残高として使用します。預金操作をシミュレートするために 4 つのスレッドを作成してください(各スレッドは 10 元から 100 元の間のランダムな金額を預金します)。メインスレッドは、すべての入金スレッドが終了するのを待ってから、最終残高を読み取ります。追加要件:レースコンディションが発生していないことを確認するため、各入金前後の残高の変化を出力してください(最終残高 = 初期残高 + すべての入金の合計)。
Web-Tutorial.com

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

100%