异步

运行中的服务持有哪些任务，一条消息怎样经过它们，以及关闭时按什么顺序停下它们。

以文字阅读本图

1. 应用运行时：Tokio，默认多线程: 默认 一个任务 完成所有步骤; workers(n) 一个循环任务 分发消息; threads(n) 一个投递任务 填充环形队列; n 个 worker 任务 各有一格收件箱; n 个环形队列 各有八个槽位; 受跟踪的任务：and_after、 钩子、retry_after 定时器
2. 已连接 Broker 的订阅流 → 默认 一个任务 完成所有步骤: poll_next
3. 已连接 Broker 的订阅流 → workers(n) 一个循环任务 分发消息
4. 已连接 Broker 的订阅流 → threads(n) 一个投递任务 填充环形队列
5. workers(n) 一个循环任务 分发消息 → n 个 worker 任务 各有一格收件箱: send
6. threads(n) 一个投递任务 填充环形队列 → n 个环形队列 各有八个槽位: push
7. n 个环形队列 各有八个槽位 → n 个系统线程 各自运行一个 Tokio 运行时: pop
8. 默认 一个任务 完成所有步骤 → 受跟踪的任务：and_after、 钩子、retry_after 定时器: spawn
9. n 个 worker 任务 各有一格收件箱 → 受跟踪的任务：and_after、 钩子、retry_after 定时器

服务运行在 Tokio 上。#[ruststream::app] 构建一个多线程运行时；flavor = "current_thread" 改用单线程运行时，worker_threads = n 设定线程数。

订阅者是消息的 Stream，背压来自轮询本身。默认的循环在上一条消息确认之后才轮询下一条。workers(n) 同时处理的消息不超过 n 条。workers(n, by_key) 把同一个键的所有消息交给同一个 worker，所以同一个键内保持顺序。threads(n) 最多预读 n × 9 条消息：每个线程手上一条，环形队列里八条。

专用线程在自己的单线程运行时上完成一条消息的全部处理，从解码到确认。处理器通过 Ctx<MainRuntime> 交给 MainRuntime 的工作，会回到应用运行时上执行。

一条消息

以文字阅读本图

1. 循环任务: 轮询 流; 轮询 下一条
2. 处理任务: 构建 上下文; 消费层; 解码; 处理器 函数体; 发布 应答; 确认
3. 稍后，受跟踪: 后续任务 与定时器
4. 轮询 流 → 构建 上下文
5. 构建 上下文 → 消费层
6. 消费层 → 解码
7. 解码 → 处理器 函数体
8. 处理器 函数体 → 发布 应答
9. 发布 应答 → 确认
10. 确认 → 轮询 下一条: 下一轮
11. 确认 → 后续任务 与定时器: 启动

默认情况下，处理任务就是循环任务本身；使用 workers(n) 时是一个 worker 任务；使用 threads(n) 时是一个专用线程。and_after 后续任务、确认之后的钩子和 retry_after 定时器在确认之后启动，作为应用跟踪的任务运行。它们最多执行一次：此时消息已经确认。

关闭

以文字阅读本图

1. 信号、fail-fast 或 RunningApp::shutdown → 运行 on_shutdown 钩子， 此时 Broker 仍在连接
2. 运行 on_shutdown 钩子， 此时 Broker 仍在连接 → 各循环不再 接收新消息
3. 各循环不再 接收新消息 → 手上的消息 处理完并确认
4. 手上的消息 处理完并确认 → 受跟踪的后续任务 和重试定时器执行完
5. 受跟踪的后续任务 和重试定时器执行完 → Broker 按注册的 相反顺序关闭
6. Broker 按注册的 相反顺序关闭 → 运行 after_shutdown 钩子， 此时 Broker 已关闭

关闭超时限制等待的时长。超时之后，仍在运行的循环会被中止，尚未完成的受跟踪任务则任其自行结束。不设超时时，关闭会等待全部任务完成。在专用线程上用普通 tokio::spawn 启动的任务不受跟踪，随线程一起结束。