漫谈 AI Infra|消息队列在 AI 时代能做什么?(上)

多客科技 · 发表于 2025-12-25 14:54

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要登录才可以下载或查看，没有账号？注册

×

作者：微信文章

没有意外，随着模型规模的持续增长和应用场景的日益复杂，AI Infra 也自然的从"单体架构" -> "分布式架构"进行演进，例如：

然而这个从单体架构到分布式架构的升级，最核心的变化就是通过消息中间件让数据、模型、服务之间能够异步、可靠、松耦合地协同工作，从而构建可扩展、可维护、可演进的 AI 平台的基础设施。

Pulsar 作为消息中间件的中流砥柱，以其更鲜明的存算分离、云原生特性，发挥着更大的价值。

场景一【多模态】

让 Pulsar 直接吞进超大消息，

多模态训练“零”切片

传统的单模态模型，如自然语言处理（NLP）模型仅处理文本，计算机视觉（CV）模型仅处理图像，自动语音识别（ASR）模型仅处理音频，它们彼此独立。多模态AI旨在让机器能够像人类一样，通过融合和理解来自多种感官通道（如视觉、听觉、语言）的信息来进行感知、推理和交互。这个给多模态训练增加了不小的难度；

多模态AI系统处理的数据类型远超传统文本，包含了图像、视频、音频、3D 点云等大体积的非结构化数据。这些数据单个文件的大小就可能从几 MB 到几 GB 不等。其他的消息队列系统往往对单条消息的大小有严格限制（例如，Kafka 默认单条消息上限约 1 MB，调参后虽可放大，但需权衡副本同步压力），这迫使开发者在传输大文件时采用复杂的变通方案，如将文件存储在对象存储中，然后在消息中只传递文件的路径或 URL。

这种方式虽然可行，但增加了系统的复杂性和处理延迟，并且无法充分利用消息队列在数据流管理和处理方面的优势。

然而 Pulsar 原生支持超大消息体，即 Pulsar 的 chunk message，Pulsar 的 Chunk Message 是多模态训练的数据管道利器，它解决了大消息传输的完整性、顺序性、简化性三大问题，可显著降低多模态数据管道的工程负担，使开发者聚焦模型逻辑而非传输细节。

Pulsar Chunk Message（分块消息）是 Pulsar 提供的一种用于透明处理超大消息（>5MB）的机制。它允许生产者端自动将大消息拆分为多个小块传输，并在消费者端自动重组，业务层无需感知分块细节。

场景二【多模态】

用 Pulsar 把文本、图像、音频流绑定到一起

多模态 AI 需要处理和融合的数据类型极其多样化。系统需要同时处理文本（自然语言描述）、图像（像素矩阵）、音频（波形信号）、视频（图像序列和音频流的组合）等多种异构数据。

在许多场景中，不同模态的数据在时间上存在紧密的依赖关系。例如，在视频理解任务中，音频中的对话内容需要与视频中人物的口型、动作在时间上精确对齐；在自动驾驶场景中，激光雷达的点云数据、摄像头的图像数据和GPS的定位数据必须在同一时间点或时间窗口内进行融合，才能构建出对周围环境准确的感知。因此，消息中间件不仅要能传输数据，还需要提供机制来保证跨模态数据的时间同步和顺序性。

利用 Pulsar 的 Keyshare 消费模型，可以将同一 Key 的数据总是被路由分配到同一实例完成聚合，方案如下：

Key_Shared（键共享）是 Pulsar 的一种订阅模式，它在 Shared 模式的基础上增加了按消息 Key 的路由规则：相同 Key 的消息始终被分配到同一个消费者，而不同 Key 的消息可分布在多个消费者上并行处理，实现 Key 级别的有序性与负载均衡。

场景三【模型训练】

用好 Pulsar 压缩 Topic，

实现 Checkpoint 秒级断点续训

模型训练周期长、数据量大、集群规模大，出现中断的概率显著提高，且重启代价高昂；

所以通常会使用 Checkpoint 机制来加速恢复的过程，但保存 Checkpoint 耗时较高，若存储服务瞬时故障，写入请求直接丢失，导致训练状态丢失。

