如何为TCP Listener实现有界并发？

问题背景 #

我编写了如下TCP监听器代码：

pub async fn run(port: i32) -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("127.0.0.1:{}", port);
    info!("Listening on address: {}", addr);

    let listener = TcpListener::bind(addr).await?;
    loop {
        let (socket, _) = listener.accept().await?;
        tokio::spawn(async move {
            _ = handle_connection(socket).await;
        });
    }
}

假设handle_connection是一项耗时的IO密集型操作，我希望服务器同时运行的handle_connection任务数量始终不超过n个。请问实现该需求的惯用方法是什么？

我最初的想法是在监听器中创建一个FuturesOrdered列表来收集handle_connection任务，定期对该列表执行collect()，运行n个任务并清理已完成的任务，但这种方法感觉不够优雅，想了解是否有更合适的方案。

惯用解决方案：使用Semaphore（信号量） #

在Tokio生态里，最优雅且符合惯用做法的方式是用**信号量（Semaphore）**控制并发数。信号量可以限制同时访问目标资源（这里是并发执行handle_connection的权限）的任务数量，具体实现步骤如下：

1. 创建信号量实例，初始许可数设为n（即最大并发数）
2. 接受每个连接后，先获取一个信号量许可
3. handle_connection执行完成后，许可会自动释放（利用Rust的Drop特性）

修改后的代码示例：

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

pub async fn run(port: i32, max_concurrent: usize) -> Result<(), Box<dyn std::error::Error>> {
    let addr = format!("127.0.0.1:{}", port);
    info!("Listening on address: {}", addr);

    // 创建信号量，初始许可数为设定的最大并发数
    let semaphore = Arc::new(Semaphore::new(max_concurrent));
    let listener = TcpListener::bind(addr).await?;

    loop {
        let (socket, _) = listener.accept().await?;
        // 克隆信号量的Arc引用，用于子任务中获取许可
        let semaphore_clone = Arc::clone(&semaphore);

        tokio::spawn(async move {
            // 获取许可：若当前并发数已达上限，会异步等待直到有许可释放
            let _permit = semaphore_clone.acquire().await.unwrap();
            // 执行连接处理逻辑，任务完成后_permit被销毁，自动释放许可
            let _ = handle_connection(socket).await;
        });
    }
}

方案优势 #

• 无需手动维护任务列表或清理完成的任务，信号量自动处理许可的分配与回收
• 代码简洁直观，完全契合Rust/Tokio的异步编程范式
• 性能高效：acquire操作是异步非阻塞的，不会浪费线程资源

为什么不推荐FuturesOrdered方案 #

FuturesOrdered的设计初衷是按顺序处理异步任务，而非控制并发上限。用它手动维护任务列表不仅代码冗余，还容易出现任务清理不及时、并发数控制不准确的问题，属于工具场景错配。

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