Actix Web应用中LocalPool执行器EnterError的Rust修复方法咨询

问题 #

运行Actix Web应用时，启动了一个线程订阅MQTT网关。用Swagger测试REST API端点一切正常，但通过Flutter应用发送请求后，几秒内会出现错误：cannot execute LocalPool executor from within another executor: EnterError。Flutter应用发送请求后也会订阅一个MQTT主题。

错误出自这段Rust代码：

async fn init() {
    // 连接Redis
    redis_service::RedisCache::init().await;

    // 错误出现在这里
    thread::spawn(|| {
        // 等待新消息
        crate::service::subscriber_service::subscriber();
    });
}

MQTT订阅者使用的是paho-mqtt异步客户端。Flutter请求代码如下：

dio.interceptors.add(ApiProviderTokenInterceptor());

final res = await dio.post(path, queryParameters: query);
if (context.mounted) {
  if (res.statusCode == 200) {
    final mqtt = Mqtt();
    mqtt.subscribe();
    _displayDialog(context);
  } else {
    await dialog(
        context,
        "扫描失败",
        "扫描未成功。如需帮助，请联系管理员"
    );
  }
}

对应的Rust错误栈：

0: rust_begin_unwind
             at /rustc/eeb90cda1969383f56a2637cbd3037bdf598841c/library/std/src/panicking.rs:665:5
   1: core::panicking::panic_fmt
             at /rustc/eeb90cda1969383f56a2637cbd3037bdf598841c/library/core/src/panicking.rs:74:14
   2: core::result::unwrap_failed
             at /rustc/eeb90cda1969383f56a2637cbd3037bdf598841c/library/core/src/result.rs:1679:5
   3: core::result::Result<T,E>::expect
             at /rustc/eeb90cda1969383f56a2637cbd3037bdf598841c/library/core/src/result.rs:1059:23
   4: futures_executor::local_pool::run_executor
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-executor-0.3.30/src/local_pool.rs:81:18
   5: futures_executor::local_pool::block_on
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-executor-0.3.30/src/local_pool.rs:317:5
   6: paho_mqtt::token::Token::wait_for
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/paho-mqtt-0.12.5/src/token.rs:563:9
   7: paho_mqtt::client::Client::connect
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/paho-mqtt-0.12.5/src/client.rs:89:9
   8: isumis::service::publish_service::publisher
             at ./src/service/publish_service.rs:25:21
   9: isumis::service::blacklight_service::blacklight_res::{{closure}}
             at ./src/service/blacklight_service.rs:34:9
  10: isumis::service::subscriber_service::handle_blacklight::{{closure}}
             at ./src/service/subscriber_service.rs:173:26
  11: isumis::service::subscriber_service::message_loop::{{closure}}
             at ./src/service/subscriber_service.rs:137:22
  12: isumis::service::subscriber_service::subscriber::{{closure}}
             at ./src/service/subscriber_service.rs:102:33
  13: futures_executor::local_pool::block_on::{{closure}}
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-executor-0.3.30/src/local_pool.rs:317:23
  14: futures_executor::local_pool::run_executor::{{closure}}
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-executor-0.3.30/src/local_pool.rs:90:37
  15: std::thread::local::LocalKey<T>::try_with
             at /rustc/eeb90cda1969383f56a2637cbd3037bdf598841c/library/std/src/thread/local.rs:283:12
  16: std::thread::local::LocalKey<T>::with
             at /rustc/eeb90cda1969383f56a2637cbd3037bdf598841c/library/std/src/thread/local.rs:260:9
  17: futures_executor::local_pool::run_executor
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-executor-0.3.30/src/local_pool.rs:86:5
  18: futures_executor::local_pool::block_on
             at /home/isumis/.cargo/registry/src/index.crates.io-6f17d22bba15001f/futures-executor-0.3.30/src/local_pool.rs:317:5
  19: isumis::service::subscriber_service::subscriber
             at ./src/service/subscriber_service.rs:81:23
  20: isumis::server::init::{{closure}}::{{closure}}
             at ./src/server.rs:178:9

解决方案 #

问题根源 #

错误核心原因：Actix Web的线程池已存在一个LocalPool执行器，此时在新线程中调用futures_executor::LocalPool::block_on运行异步代码，而LocalPool是线程本地执行器，不支持嵌套使用，从而触发冲突。从错误栈可看到，订阅逻辑中的block_on调用直接引发了该错误。

修复方案 #

方案1：改用Actix的spawn_blocking替代thread::spawn

Actix的spawn_blocking会将任务分配到专门的阻塞线程池，避免与主线程执行器冲突：

async fn init() {
    // 连接Redis
    redis_service::RedisCache::init().await;

    // 使用Actix提供的spawn_blocking
    actix_web::rt::spawn_blocking(|| {
        crate::service::subscriber_service::subscriber();
    });
}

方案2：在新线程中初始化独立的LocalPool

若必须使用thread::spawn，需确保订阅逻辑在新线程中创建独立的LocalPool，避免嵌套调用block_on：

// 重构subscriber函数，自行管理执行器
fn subscriber() {
    let mut local_pool = futures_executor::LocalPool::new();
    let spawner = local_pool.spawner();

    // 将异步订阅任务提交到独立的LocalPool
    spawner.spawn_ok(async {
        // 原异步订阅逻辑示例
        let cli = paho_mqtt::AsyncClient::new("tcp://mqtt.example.com:1883").unwrap();
        let rsp = cli.connect(None).await.unwrap();
        // ... 其他订阅、消息处理逻辑
    });

    // 运行执行器直到任务结束
    local_pool.run();
}

// init函数保持thread::spawn调用
async fn init() {
    redis_service::RedisCache::init().await;

    thread::spawn(|| {
        crate::service::subscriber_service::subscriber();
    });
}

方案3：改用paho-mqtt同步客户端

若异步客户端的执行器冲突难以解决，可考虑切换到同步客户端，绕过异步执行器嵌套问题：

fn subscriber() {
    let cli = paho_mqtt::Client::new("tcp://mqtt.example.com:1883").unwrap();
    let conn_opts = paho_mqtt::ConnectOptions::new();
    cli.connect(conn_opts).unwrap();
    
    // 订阅主题并循环处理消息
    cli.subscribe("topic/#", 1).unwrap();
    loop {
        if let Some(msg) = cli.receive_timeout(std::time::Duration::from_secs(1)) {
            // 处理收到的消息
            println!("收到消息: {:?}", msg);
        }
    }
}

额外注意事项 #

• Flutter端的MQTT订阅需避免与后端订阅产生重复订阅、消息风暴等问题，但这并非当前错误的直接诱因。
• 若项目中同时使用多个异步运行时，尽量统一运行时环境，降低执行器冲突概率。

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