多谢各位大神，原因是actix是单线程的runtime，不是程序的问题。

准备换web框架了。

我不知道你的rx是什么类型, 所以写了个小demo,发送信号是mpsc

use tokio::sync::mpsc;
use tokio::time::{self, Duration};

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Sig {
    Continue,
    Break,
}

#[tokio::main]
async fn main() {
    let mut ticker = time::interval(Duration::from_secs(1));

    let (tx, mut rx) = mpsc::channel(4);

    tokio::spawn(async move {
        let mut t = time::interval(Duration::from_secs(1));
        if let Err(_) = tx.send(Sig::Continue).await {
            println!("Send Err");
        }
        for _ in 0..10 {
            t.tick().await;
        }
        if let Err(_) = tx.send(Sig::Continue).await {
            println!("Send Err");
        }
        for _ in 0..2 {
            t.tick().await;
        }
        if let Err(_) = tx.send(Sig::Break).await {
            println!("Send Err");
        }
    });

    loop {
        tokio::select! {
            _ = ticker.tick() => {
                println!("tick");
            },
            msg = rx.recv() => {
                println!("message is:{:?}",msg);
                if msg == Some(Sig::Break){
                    return;
                }
            }
        }
    }
}

n年前的代码，可以参考一下https://github.com/laohanlinux/consensus-rs/blob/master/src/consensus/pbft/core/timer.rs

1.看文档你不用WebSocket的话是不需要System的，可以替换为tokio::main
2.System也可以不用actix_web::main创建，可以手动创建调with_tokio_rt自行绑定tokio runtime
3.自己在需要的地方创建tokio runtime，并block_on，这些spawn都放到block_on下层

求大神救救孩子吧！

--
👇
dlhxzb: 起码还有三种方法

起码还有三种方法

github上面也看到了。。。

so。。。没法继续啦？！

Since you are using atcix::main macro. It means you are running on a single thread runtime. calling tokio::task::spawn would not make your task run on other threads.

--
👇
dlhxzb: actix_web::main
-> system::new
-> runtime::default_tokio_runtime
看到这里就知道它是个new_current_thread单线程啦

actix_web::main
-> system::new
-> runtime::default_tokio_runtime
看到这里就知道它是个new_current_thread单线程啦

actix_web::rt是个single-threaded runtime，线程数的问题最早在群里好像就说了

一样的效果，10s

  tokio::task::spawn_blocking(move || {
                        for i in mm {
                            info!("开始tokio发送进程 {:?}", i.device);
                            mqttjob::mqtt_pub(&i.cli.clone(), to.clone(), data.clone()).unwrap();
                        }
                    }).await.unwrap();

--
👇
dlhxzb: 试试spawn_blocking吧

整体是一个actix的框架

stackoverflow上面说用了actix_web::main就不需要tokio::main了

#[actix_web::main]
pub async fn server() -> std::io::Result<()> {
    //初始化数据库
    let host =  CONFIG_DATA["server"]["host"].as_str().unwrap();
    let port =  CONFIG_DATA["server"]["port"].as_u64().unwrap();
    match db::db_init().await {
        Err(e) => info!("{e}"),
        Ok(()) => (),
    };

但是主main是非异步的

fn main() {
    //初始化日志
    logs::logger::logger_init();
    //启动APP
    router::router::server().unwrap();
}

--
👇
dlhxzb: tokio runtime怎么初始化的？看着像线程池只有一个线程，两个spawn互斥执行

tokio runtime怎么初始化的？看着像线程池只有一个线程，两个spawn互斥执行