单线程Rust中阻塞函数的处理方案问询

针对你在单线程本地异步执行器中处理阻塞函数的问题，分两种场景给出具体方案：

一、阻塞操作的对象可跨线程（实现Send trait） #

这种场景下可以用轻量线程池+一次性通道封装阻塞函数为Future，无需手动管理线程，也不用显式使用Arc（只要闭包捕获的是对象所有权而非引用）：

1. 引入轻量线程池 crate（比如threadpool），初始化一个固定大小的后台线程池（根据业务需求设置，比如2-4个线程）
2. 编写包装函数，将阻塞逻辑打包成任务提交到线程池，通过oneshot通道返回结果
3. 将通道的接收端作为Future在本地执行器中await

示例代码：

use threadpool::ThreadPool;
use futures::channel::oneshot;
use std::sync::LazyLock;

// 全局初始化阻塞任务线程池
static BLOCKING_TASK_POOL: LazyLock<ThreadPool> = LazyLock::new(|| ThreadPool::new(2));

/// 将阻塞函数包装为可await的Future
fn spawn_blocking<F, R>(f: F) -> impl futures::Future<Output = R> + 'static
where
    F: FnOnce() -> R + Send + 'static,
    R: Send + 'static,
{
    let (sender, receiver) = oneshot::channel();
    BLOCKING_TASK_POOL.execute(move || {
        let result = f();
        // 忽略发送失败（比如接收端已被丢弃）
        let _ = sender.send(result);
    });
    // 将通道接收转为Future
    async move { receiver.await.unwrap() }
}

// 使用示例
async fn handle_nanomsg() {
    let mut socket = nanomsg::Socket::new(nanomsg::Protocol::Pull).unwrap();
    socket.connect("tcp://127.0.0.1:5555").unwrap();
    
    // 包装阻塞的read_to_string为Future
    let msg = spawn_blocking(|| socket.read_to_string().unwrap()).await;
    println!("Received: {}", msg);
}

二、阻塞操作的对象不可跨线程（未实现Send trait） #

如果像nanomsg socket这类对象无法安全跨线程，就不能用线程池方案，必须通过事件驱动+非阻塞IO将阻塞操作转为异步：

1. 将socket设置为非阻塞模式
2. 给你的本地执行器添加事件循环支持（可借助mio这类底层IO事件库）
3. 监听socket的可读事件，当事件触发时再调用非阻塞的读取方法，避免主线程阻塞

示例代码：

use mio::{Events, Interest, Poll, Token};
use nanomsg::{Socket, Protocol, Error};
use std::time::Duration;

// 假设你的本地执行器已集成mio的Poll实例
async fn async_read(socket: &mut Socket) -> Result<String, Error> {
    // 将socket设为非阻塞
    socket.set_nonblocking(true)?;
    
    let poll = Poll::new().unwrap();
    let token = Token(0);
    poll.registry().register(&socket.as_fd(), token, Interest::READABLE).unwrap();
    
    let mut events = Events::with_capacity(1);
    loop {
        match socket.read_to_string() {
            Ok(msg) => return Ok(msg),
            Err(Error::WouldBlock) => {
                // 等待socket可读事件，这里需要将mio的阻塞等待转为异步
                // （可通过执行器的事件调度逻辑实现非阻塞等待）
                poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
                if events.iter().any(|e| e.token() == token && e.is_readable()) {
                    continue;
                }
            }
            Err(e) => return Err(e),
        }
    }
}

// 使用示例
async fn handle_nanomsg() {
    let mut socket = Socket::new(Protocol::Pull).unwrap();
    socket.connect("tcp://127.0.0.1:5555").unwrap();
    
    let msg = async_read(&mut socket).await.unwrap();
    println!("Received: {}", msg);
}

两种方案都能避免主线程阻塞，且无需手动管理次级线程的生命周期；第二种方案还能保留单线程数据访问的优势，不用引入多线程同步机制。

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