使用Rust Tokio操作MongoDB时遇I/O超时错误的排查与解决

错误解释 #

这个错误的核心逻辑是：MongoDB驱动的连接池被主动清空了，触发原因是某个数据库操作出现了I/O超时。驱动会在检测到连接失效（比如超时、断开）时清空整个连接池，避免后续操作复用无效连接。而RetryableWriteError标签说明这类写入错误属于可以重试的范畴，但你的代码使用try_join_all，只要有一个任务失败就会终止全部流程，再加上那个特定文件的操作反复触发超时，导致整个处理无法推进。

解决方法 #

1. 限制并发请求数，匹配连接池能力 #

你当前的逻辑是每个信息组都单独用tokio::spawn启动任务，100万文件+每组几百个操作的规模会直接撑爆MongoDB默认连接池（默认最大连接数一般为100），导致连接耗尽、超时。

• 用tokio::sync::Semaphore控制并发数，设置值与MongoDB连接池大小一致（比如100）：

use tokio::sync::Semaphore;
use std::sync::Arc;

// 初始化信号量，限制并发数
let semaphore = Arc::new(Semaphore::new(100));

// 遍历文件时，每个任务先获取许可
let permit = semaphore.acquire().await.unwrap();
let task = tokio::spawn(async move {
    // 执行update_one操作
    let result = collection.update_one(filter, update, options).await;
    drop(permit); // 释放许可
    result
});

• 同时调整MongoDB客户端的连接池配置，增大连接池上限并延长超时时间：

use mongodb::options::ClientOptions;
use std::time::Duration;

let mut client_options = ClientOptions::parse("mongodb://localhost:27017").await?;
// 调整连接池最大大小
client_options.max_pool_size = Some(200);
// 延长socket超时时间（比如10秒）
client_options.socket_timeout = Some(Duration::from_secs(10));
// 延长连接超时时间
client_options.connect_timeout = Some(Duration::from_secs(5));

let client = mongodb::Client::with_options(client_options)?;

2. 给可重试错误添加重试逻辑 #

既然错误标记了RetryableWriteError，说明这类超时、连接池错误可以通过重试解决。给每个update_one操作添加指数退避式重试：

use std::time::Duration;
use backoff::{ExponentialBackoff, Error as BackoffError, future::retry};

// 定义重试策略：最多重试5次，初始间隔1秒，最大间隔5秒
let backoff = ExponentialBackoff {
    max_elapsed_time: Some(Duration::from_secs(30)),
    max_interval: Duration::from_secs(5),
    ..Default::default()
};

let result = retry(backoff, || async {
    match collection.update_one(filter, update, options).await {
        Ok(res) => Ok(res),
        Err(e) => {
            // 判断是否为可重试错误
            if e.labels().contains("RetryableWriteError") 
                || matches!(e.kind(), mongodb::error::ErrorKind::ConnectionPoolCleared(_))
            {
                Err(BackoffError::Transient { err: e, retry_after: None })
            } else {
                Err(BackoffError::Permanent(e))
            }
        }
    }
}).await;

3. 改用批量操作减少请求次数 #

不要每个信息组单独调用update_one，把同一文件的所有信息组整理成批量更新请求，用bulk_write执行，大幅减少网络请求和连接占用：

use mongodb::options::UpdateOptions;
use mongodb::operations::UpdateOneModel;

let mut models = Vec::new();
for info_group in file_info_groups {
    let filter = // 你的查询条件
    let update = // 你的更新逻辑
    let options = UpdateOptions::builder().upsert(true).build();
    models.push(UpdateOneModel::builder()
        .filter(filter)
        .update(update)
        .options(options)
        .build());
}

// 执行批量更新
let result = collection.bulk_write(models, None).await;

4. 替换try_join_all，避免全局终止 #

try_join_all会在任意一个任务失败时立即终止所有任务，改用join_all配合单独的错误处理，或者用Stream逐个处理任务结果，失败的任务单独重试：

use futures::stream::{self, StreamExt};

// 把任务转换成流，用buffer_unordered控制并发
let tasks = stream::iter(file_tasks)
    .map(|task| async {
        match task.await {
            Ok(res) => Ok(res),
            Err(e) => {
                // 单独处理错误：记录日志+重试该任务
                eprintln!("Task failed: {:?}", e);
                // 这里插入重试逻辑...
                Err(e)
            }
        }
    })
    .buffer_unordered(100); // 控制并发数

tasks.collect::<Vec<_>>().await;

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