企业官网建设流程全解析

在实际做AI基础设施的这段时间，我越来越发现：大家聊模型、聊算力、聊数据标注，但很少有人把“消息管道”当成一个正经的AI组件来设计。直到有一次，我们某个实时推理服务的日志和样本回传通道在流量高峰直接被打爆，我才痛下决心把Pulsar特性真正用到AI场景里，而不是把它当成一个普通的“Kafka替代品”挂着。

Pulsar这个名字在AI圈子里不算特别响，但它的很多底层特性，恰恰是AI流水线里最需要的东西。比如消息回溯、多租户隔离、队列加流的双模型消费，还有分层存储。这篇文章我会从我的实际使用经验出发，讲清楚Pulsar的哪些特性在AI场景下真的有用，怎么用，以及过程中踩过的坑。如果你是做AI平台、做训练数据管道、做推荐系统或在线推理服务的工程师，这篇内容大概率能帮你省掉几个月的试错时间。

1. 为什么AI场景需要重新审视消息系统

1.1 AI流水线里消息系统的真实位置

传统上，消息队列在业务系统里干的活很纯粹：削峰填谷、异步解耦、系统间通信。但AI场景里的数据流动完全不是这个逻辑，它有更复杂的节奏。

我梳理了一下，至少有这么几类任务离不开消息系统：

• 在线推理请求的异步化。比如图像生成、文本审核这类延迟敏感度不一的请求，前端先扔进队列，后端根据GPU资源慢慢消费。
• 训练样本回流。线上推理得到的预测结果，经过人工或自动标注后，要回流到样本池，再定期抽出来做增量训练。这一条链路对消息的“可回溯性”要求极高。
• 实时特征计算。推荐系统里，用户行为事件要实时入库并拼接特征，比如点击、曝光、停留时长，每一跳都可能经过消息总线。
• 模型版本发布通知。新模型上线后，需要通知所有相关服务预热缓存、切换流量，这种事件本身就是一条消息。

这些任务有个共同点：它们不只要“把消息送出去”，更要“把消息留住”、“能重放”、“能隔离”。而这恰恰是Pulsar和很多传统队列拉开差距的地方。

1.2 为什么不是Kafka，也不是Redis Stream

我接触过不少团队，一提到消息管道就直接上Kafka，或者图省事直接用Redis Stream。不是说这些东西不能用于AI场景，而是它们在某些关键点上的设计目标不同。

Kafka的架构是分区日志模型，每个分区的消息严格有序，但消费组之间是共享同一个日志的。它在吞吐量上很强，但多租户隔离、消息按需回溯、还有不同业务之间的资源隔离，都是偏后期才补上的能力。更现实的问题是，Kafka的存储和计算是绑在一起的，扩容缩容动不动要搬数据，这在AI实验频繁变动的场景下很折腾。

Redis Stream适合轻量级任务，比如几百条消息每秒的小流量场景。但一旦牵扯到数据回溯、多消费组独立游标、长时间保留，Redis的内存成本会让你怀疑人生。

Pulsar给我最直观的感受是：它把“存储”和“服务”拆开了。Broker只管消息的路由和缓存，数据真正落到BookKeeper里，再通过分层存储导到对象存储。这就让“消息保留多久”“能不能从三天前开始重新消费”变成了配置问题，而不是容量问题。这两种模型——队列模型和流模型——在一套系统里共存，正好匹配AI场景里“既要一次性消费，又要广播回放”的矛盾需求。

1.3 选型时最容易被忽视的三个底层能力

我在推动Pulsar落地时，团队里一个老工程师问了我一句话：你说的那些特性，Kafka也有啊，有什么本质区别？

我后来用三个例子回答他：

第一个，多租户。我们同时跑着好几个业务线，有的是推荐、有的是内容审核、有的是AIGC。每一条线的数据重要性和吞吐要求完全不一样。Pulsar在租户、命名空间维度直接做隔离，可以给某个业务线设置存储配额、消息保留策略、消费速率限制。另一个团队就算写了个死循环疯狂发消息，也影响不到你的业务。这个Kafka原生做起来很费劲。

第二个，消费模型的多样性。AI场景里，有些消息只能被一个消费者处理，比如训练任务的指令；有些消息要广播给多个消费者，比如特征更新通知；有些消息要按某个Key路由，让同一个用户的所有行为落到同一个消费者上。Pulsar的四种订阅类型可以在同一个主题下共存，每条消息按照订阅类型决定投递策略。

