基于Tokio实现Rust异步递归调度：避免重复处理对象

问题原因 #

你的代码无法正常终止（或出现看似未进入循环的情况），核心问题在于：

• 主线程始终持有mpsc的原始发送端tx，从未关闭。当所有初始对象处理完成后，rx.recv().await会一直阻塞等待新消息，无法返回None结束循环。
• 虽然seen集合的逻辑能确保每个对象仅被处理一次，但发送端未正确管理导致程序无法正常退出。

修正后的代码 #

use std::collections::HashSet;

use tokio::sync::mpsc;

async fn process(a: i32) -> Result<Vec<i32>, Box<dyn std::error::Error>> {
    Ok(vec![1])
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel(32);
    let mut seen = HashSet::new();

    println!("Sending initial values");
    // 克隆临时发送端用于初始值发送，避免占用原始发送端
    let tx_initial = tx.clone();
    for n in vec![4, 5, 6] {
        tx_initial.send(n).await.unwrap();
    }
    // 发送完成后立即释放临时发送端，不影响后续发送端的生命周期
    drop(tx_initial);

    while let Some(n) = rx.recv().await {
        println!("Processing {}", n);
        // 仅当对象未被处理过才启动异步任务
        if seen.insert(n) {
            let tx_clone = tx.clone();

            tokio::spawn(async move {
                println!("Start processing {}", n);
                let transitive = process(n).await.unwrap();

                for new_n in transitive {
                    println!("Scheduling {}", new_n);
                    // 忽略发送失败（如接收端已关闭）的情况，避免panic
                    let _ = tx_clone.send(new_n).await;
                }
            });
        }
    }

    println!("All processing tasks completed");
    Ok(())
}

关键改进点 #

1. 发送端生命周期管理：

   • 使用临时克隆的tx_initial发送初始值，发送完成后立即drop，确保原始tx仅被派发给异步任务。
   • 当所有异步任务执行完毕并释放各自的tx_clone后，mpsc的所有发送端都被关闭，rx.recv().await会返回None，循环正常退出。

2. 容错处理：

   • 将tx.send().await.unwrap()改为let _ = tx.send().await，避免在接收端已关闭时发送消息导致程序panic。

3. 重复处理防护：

   • 保留seen集合在主线程中处理，确保每个对象只有在首次被接收时才会启动异步处理任务，严格保证process函数不会被重复调用同一对象。

额外说明 #

如果你的实际A类型无法直接存入HashSet，需要为A实现Hash和Eq trait，这是使用HashSet去重的必要条件。

内容的提问来源于stack exchange，提问作者Maël Nison