Rust中含MPSC通道的线程无法Join的原因及解决方案咨询

为什么银行线程无法join()？ #

核心原因在于MPSC通道的接收端不知道所有发送端已经关闭。

你用mpsc::channel()创建了一对(Sender, Receiver)：customers是发送端，banker是接收端。循环里为每个客户线程克隆了customers发送端，但主线程里的原始customers发送端并没有被销毁。

MPSC通道的规则是：只有当**所有发送端（包括克隆的实例）都被drop**之后，接收端的迭代器（比如into_iter()）才会终止。因为主线程还保留着一个Sender实例，银行线程的banker.into_iter()会一直阻塞，等待这个发送端可能发送的新消息，永远不会主动退出，自然主线程的bank.join()就会卡住。

可靠的修复方案 #

最简单的修复是在所有客户线程启动完成后，主动drop掉主线程里的原始Sender，这样所有发送端都被销毁，接收端就会知道不会再有新消息，迭代器会正常结束。

修复后的代码如下：

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

// Transaction enum
enum Transaction {
    Withdrawal(String, f64), // 注：原拼写Widthdrawl有误，修正为规范的Withdrawal
    Deposit(String, f64),
}

fn main() {
    //A banking send receive example...
    // set the number of customers
    let n_customers = 10;
    // Create a "customer" and a "banker"
    let (customers, banker) = mpsc::channel();

    let handles = (0..n_customers + 1)
        .into_iter()
        .map(|i| {
            // Create another "customer"
            let customer = customers.clone();
            // Create the customer thread
            let handle = thread::Builder::new()
                .name(format!("{}{}", "thread", i).into())
                .spawn(move || {
                    // Define Transaction
                    let trans_type = match i % 2 {
                        0 => Transaction::Deposit(
                            thread::current().name().unwrap().to_string(),
                            (i + 5) as f64 * 10.0,
                        ),
                        _ => Transaction::Withdrawal(
                            thread::current().name().unwrap().to_string(),
                            (i + 10) as f64 * 5.0,
                        ),
                    };
                    // Send the Transaction
                    customer.send(trans_type).unwrap();
                });
            handle
        })
        .collect::<Vec<Result<thread::JoinHandle<_>, _>>>();

    // 关键操作：销毁主线程中的原始Sender，让接收端知晓所有发送端已关闭
    drop(customers);

    // Wait for threads to finish
    for handle in handles {
        handle.unwrap().join().unwrap()
    }

    // Create a bank thread
    let bank = thread::spawn(move || {
        // Create a value
        let mut balance: f64 = 10000.0;
        println!("Initially, Bank value: {}", balance);
        // Perform the transactions in order
        banker.into_iter().for_each(|i| {
            let mut customer_name: String = "None".to_string();
            match i {
                // Subtract for Withdrawals
                Transaction::Withdrawal(cust, amount) => {
                    customer_name = cust;
                    println!(
                        "Customer name {} doing withdrawal of amount {}",
                        customer_name, amount
                    );
                    balance = balance - amount;
                }
                // Add for deposits
                Transaction::Deposit(cust, amount) => {
                    customer_name = cust;
                    println!(
                        "Customer name {} doing deposit of amount {}",
                        customer_name, amount
                    );
                    balance = balance + amount;
                }
            }
            println!("Customer is {}, Bank value: {}", customer_name, balance);
        });
    });

    // Let the bank finish
    bank.join().unwrap(); // 现在可以正常结束了！
}

其他可靠的结束方式 #

除了依赖Sender的drop，你还可以通过显式控制消息来通知银行线程停止，比如给Transaction添加一个Exit变体：

enum Transaction {
    Withdrawal(String, f64),
    Deposit(String, f64),
    Exit, // 新增结束消息
}

在所有客户线程完成后，发送一个Exit消息，银行线程收到后就可以跳出循环主动结束。这种方式更灵活，适合需要精确控制线程生命周期的复杂场景。

内容的提问来源于stack exchange，提问作者Prana