第三个，存算分离带来的弹性。训练任务有周期性，白天在线服务忙，晚上离线任务重。Pulsar的Broker是无状态的，要扩就加节点，不需要搬历史数据。这种弹性在云环境里特别舒服。

2. 几个真正改变AI工作方式的Pulsar核心特性

理论讲完，聊点实在的。我把过去半年在AI场景里高频用到的Pulsar特性挑出来，逐个说说它们解决的是什么问题。

2.1 双模型消费：队列与流在同一套系统共存

我记得自己第一次看Pulsar文档时，最大的冲击来自它的一语双关：Pulsar同时支持队列模型和流模型。

听不懂这句话的话，我用一个类比：

• 队列模型就像食堂打饭。一份菜出来，一个人端走，别人就吃不到。对应的是任务分发、请求异步处理。
• 流模型就像广播电台。谁都能听，多少人听都不会影响信号。对应的是数据分发、模型更新通知。

有些AI流水线是混合的。比如，线上推理结果要发给“模型监控系统”一份做实时告警，同时还要发给“样本采集系统”一份做训练数据回流。这两个消费方如果互相抢消息就有问题，应该各自拥有独立的消费进度。

Pulsar的Subscription机制把这件事处理得很干净。同一个主题，你既可以创建Shared订阅让一个消费组内多个消费者分摊消息，也可以创建Failover订阅让多消费者互为备份，或者创建Key_Shared订阅让相同Key的消息路由到同一个消费者。更重要的是，这些订阅之间互不干扰，各记各的游标。

这一点在AI工程里特别有用。比如同样的用户行为日志，推荐团队要用它做实时特征，风控团队要用它做离线分析，两者互不干扰。

2.2 多租户隔离：团队协作不互相踩踏

我在前面提到多租户，这里展开讲。

Pulsar的命名空间结构是这样的：tenant/namespace/topic。每个租户下面可以有多个命名空间，每个命名空间下可以有任意数量的主题。

实际操作里，我会给每个业务线建一个独立的租户。比如有个租户叫recommend，另一个叫safety。每个租户内部再按环境拆分命名空间：rec-prod、rec-exp、rec-dev。

隔离体现在哪呢？

首先是配额控制。我能给某个命名空间设置最大存储用量，超过就报警或者拒绝写入，免得某个测试任务把磁盘写满，连累生产链路。

其次是策略独立。不同命名空间可以设置不同的消息保留时间、不同的大小阈值、不同的备份策略。实验环境的Topic一天一清，生产环境的数据保留九十天，这在配置文件里就是一两个参数的事。

还有一点容易被忽略：Pulsar的认证和授权是体系化的。你可以给A团队只授予某个命名空间的读写权限，B团队只能读不能写。我们团队后来发展到十几个人共同维护一个集群，没有再发生过“有人误删了别人的Topic”这种事故。

2.3 分层存储：让训练数据“想追溯多久就多久”

AI场景里有一类需求很磨人：模型效果变差了，想拿三个月前的数据重新训练一版；或者审计要求保留原始日志，但又没人说得清到底要保存多久。

如果消息系统只支持固定容量的环形覆盖，这种事就干不了。Pulsar的分层存储是我用下来最惊喜的功能。

它的原理不复杂：最早写入的数据在BookKeeper里，超过某个阈值或者时间后，自动转移到对象存储中，比如廉价的云存储桶。消费端完全无感知，你照样可以用Reader或者新的消费订阅去读取几个月前的消息，就跟发生在昨天一样。

对AI团队来说，这意味着什么呢？

第一，训练样本的回放变得极其简单。不需要再额外搭一套数据同步管道，把消息从Topic复制到数据库。样本数据本身就在消息系统里躺着，要用的时候直接从头开始消费就行。

第二，存储成本大幅下降。热数据在高速存储里，冷数据沉淀到对象存储，单位成本能差一个数量级。我们当时算过一笔账，如果全量保留在本地盘，存储开销直接爆炸；分层存储一开，成本大概只有原来的五分之一。

第三，消息数据不再“一次性”。传统的消息队列消费完就删，分层存储模式下，数据默认就留着。哪怕线上已经消费完，后面再开一个订阅重新读一遍，数据依然在。这就是流和存储的结合体。

2.4 跨地域复制：多数据中心协同不再是噩梦

