是否存在多线程竞速完成的API？如何优先获取首个线程输出？

嘿，我来帮你搞定这两个Rust线程相关的问题，都是日常并发编程里挺实用的场景：

1. 实现N个线程竞速的方案 #

Rust标准库本身没有直接提供“一键让线程竞速”的API，但我们可以用std::sync::Barrier这个工具来实现所有线程同时启动任务，这样就能达到公平竞速的效果啦。

Barrier就像一个集合点：你创建它的时候指定要等多少个线程，每个线程执行到barrier.wait()的时候就会停下来，直到所有线程都到了这个点，大家再一起继续往下跑。这样所有线程的任务就相当于同时启动，完美实现竞速。

给你写个示例代码：

use std::sync::Barrier;
use std::thread;
use std::time::Instant;

struct Output(i32);

fn main() {
    let thread_count = 10;
    // 创建一个等待10个线程的屏障
    let barrier = Barrier::new(thread_count);
    let mut thread_handles = Vec::with_capacity(thread_count);

    for i in 0..thread_count {
        let barrier_clone = barrier.clone();
        thread_handles.push(thread::spawn(move || {
            // 等所有线程都准备好，一起出发
            barrier_clone.wait();
            
            // 这里开始模拟耗时任务，每个线程的耗时不一样
            let start_time = Instant::now();
            thread::sleep(std::time::Duration::from_millis(i as u64 * 10));
            let elapsed = start_time.elapsed();
            
            Output(i as i32)
        }));
    }

    // 等所有线程跑完，看看结果
    for handle in thread_handles {
        let output = handle.join().unwrap();
        println!("线程{}完成竞速", output.0);
    }
}

要是你想更灵活（比如只关心第一个跑完的线程，不用等所有），那可以结合通道和select操作，这个咱们在第二个问题里详细说。

2. 获取第一个产生的Output，并按顺序拿剩余结果（还要处理线程不终止的情况） #

这个需求有点意思，既要抓第一个完成的结果，还要按顺序收剩下的，同时得防着有些线程卡死不结束。核心思路是让每个线程把结果发到通道里，然后我们通过监听多个通道的消息来实现，这里推荐用crossbeam-channel库，它比标准库的mpsc通道功能强很多，尤其是select!宏特别好用。

首先得在Cargo.toml里加依赖：

[dependencies]
crossbeam-channel = "0.5"

然后看示例代码，我特意加了一个永远不终止的线程（i=5那个），演示怎么处理这种情况：

use crossbeam-channel::{unbounded, Receiver, Select};
use std::thread;
use std::time::{Duration, Instant};

struct Output(i32);

fn main() {
    let thread_count = 10;
    let mut receivers = Vec::with_capacity(thread_count);

    for i in 0..thread_count {
        let (sender, receiver) = unbounded();
        receivers.push(receiver);

        thread::spawn(move || {
            // 模拟某个线程无限休眠，永远不返回结果
            if i == 5 {
                loop {
                    thread::sleep(Duration::from_secs(1));
                }
            }
            
            // 正常线程：模拟随机耗时任务
            let sleep_duration = Duration::from_millis((i as u64 * 50) % 500);
            thread::sleep(sleep_duration);
            // 把结果发出去
            sender.send(Output(i as i32)).unwrap();
        });
    }

    // 准备select操作，监听所有接收器
    let mut select = Select::new();
    for r in &receivers {
        select.recv(r);
    }

    let mut received_results = Vec::new();
    let mut remaining_threads = receivers.len();

    // 循环获取结果，直到收完所有正常线程的结果，或者超时
    while remaining_threads > 0 {
        match select.select_timeout(Duration::from_secs(3)) {
            Ok(operation) => {
                let receiver = operation.receiver();
                if let Ok(output) = operation.recv() {
                    println!("拿到结果啦：线程{}", output.0);
                    received_results.push(output);
                    remaining_threads -= 1;
                    // 把已经完成的接收器从select里移除，不再监听
                    select.remove(receiver);
                }
            }
            Err(_) => {
                println!("超时啦！还有{}个线程没结束，不等了", remaining_threads);
                break;
            }
        }
    }

    println!("最终收到的结果列表：{:?}", received_results.iter().map(|o| o.0).collect::<Vec<_>>());
}

关键点解释： #

• 第一个结果怎么拿：select!会立刻响应第一个有消息的通道，所以第一个被处理的就是最早完成的线程的结果。
• 按顺序拿剩余结果：每次处理完一个结果，我们就把对应的接收器从Select里删掉，后续的select会继续监听剩下的通道，自然就能按结果产生的顺序获取剩下的输出。
• 处理不终止的线程：用select_timeout设置一个超时时间，要是超过这个时间还有线程没返回结果，就直接停止等待，避免程序一直挂着。

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