使用tokio::select实现多线程关闭通知的问题

heliping 发表于 2022-11-07 17:38

求救大神！！

后台希望实现发送n条mqtt消息的发送，希望

1.发送完成后退出

2.中间如果收到信道传过来的cancel信号，停止发送

使用tokio::select实现

  for i in 0..100 {
        tokio::select! {
            _ = async{ sendmsg(i) }=> {
                info!("开始发送第{:?}条消息完成", i);
            },
            Ok(sig) = async{ c_s.recv() } =>  {
                        if sig == tcid {
                           break;
                        }
                }
        }
        sleep(Duration::from_secs(mqttinfo.interval.try_into().unwrap()));
    }

如上，正常情况希望执行sendmsg，发送mqtt消息

当信道有消息，并且发送的消息和本线程的id一致，就退出

问题：这个select后面的两条路，会随机执行

随机发了几条消息，然后pending了，可能在等channel的消息。

求大神看看这个程序哪里有问题。

评论区

写评论

作者 heliping 2022-11-08 10:55

多谢大神，使用select解决，tokio::select还是比较麻烦

 tokio::spawn(async move {
        for i in 0..mqttinfo.msg_number {
            select! {
                default() => {
                    sendmsg(i.clone(), tcid.clone())
                },
                recv(c_s) -> sig  =>  {
                    info!("收到删除信号 {:?}", sig);
                    if sig == Ok(tcid.clone()) {
                        info!("关闭发送 {:?}", sig);
                        break;
                    }
                    }
            }
            sleep(Duration::from_secs(mqttinfo.interval.try_into().unwrap()));
        }
    });

JLerxky 2022-11-07 19:03

select里的分支不用async包着，换成对应的异步方法

1 共 2 条评论, 1 页