AI训练经常有跨地域的需求。比如我们在两个机房各有一批GPU资源，训练任务可能在A机房跑，也可能在B机房跑，训练数据的来源还不一样。

Pulsar的跨地域复制是我大力推荐的特性。它在集群之间异步复制消息，而且可以配置双向复制或者单向复制，容灾切换时也比较省心。

举个例子，我们在广州和上海各自部署了一套Pulsar集群，用的是双向复制配置。线上服务在北京和上海两地的请求都能写入就近的集群，消息会自动同步到另一个区域。训练任务不管在哪边跑，都能消费到全量的样本数据，不需要跨地域拉取原始日志。

跨地域复制还有一个好处：容灾。某个区域出问题了，消费者可以平滑切换到另一个区域的集群，消息不会丢，只是稍微有些延迟。这对在线推理链路来说，算是一个比较扎实的保险。

2.5 Schema与消息属性：让数据管道自带类型约束

AI工程师写出来的消息，格式飘忽不定是常态。今天这个字段叫user_id，明天改成uid，后天又在消息体里塞了个嵌套JSON。这类问题在联调阶段经常炸，而且很难排查。

Pulsar的Schema机制能把数据格式问题提前暴露出来。比如你可以给一个Topic指定消息体的Schema是JSON或者Protobuf，生产者写入时必须符合结构，消费者读取时拿到的是已经反序列化的对象。

更实用的是Schema兼容性策略。我一般会配置成FORWARD_TRANSITIVE，意思是新Schema必须兼容历史所有版本。比如你可以新增一个带默认值的字段，但不能删除已有字段，也不能修改字段类型。这样数据管道演进时，老数据和New消费者之间不会突然断掉。

消息属性也值得一提。Pulsar每条消息可以附带一组Key-Value属性，这些属性不影响消息体本身，但可以用来做路由、过滤、标记。比如我在样本回传时，会在属性里标上model_version、data_source、label_status，消费端直接通过属性做分流，不用解析整个消息体。

3. 一个AI推理服务的完整消息链路实操

前面聊的偏概念，接下来我拿一个真实落地的例子拆解：某图像生成服务的消息化改造。这个服务原本是同步调用，前端请求直接打到GPU推理服务上。高峰期经常因为排队超时被用户投诉，而且推理结果丢失后很难排查。

改造后，我们把整条链路搬到了Pulsar上。

3.1 场景设定与整体目标

目标有三个：

• 前端请求先写入队列，后端按GPU空闲度消费，服务不丢请求。
• 推理结果回传后，同步写入样本回流Topic，供后续训练使用。
• 同一用户的多张生成请求，尽可能路由到同一个后端实例，保证上下文连续。

整体链路如下：

1. 接入层把用户请求包装成消息，发送到img-gen/request主题。
2. 推理服务以Shared订阅消费请求，执行推理，将结果写回img-gen/result。
3. 推理服务同时把结果和原始请求通过img-gen/sample-feedback主题发送给样本系统。
4. 样本系统以独立订阅消费，完成去重、清洗、入库。

3.2 主题结构与分区规划

为了避免不同版本的消息混乱，我给每个主题都加了版本后缀，方便后续兼容演进。

分区数不是拍脑袋定的。我按预期的峰值吞吐来估算：单条请求消息平均2KB，峰值每秒500条，单分区可以稳定支撑每秒100条以上的吞吐。8个分区的情况下，即便单个消费者出现故障，剩下的分区也能扛住大部分流量。

这里要给个提醒：分区数决定了并行度上限，同时也影响管理复杂度。AI场景消息量波动大，别一上来就开几十个分区，后续格式化Topic或调整分区都要付出成本。

3.3 生产端配置：可靠性与时效性并重

生产端的Java客户端，我用的是pulsar-client。

核心配置里，我特别关注这几个参数：

PulsarClient.builder() .serviceUrl("pulsar://...") .operationTimeout(30, TimeUnit.SECONDS) .build(); Producer<String> producer = client.newProducer(Schema.STRING) .topic("persistent://img-gen/prod/request") .enableBatching(true) .batchingMaxPublishDelay(5, TimeUnit.MILLISECONDS) .batchingMaxMessages(1000) .blockIfQueueFull(true) .sendTimeout(10, TimeUnit.SECONDS) .create();

