求助用 tokio::task::JoinHandle 实现 stop 功能

林深好材发表于 2026-09-10 17:09

Tags：tokio，阻塞

struct DispatchTaskInner {
    stream: Arc<RecordStream>,
    flow_mode: FlowMode,
    config: DispatchConfig,
}

pub struct DispatchTask {
    inner: Arc<DispatchTaskInner>,
    joiner: Mutex<Option<JoinHandle<()>>>,
    token: CancellationToken,
    subs: Arc<Subscriptions>,
}

impl DispatchTask {
  pub fn start(&self) -> bool {
        let mut guard = self.joiner.lock();

        if guard.is_some() {
            return false;
        }

        let inner = Arc::clone(&self.inner);
        let subs = Arc::clone(&self.subs);
        let token = self.token.clone();

        debug!(stream=%inner.stream.name(), flow=%inner.flow_mode, "starting to dispatch");

        let joiner = tokio::spawn(async move {
            let mut dispatcher = make_dispatcher(subs.clone(), inner.config.clone());

            loop {
                let (records, more) = inner.stream.poll();

                for record in &records {
                    dispatcher.dispatch(record);
                }

                if !more {
                    debug!("no more record: {}", inner.stream.name());
                    break;
                }

                if token.is_cancelled() {
                    info!("dispatch loop exited: {}", inner.stream.name());
                    break;
                }
            }

            dispatcher.stop().await;

            debug!(stream=%inner.stream.name(), flow=%inner.flow_mode, "dispatch terminated");
        });

        *guard = Some(joiner);
        true
    }
}

pub async fn stop(&self) {
        let joiner = {
            let mut guard = self.joiner.lock();
            guard.take()
        };

        let Some(joiner) = joiner else {
            return;
        };

        self.token.cancel();

        debug!("dispatch stopping");
        let _ = joiner.await;
        debug!("dispatch stopped");
    }

调试时：

都已经看到 dispatch terminated 日志了，但是看不到 dispatch stopped 这日志，不理解。
类似的结构有几层，发现从其它让 tokio 跑几个任务，堵住的位置就不同，但是一样的操作，每次堵住的位置是一样的。

评论区

写评论

作者林深好材 2026-09-11 16:52

感觉找到了部分原因：

上游数据流的poll是同步阻塞方法，没有数据不返回；
这样的任务数角度较多，有 cpu core 个；
一旦进入，始终占用了cpu，内部没有出让cpu的细节；

作者林深好材 2026-09-11 16:46

保存 JoinerHandle，主要是有管理需求：动态的部署和撤销数据流

👇
ThinkCodeStudio: 我不理解你为什么这么用, 这段代码好复杂, 我是启动任务后就不管task了, 也不会保存JoinHandle, 就相当于一个脱离控制的后台线程, 我控制这个task用的都是tokio的管道和事件, 比如关闭task我会发一个Notify信号, 数据交互都用的是管道

似乎不响应结果。主要是我这里stop要 &mut，要清除保存它的Vec，放后面更顺手些。

👇

我是先await后关闭的, 不是先关闭后await, 我也不理解你为什么把await放的这么靠后

作者林深好材 2026-09-11 16:35

用的tokio的Mutex

--
👇
asuper: joiner: Mutex<Option<JoinHandle<()>>>,

你用的是std::sync::Mutex吧，这个不能在async里面用的

asuper 2026-09-11 10:44

joiner: Mutex<Option<JoinHandle<()>>>,

你用的是std::sync::Mutex吧，这个不能在async里面用的

ThinkCodeStudio 2026-09-11 09:24

我不理解你为什么这么用, 这段代码好复杂, 我是启动任务后就不管task了, 也不会保存JoinHandle, 就相当于一个脱离控制的后台线程, 我控制这个task用的都是tokio的管道和事件, 比如关闭task我会发一个Notify信号, 数据交互都用的是管道

pub async fn start(
    db: Database,
    ch: SchedulerChannel,
    shutdown: Arc<Notify>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        info!("startup!");
        let mut scheduler = Scheduler::init(&ch.response).await;
        'main_loop: loop {
            tokio::select! {
                cmd = ch.sched_rx.recv_async() => {
                    match cmd{
                        Ok(pkg) => {
                            scheduler.recv_handle(pkg, &db, &ch.response).await;
                        },
                        Err(RecvError::Disconnected) => {
                            warn!("closed");
                            break 'main_loop;
                        }
                    }
                }
                work_rps = ch.event_rx.recv_async() =>{
                    match work_rps{
                        Ok(event) => {
                            scheduler.event_handle(event, &ch.response).await;
                        },
                        Err(RecvError::Disconnected) => {
                            warn!("closed");
                            break 'main_loop;
                        }
                    }
                }
                _ = shutdown.notified() => {
                    info!("shutdown!");
                    break 'main_loop;
                }
            }
        }
        info!("exit!");
    })
}

我是先await后关闭的, 不是先关闭后await, 我也不理解你为什么把await放的这么靠后

1 共 5 条评论, 1 页