引入 Pulsar 作为中间层后，可以将异常数据跳过、Checkpoint 异步缓存、任务级重试等操作都交给 Pulsar 的特性来解决，方案如下：

Checkpoint 数据具有明显的历史消息无效的特性，如果发生积压时，只有最新的一条 Checkpoint 才有价值，这时可以使用 Pulsar 的压缩 Topic(Compaction Topic)，压缩 Topic 将 Checkpoint Topic 从日志流变为 KV 存储，仅保留每个 Key 的最新消息，自动清理历史版本，这样对比传统方案（扫描 S3 文件列表 → 排序 → 下载）需要耗时 3-5 分钟到直接接收最新 Key 的方案，耗时 <1S;

Compaction Topic 是 Pulsar 提供的一种基于消息 Key 的日志压缩机制，它会自动清理主题中每个 Key 的旧版本消息，仅保留最新版本，从而显著减少主题体积、加速消费回溯，适用于"只关心最终状态"的场景。

场景四【模型训练】

以 Pulsar 为“输油管”：

优化模型训练中的 GPU 饥饿

在大规模模型训练中，数据是驱动整个训练过程的“燃料”，特别是针对拥有数十亿甚至万亿级参数的深度学习模型，能高效且稳定的确保“燃料”能够持续、稳定地供应给计算引擎（如 GPU 集群）是关键所在。

训练这些庞然大物需要海量的训练数据，这些数据通常以 TB 甚至 PB 计。数据加载和预处理的速度直接决定了 GPU 这一昂贵计算资源的利用率。有数据表明 I/O 延迟使 GPU 每步等待数百毫秒，空闲率可达 30-50%。为了充分利用昂贵的计算资源，必须确保数据能够以足够快的速度被加载到每个计算节点的内存中，如果数据供给速度跟不上模型消耗数据的速度，就会有大量时间浪费在等待数据上，即所谓的“数据饥饿”问题。

历史的架构中，数据预处理模块与训练模块存在耦合的情况，然而耦合的模块可能相互影响从而降低了 GPU 的读取效率；

这种架构中，非常适合引入 Pulsar 在其中作为缓冲层，在数据平面预处理服务独立扩展，训练节点只专注消费，利用 Pulsar 的高吞吐特性，“喂养” GPU 的数据高速且稳定；

并且当 GPU 消费慢时，还可以利用 Pulsar 的背压机制，预处理消费时自动降低预取速率，避免 OOM，从而让整个链路更加健壮；

不止如此，还可以继续针对 Topic 的消费进行积压监控，如果出现积压，辅以 K8S 的 KEDA 机制 + Pulsar 的 Share 消费类型可以整个扩缩容过程更加平滑和稳定；

背压（Backpressure）是 Pulsar 中用于防止生产者过载消费者的流量控制机制。当消费者处理速度跟不上生产者发送速度时，系统通过多级反馈控制主动减缓上游生产速率，避免内存溢出、数据丢失和系统崩溃。

KEDA（Kubernetes Event-driven Autoscaling），是一种基于事件驱动的自动扩容解决方案，支持通过外部事件源动态调整 Pod 副本数；

Share 消费类型（Shared Subscription） Pulsar 的一种订阅模式，允许多个消费者绑定到同一个订阅名上，消息通过轮询机制分发给不同的消费者，每个消息仅会被分发给一个消费者，不保证消息顺序，适合高吞吐、无需顺序消费的场景。

本文介绍了 Pulsar 特性在多模态及模型训练中的应用；下篇，我们将继续解读 Pulsar 在智能体场景中的应用。欢迎大家通过评论区留言或加入社群，与我们共同交流～

Apache Pulsar 作为一个高性能、分布式的发布-订阅消息系统，正在全球范围内获得越来越多的关注和应用。如果你对分布式系统、消息队列或流处理感兴趣，欢迎加入我们！

Github:

https://github.com/apache/pulsar

账号		自动登录	找回密码
密码			注册

漫谈 AI Infra|消息队列在 AI 时代能做什么?(上)

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

浏览过的版块