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 跑几个任务，堵住的位置就不同，但是一样的操作，每次堵住的位置是一样的。