开批量发送是为了提吞吐。默认单条发送在低消息量下没问题，但高峰时，网络往返会成为瓶颈。batchingMaxPublishDelay设成5毫秒，既能攒足够多的消息，又不会让用户感知到明显延迟。

blockIfQueueFull(true)很重要。生产端消息堆积太快时，让它阻塞而不是丢消息。AI场景里丢一条请求，可能就是一次用户投诉。

消息体里，我放的是一个JSON字符串，包含task_id、user_id、prompt、params、timestamp。然后在properties里放model_version和priority，作为后续路由和过滤的线索。

实际发送时，如果是同一个用户的连续请求，我会设置MessageKey为user_id。这样后面用Key_Shared订阅时，同一用户的所有请求都会被同一个消费者处理，上下文就能连贯起来。对于图像生成服务来说，同一个用户的两次连续修图操作，如果落到不同实例，状态同步会很麻烦。

3.4 消费端配置：重试、超时与批量处理

消费端的陷阱比生产端多得多，尤其是AI推理服务的消费逻辑往往不是纯IO，而是吃CPU、吃GPU。

我在消费推理请求的时候，用的是Shared订阅。但为了同一用户的上下文不丢，我后来改用了Key_Shared。这里有一个实际的权衡：

• Shared：均衡分发，吞吐高，但同一个Key的消息可能落到不同消费者。
• Key_Shared：相同Key的消息必定落到同一个消费者，但消费者的数量变化会导致Key重新分配。

对于我们这个场景，稳定优先，所以我选了Key_Shared。

然后是最关键的Ack策略。推理服务接到消息后，执行一张图片生成可能要5到10秒。如果不设置Ack超时，消费者在消息处理失败后无法把这个消息交给别的消费者重试。

Consumer<String> consumer = client.newConsumer(Schema.STRING) .topic("persistent://img-gen/prod/request") .subscriptionName("infer-worker") .subscriptionType(SubscriptionType.Key_Shared) .ackTimeout(15, TimeUnit.SECONDS) .negativeAckRedeliveryDelay(2, TimeUnit.SECONDS) .receiverQueueSize(100) .subscribe();

这里有个参数组合要理解：ackTimeout是如果消费者干活超过15秒还没Ack，Broker就认为处理失败，会重新投递。但只要消费者还在正常处理，每条消息都及时Ack，这个超时就不会触发。

negativeAckRedeliveryDelay是我主动调用negativeAcknowledge(message)之后，消息多久再次投递。比如推理服务发现某条请求的参数不合法，但不想丢消息，就标记Nack，2秒后重新投递。

消费者处理批量消息也要注意。Pulsar客户端支持批量拉取，配合receiverQueueSize能在本地缓存一批消息，减少网络开销。但AI场景里，队列里攒太多消息，意味着内存里积压了大批请求。如果后端推理服务跟不上，这批消息会一直占着内存。所以receiverQueueSize不建议设太大，100左右是一个稳妥的起点，后续再按实际吞吐调整。

失败重试链路我也单独建了重试主题。推理服务消费时，遇到临时错误比如模型没加载完、GPU过热，我会判断重试次数。超过三次的，发到img-gen/request-dlq死信主题，后续人工排查。

3.5 用Pulsar Functions做轻量前置处理

在消息进入主链路之前，有些简单的清洗工作，我直接用Pulsar Functions做了，没有单独部署一套流处理服务。

举个例子，sample-feedback主题里的消息，偶尔有字段缺失或者格式错误。我用一个三行代码的Function做校验，非法消息直接丢弃到隔离主题，合法消息放行。

public class ValidateFeedback implements Function<String, String> { public String process(String input, Context context) { try { JSONObject obj = JSONObject.parseObject(input); if (obj.containsKey("task_id") && obj.containsKey("label")) { return input; } context.getLogger().info("invalid feedback: " + input); } catch (Exception e) { context.getLogger().error("parse failed", e); } return null; } }

Function返回null表示丢弃，返回字符串则继续传递。对于这类轻量ETL，Pulsar Functions比单独部署Flink或者Spark省事得多,不需要额外运维一个计算集群。

不过要注意，Pulsar Functions不适合复杂的状态计算、窗口聚合，那类需求还是交给Flink更合适。Functions的定位就是消息管道里的“小工具”。

4. AI场景里常见的坑与排查清单

实操过程中踩过的坑，整理成清单，每个都是真金白银换来的经验。

