如何给UDP数据包循环接收的异步函数添加1分钟超时功能

这问题我之前也碰到过，普通的async_std::future::timeout确实没法满足需求——毕竟我们要的是到点就返回已收集的所有数据，而不是直接丢弃成果返回错误。用futures库的select!宏就能完美解决这个问题，它可以同时等待多个future，哪个先触发就执行对应的逻辑，完全能保留我们已经收集到的内容。

Step-by-Step Implementation #

首先，确保你的Cargo.toml包含必要的依赖：

[Dependencies]
async-std = { version = "1.12", features = ["attributes"] }
futures = "0.3"
pin-utils = "0.1"

接下来是修改后的函数：

use futures::select;
use std::time::Duration;
use pin_utils::pin_mut;
use async_std::net::UdpSocket;

async fn recv_multiple_with_timeout(socket: &UdpSocket) -> Vec<String> {
    let mut buf = [0; 1024];
    let mut collected_messages = Vec::new();

    // 创建全局超时future（仅一次，确保总时长为1分钟）
    let timeout_fut = async_std::task::sleep(Duration::from_secs(60));
    pin_mut!(timeout_fut); // 固定future，满足select对Unpin的要求

    loop {
        select! {
            // 分支1：等待UDP数据报接收完成
            recv_result = socket.recv_from(&mut buf).fuse() => {
                match recv_result {
                    Ok((amt, _src)) => {
                        // 将字节数据转换为String并加入收集列表
                        let msg = String::from_utf8_lossy(&buf[..amt]).to_string();
                        collected_messages.push(msg);
                    }
                    Err(e) => {
                        // 处理接收错误，此处选择退出循环并返回已收集数据
                        eprintln!("UDP接收失败: {}", e);
                        break;
                    }
                }
            }
            // 分支2：等待超时触发
            _ = timeout_fut => {
                // 超时后直接退出循环，返回已收集的所有数据
                break;
            }
        }
    }

    collected_messages
}

Key Explanations #

• 全局超时而非单次超时：我们只创建一次sleep future，确保整个收集过程最多持续1分钟，而不是每个数据包单独等待1分钟。如果每次循环都重新创建sleep，逻辑就完全偏离需求了。
• pin_mut!的作用：select!宏要求被等待的future实现Unpin trait（防止异步过程中future被移动），用pin_mut!可以轻松将一个future固定在内存中，满足这个要求。
• fuse()的作用：给recv_from返回的future加上fuse()，确保它在完成后不会被再次poll，这是select!宏的最佳实践，避免出现意外行为。
• 灵活的错误处理：如果接收过程中出现错误（比如套接字意外关闭），我们会打印错误并退出循环，返回已经收集到的所有数据——你可以根据业务需求调整这里的逻辑（比如重试或者忽略错误继续收集）。

Usage Example #

use async_std::net::UdpSocket;

fn main() {
    let socket = async_std::task::block_on(UdpSocket::bind("0.0.0.0:8080")).unwrap();
    let results = async_std::task::block_on(recv_multiple_with_timeout(&socket));
    println!("收集到{}条消息: {:?}", results.len(), results);
}

这样运行后，函数会持续收集UDP数据报，1分钟后自动返回已收集的所有内容（即使为空也会返回空Vec），完全符合你的需求。

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