4.1 消费端背压不明显，服务悄悄雪崩

消息队列的一个错觉是：只要写入成功，就万事大吉。但AI推理服务的消费速率受GPU吞吐限制，如果消费者线程池太小，消息会在消费端堆积，表现为消费延迟飙升，但生产者端一切正常。

我排查过一起事故，现象是核心链路超时，但Pulsar指标看起来很正常。后来才发现是推理服务里的线程池被消息处理占满，Ack迟迟无法发出，Broker超时后重新投递，又加剧了堆积。

解决方案是给消费端指标加了一个关键监控：Consumer backlog和Ack超时次数。一旦backlog持续上涨，说明消费速率跟不上，立刻扩容消费者或者降级推理请求。

4.2 Key_Shared下消息处理乱序

Key_Shared订阅保证同一个Key的消息顺序投递，但前提是Key的哈希分配保持不变。如果消费者数量经常变化，比如某个消费者挂了再重启，或者扩容，会触发Key重分配，这时候可能短暂出现顺序错乱。

我们的踩坑记录是：一个消费者节点因为内存问题被容器调度掉了，重启以后，部分用户的消息被另一个消费者接走，结果生成请求的上下文对不上，出图风格突变。后来我们在消费逻辑里做了检查：如果消息里的timestamp早于上一条已处理消息，就标记为乱序，放入重试队列，而不是直接处理。

4.3 从三天前回溯消费，追不上实时写入

分层存储虽然让回溯消费成为可能，但如果你同时开着实时消费和回溯消费，实时游标在往前推进，回溯游标在从三天前追赶。如果回溯数据量太大，而消费者实例只有一个，回溯速度是有限的。

我们的经验是：回溯任务单独建一个订阅，不要和实时任务共用一个订阅。同时给回溯消费者开足够多的并行度，分区数如果不够，就新建一个专门的回溯主题，先把数据复制过去再消费。

4.4 租户配额被占满，写入直接失败

多租户隔离在保障稳定性同时，也带来一个副作用：如果某个团队的消息量超出配额，新消息会写入失败。

我记得有一回，一个实验团队写了个数据回刷脚本，用单线程把三个月的数据重新灌回Topic。两天时间，存储配额爆炸，生产环境的消息写入开始报错。虽然隔离本身没失效，但因为配额是按租户共享的，生产环境也遭了殃。

排查后确认，问题出在配额分配太粗放。改动方案是：给生产命名空间单独划分高配额，给实验命名空间设置更严格的保留策略和写入速率限制，并且在配额接近80%时就开始告警。这个告警阈值很关键，别等爆了再处理。

4.5 Schema演进踩坑：加了新字段，老数据全读不了

一次我在某个特征回传主题上加了新字段，没有配置兼容策略。上线后，消费端反序列化直接抛异常，因为在默认策略下，新增字段不被识别。

后来我把所有涉及跨团队协作的Topic，Schema兼容策略统一设为FORWARD_TRANSITIVE，并明确约定：新增字段必须有默认值；删除字段必须确认没有老消费者依赖；字段类型不允许变更。这个约定帮我们免掉了好几次联调事故。

5. 关于Pulsar和AI结合，我还在尝试的方向

Pulsar不是银弹，但它确实在AI场景里解决了不少实际问题。分享两个我个人还在尝试的方向，希望能给你一些启发。

第一个方向是用分层存储做训练数据版本管理。以往我们做模型回滚，要重新找数据、重新清洗、重新拼接，非常费劲。现在数据天然在Topic里待着，模型版本和Topic游标绑定。只要记录每个训练版本消费到了哪条消息，回滚时直接把消费游标指回去，就能复现当时的训练数据分布。这个思路在增量训练和线上模型回滚场景里潜力很大。

第二个方向，是把Pulsar Functions用于轻量级实时特征计算。比如用户行为事件进入Topic后，用一个Function做字段提取、概率过滤、实时统计量更新，再把结果发到特征存储。虽然复杂特征工程还是要靠Flink，但80%的简单逻辑都可以下沉到消息管道里完成，架构会清爽很多。

工具选型这件事，从来不是比拼谁的宣传更好听，而是看它在你最痛的场景里能不能站住脚。对我个人而言，Pulsar在AI场景的最大价值，是它把“消息”重新定义成了一种可重放、可隔离、可弹性伸缩的数据基础设施。这一点，和AI对数据的真实需求恰好对上了。