很长一段时间只要跟做后端或者数据平台的朋友聊到技术选型大家都有一种默契碰到消息量上来、系统被突发的流量打穿、或者多个服务之间需要解耦第一反应就是把 Kafka 拉进来。这个判断在传统后端里已经反复验证过了没什么争议。但最近这半年情况明显起了变化——越来越多的 AI 项目开始把 Kafka 当成标配甚至可以说Kafka 已经正式接入了 AI 这条赛道。我最近在帮团队做一套大模型本地部署的方案最初所有人的注意力都集中在 GPU 资源、模型量化、推理延迟这些听起来更AI的环节上。可真到上线压测才发现最先扛不住的根本不是模型而是数据链路。尤其是做 Agent 类应用和多路推理日志聚合的时候请求量一上来同步接口直接超时日志和事件散得到处都是排查问题和回溯数据都变得异常困难。我们最后是用 Kafka 把整条链路重新捋了一遍才把这套系统真正推到可以稳定运行的状态。这篇文章我打算不聊那些虚的模型评测就讲 Kafka 在 AI 场景里到底怎么用、为什么非它不可、从安装到工程落地怎么一步步跑通以及我在实际项目中踩过的那些消息延迟、消费堆积、重复消费的坑。如果你也在做 AI 应用或者正准备把大模型能力接进自己的业务系统这篇内容应该能帮你少走不少弯路。1. 为什么AI系统最先卡住的是数据管道从同步调用到事件流的范式转移先从我观察到的一个现象说起。很多刚开始尝试接 AI 能力的团队第一个版本的服务架构非常类似客户端发起请求后端服务拿着请求直接调用模型接口等模型推理完再把结果同步返回。这种模式在 demo 和低并发场景下完全没问题。但一旦进入真实业务问题就成串地冒出来。1.1 同步调用的天花板在哪里同步调用本质上是把请求的成功与否压在模型服务的响应时间上。大模型推理本身就是一件耗时的事情哪怕是一个轻量级模型单次推理几百毫秒也是常事。如果模型服务刚好在高峰时段出现抖动调用方就会直接超时。更麻烦的是AI 应用通常不是只有一路调用而是多个业务方同时要使用模型能力。举个例子我们当时接了一个多模态需求用户上传一张图片系统要同时做目标检测、OCR 文字识别、图像标签生成最后还要把所有结果写进日志系统做后续分析。如果这几路操作都是同步串行调用一次请求的完整耗时就是所有模型推理时间的累加用户体验可想而知。再加上请求量波动极大白天和晚上的流量能差出好几倍同步架构在流量突增时几乎必然会打爆。1.2 数据库直连为什么也扛不住对很多开发团队来说面对流量冲击最先想到的替代方案是引入一个队列但如果只是简单地把数据写到数据库表里当一个假队列同样会踩坑。比如直接把日志或推理请求写到 MySQL一是写入并发受数据库连接数和磁盘 I/O 限制二是这块的存储量增长极快AI 应用里的日志和中间数据跟普通业务数据不同你很难提前预测到底会积累多少。我见过一个项目用 Redis List 做任务队列入队出队确实快但一旦消费者崩了或者需要重新处理积压消息数据在内存里说没就没根本没有回溯能力。Redis 队列适合那些丢了也不影响大局的轻量任务但 AI 链路里很多中间结果是有业务价值的比如模型产出的向量、用户行为特征、推理前后的比对日志这些都需要持久化保存一段时间。Redis 在这个场景里显得过于健忘。1.3 Kafka在缓冲和重放上的独特价值Kafka 的价值恰恰体现在这里。它本质上是一个分布式提交日志消息落盘之后可以长期保存消费者不要求必须实时处理完。流量高峰时生产者把数据往 Kafka 里一写就返回消费端按自己的节奏慢慢处理这天然形成了一个削峰缓冲区。我们不需要再为高峰会不会打垮下游焦虑只需要盯住消费组的裸奔程度就可以。更关键的一点是重放能力。AI 场景里经常需要做效果对比或问题复盘。之前有一次我们发现某个模型产出的标签质量有问题但当时的结果已经写进了业务库数据和客户端请求的原始内容对不上导致排查非常痛苦。后来把生产端的原始数据全部接入 Kafka保留周期设置为 7 天出现任何问题都可以重新消费一遍原始数据去复现。这个能力是同步调用和普通队列都无法提供的。1.4 多消费者扩展的模式红利Kafka 还有一个隐藏优势同一个 Topic 的数据可以广播给多个不同的消费者组。同一个推理请求A 组拿去实时统计延迟指标B 组拿去落库做归档C 组拿去训练后续的模型。这在 AI 系统里非常实用因为同一个数据往往有多种不同的下游用途。如果不用 Kafka你得把同一份数据手动复制 N 份分别投递给不同的服务维护成本直接翻倍。所以从整体趋势来看AI 系统从同步调用转向事件驱动不是某个架构师拍脑袋想出来的而是数据规模和业务复杂度倒逼出来的结果。Kafka 作为这个范式转移里最关键的基础设施自然成了 AI 技术栈里绕不开的一环。2. 三条主线理清Kafka在AI落地中的具体位置聊清楚了 Kafka 的底层价值我们得看看它在实际的 AI 项目里到底能站在哪些位置上。我在做了几个 AI 相关项目之后把 Kafka 的落地场景归纳成了三条主线这三条主线基本覆盖了市面上大部分 AI 系统的接入方式。2.1 大模型本地部署时的请求管道与日志管道第一类场景是大模型本地化部署时的数据支撑。现在很多企业出于数据安全考虑选择把模型部署在自己的机房或者私有云里。本地部署之后你会面对一个现实问题模型服务本身的观测数据、请求日志、推理耗时、token 消耗都需要收集、存储和分析。这些数据如果走普通等日志系统直接写文件再定时采集时效性会比较差而且日志格式难以统一。更好的方式是把所有请求和响应数据都打到 Kafka下游接一个日志引擎做全文检索再接一个时序数据库做指标统计。这样模型服务本身的职责会非常单一——只负责推理并上报数据至于数据怎么用完全由下游消费者决定。我们当时做了个 AI 网关所有对本地模型的调用请求都会先通过网关网关在转发给模型之前和拿到结果之后各发一条消息到 Kafka。这样每一个请求的完整链路就都被记录下来了。后来面试官朋友还问起来过AI 网关怎么观测和审计我直接把这个设计逻辑讲了一遍他听完觉得这是很典型的 AI 平台层基础设施的做法。2.2 Agent事件流的异步驱动机制第二类场景是 AI Agent 的事件驱动架构。智能体应用跟传统的请求响应模型差异非常大一个复杂任务往往需要多个 Agent 协作完成。比如用户说帮我安排一下下周三的会议并预订会议室背后可能是意图识别 Agent、日历操作 Agent、会议室资源 Agent 三个模块在协作。如果让这些 Agent 同步地互相调用整个调用链路会变得极其复杂任何一个环节失败都可能导致整条链路的全局超时。把 Kafka 引入之后Agent 之间的通信模式就从函数调用变成了事件发布和订阅。意图识别 Agent 分析完用户请求后直接把日程安排任务事件写入 Kafka日历 Agent 订阅这个事件执行完操作后写入会议室预定请求事件会议室 Agent 再订阅处理。每一步都是异步的任何一步失败都可以重试而且整个过程有完整的日志记录。这种事件总线模式才是 Agent 应用在真实业务环境中能够稳定运行的关键。2.3 推理结果回流与向量数据持久化第三类场景是模型推理结果的数据回流。AI 系统不只是把模型结果返回给用户就结束了你往往需要把结果存下来用于后续的数据分析或者模型迭代。比如内容安全审核模型对每一段文本、每一张图片都会给出一个风险评分和命中的策略标签。这些结果不仅要在客户端进行拦截还要进入数据仓库供安全团队做规则优化。如果结果数据量不大直接用接口同步写入数仓也能跑但当数据量增长到每天几千万条时同步写就很容易拖垮数仓的写入性能。Kafka 作为缓冲层让推理结果先快速落盘数仓侧根据自己的写入能力匀速消费整体系统的吞吐量就不再受制于最慢的那一个环节了。2.4 AI场景中Kafka的典型角色对比为了让人更直观地理解我把这几类场景的差异整理成了一个简单的表场景方向消息内容典型Topic消费者角色核心诉求模型观测请求日志、推理耗时、token统计ai-gateway-logs日志系统、监控系统审计与回溯Agent协作任务事件、状态变更、工具调用结果agent-task-event各功能Agent异步解耦与重试结果回流推理结果、标签、向量数据model-inference-result数据仓库、特征存储削峰与持久化训练数据用户反馈、行为序列、标注任务training-data-pipeline模型训练平台数据积累与重放这个表对刚入门的同学会比较友好你可以在做架构设计时先去对号入座看自己的系统属于哪一种情况再决定 Kafka 应该接在哪里。3. 从安装到命令行生产消费半小时跑通Kafka全流程不管概念聊得多深刻最终还是要落到实际操作上。这一节我会完整带大家走一遍 Kafka 的环境搭建、Topic 管理、生产消费命令的使用以及一个很多人都会困惑的问题——生产消费命令启动一次会一直运行吗这里面的门道我当初也是踩了坑才搞明白的。3.1 本地开发环境最快的安装方式Kafka 安装通常有二进制包和 Docker 两种方式。如果你的本机已经装了 Docker用 Docker 方式是最省心的一条命令就能把服务拉起来。docker run -d --name kafka-p 9092:9092-e KAFKA_BROKER_ID1-e KAFKA_ZOOKEEPER_CONNECTzookeeper:2181-e KAFKA_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR1apache/kafka:3.7.0不过要注意官方镜像对 Zookeeper 也有依赖Kafka 3.x 虽然引入了 KRaft 模式可以摆脱 Zookeeper但我个人建议在本地快速体验时直接使用 docker-compose 把 Zookeeper 和 Kafka 一起启动这样兼容性最好不会因为某个细节没配好导致启动失败。如果你不想用 Docker直接去官网下载二进制包也是一种选择。下载完解压后先启动 Zookeeper再启动 Kafka Server。Windows 用户会用到kafka-server-start.bat这里要特别留意脚本参数里的配置文件路径很多新手的启动失败都出在路径问题上比如路径中包含中文或空格。我见过有人写kafka-server-start.bat d:/rk/zy/kafka/kafka_2.13-3.0.0/config/server.properties结果因为目录名称不对导致配置读取失败所以用二进制包时最好全局不要出现空格和中文路径。3.2 命令行创建Topic、生产消息、消费消息服务启动成功后先创建一个测试用的 Topic。Kafka 3.x 版本自带kafka-topics.sh脚本。创建一个名为ai-events的 Topic分区数为 3副本数为 1。bin/kafka-topics.sh --create--topic ai-events--partitions 3--replication-factor 1--bootstrap-server localhost:9092然后打开两个终端窗口一个跑生产者一个跑消费者。在终端 A 启动生产者控制台bin/kafka-console-producer.sh--topic ai-events--bootstrap-server localhost:9092启动后你会看到光标停留在新行上可以直接输入消息内容并按回车发送。这时候去终端 B 启动消费者bin/kafka-console-consumer.sh--topic ai-events--from-beginning--bootstrap-server localhost:9092如果你先启动了消费者再发送消息消费者会立即打印出消息内容。这个流程就是 Kafka 最核心的模型生产者写消息消费者读消息两者完全分离。3.3 生产消费命令启动一次会一直运行吗很多人在第一次执行生产者和消费者命令时都会问同一个问题启动一次之后命令会一直运行吗还是执行完就自动退出答案是会一直运行。Kafka 的生产者和消费者在设计上都是长驻进程。生产者启动后它会保持对 Kafka 集群的连接持续等待你在控制台输入新的消息。消费者更是如此它启动后会一直监听这个 Topic 上是否有新的消息进来。如果你想退出需要直接按 CtrlC 中断进程。注意如果在没有指定--from-beginning的情况下启动消费者新消费者只会收到它启动之后新产生的消息。这一点跟传统的消息队列消费者的语义不一样它默认不是把所有历史消息都重新读一遍而是从自己当前消费组记录的 offset 位置继续往后读。这个机制在后面的重复消费问题里还会提到。理解了这个长驻特性你就不会在写自动化脚本时踩坑了。比如有些同学写 CI 脚本执行完消费者命令后发现进程一直挂在那不退出导致整个流水线卡死这是很典型的入门问题。3.4 查看Topic中的数据与监控工具当你用消费者消费完消息之后还想回头看某个 Topic 里现在有多少条数据、各分区的消息积压情况怎么样这时候有几个常用的查询命令。查看 Topic 列表bin/kafka-topics.sh --list --bootstrap-server localhost:9092查看某个 Topic 的详细信息bin/kafka-topics.sh --describe--topic ai-events--bootstrap-server localhost:9092这个describe命令会输出当前 Topic 的分区数量、每个分区的 Leader、副本分布信息。如果想知道某个消费组当前的消费进度、以及积压了多少消息可以用kafka-consumer-groups.sh命令来查。bin/kafka-consumer-groups.sh--describe--group my-ai-consumer--bootstrap-server localhost:9092输出的表格里会有LAG一列这个值代表当前消费者落后的消息数是排查性能问题最重要的指标。除了命令行可视化工具也能大幅降低操作门槛。我在不同项目里用过几款这里做一个简单的对比工具名称安装方式主要功能适用场景Kafka UIDocker或二进制包Topic管理、消费组查看、消息预览开发环境快速调试Offset Explorer桌面客户端浏览Topic、查看分区offset、模拟收发消息本地可视化查询CMAKDocker或源码编译集群状态监控、分区重平衡操作集群运维管理夜莺/自研监控接入数据源指标监控、告警生产环境长期运行如果你只需要在本地看消息内容首选 Kafka UI 或 Offset Explorer它们在几分钟内就能配好足够帮你省去敲命令的麻烦。4. Spring Kafka工程化实战把AI推理日志改造成异步管道命令行只是熟悉概念的手段真实业务里你肯定还是要用代码去操作 Kafka。我用得最多的是 Spring Kafka因为 Java 后端生态里它最成熟配置方式清晰出问题的资料也最丰富。这一节我会带你完整写一个 AI 场景的示例把 AI 推理请求的日志异步发送到 Kafka并让消费端做处理。4.1 Spring Kafka的核心依赖加入在pom.xml中加入 Spring Kafka 依赖即可org.springframework.kafka spring-kafka版本跟随 Spring Boot 的 BOM 管理一般不需要自己指定。如果你的项目是 Maven 中央仓库直接拉项目也可以手动指定 2.9 或 3.x 版本核心 API 差异不大。4.2 Spring Kafka yaml配置详解接下来在application.yml里配置 Kafka 连接信息和序列化方式这是接入过程中最容易出错的地方。下面是一份可以实际用的配置spring: kafka: bootstrap-servers: localhost:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer acks: all retries: 3 properties: enable.idempotence: true consumer: group-id: ai-inference-log-group key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer auto-offset-reset: earliest enable-auto-commit: false properties: spring.json.trusted.packages: * max.poll.interval.ms: 300000 listener: type: batch有几个关键点要单独解释JsonSerializer/JsonDeserializer消息体默认使用 JSON 序列化发送和接收的对象必须能被 Jackson 正常序列化。spring.json.trusted.packages由于反序列化涉及类安全问题Spring Kafka 默认只信任部分包名下的类。如果不添加这个配置消费反序列化时会报UntrustedPackageException这是新手最容易踩的坑。你可以指定为你的项目包名如果图省事也可以像我一样在本地环境配成*生产环境建议收紧。enable-auto-commit: false关闭自动提交 offset结合手动提交机制来保证消息不丢。这一点对 AI 场景格外重要因为消费端经常要写数据库、调其他服务处理耗时远高于普通的日志消费。这份配置我实测下来是可以直接用的。如果你的项目中存在多个不同的消息类型可以考虑用不同的 Template 方法或者在消息头中带上类型信息避免反序列化混乱。4.3 生产者代码发送AI推理请求记录配置好之后我们写一个简单的生产者组件用来发送 AI 请求数据。先定义一个消息对象public class InferenceLog { private String requestId; private String modelName; private String prompt; private String result; private long latencyMs; private long timestamp; }然后注入KafkaTemplate来发送消息Service public class InferenceLogProducer {private static final String TOPIC ai-inference-log; private final KafkaTemplateString, Object kafkaTemplate; public InferenceLogProducer(KafkaTemplateString, Object kafkaTemplate) { this.kafkaTemplate kafkaTemplate; } public void sendLog(InferenceLog log) { ListenableFutureSendResultString, Object future kafkaTemplate.send(TOPIC, log.getRequestId(), log); future.addCallback( result - { // 发送成功的回调 }, ex - { // 发送失败可以记录日志并考虑重试 } ); }}注意这里我把消息的 key 设置成了requestId。这样同一个请求的日志会进入同一个分区保证分区内的消息顺序和请求产生顺序一致。如果你不关心局部顺序Key 可以直接传 null这样消息会按分区策略均匀分布吞吐会更高。做 AI 日志场景我建议按 requestId 做 Key方便后续按请求级别去链路追踪。4.4 消费者代码处理并落库消费者的核心逻辑是接收消息后写数据库或者调用别的服务做后续处理。这里是一个简单的批量消费示例Service public class InferenceLogConsumer {private static final Logger log LoggerFactory.getLogger(InferenceLogConsumer.class); KafkaListener(topics ai-inference-log, groupId ai-inference-log-group) public void onMessage(ListConsumerRecordString, InferenceLog records, Acknowledgment ack) { ListInferenceLog logs new ArrayList(); for (ConsumerRecordString, InferenceLog record : records) { logs.add(record.value()); } // 批量写入数据库减少单条写入开销 inferenceLogMapper.batchInsert(logs); // 手动提交 offset确保数据处理成功之后再告诉 Kafka 可以继续消费 ack.acknowledge(); }}这里有一个很容易踩的坑KafkaListener方法如果返回类型为 void它会在方法完成后自动提交 offset但是 Spring Kafka 对自动提交的时机和异常情况的处理有它自己的内部逻辑稍不注意就可能出现消息消费了但数据库还没写入就提交 offset 的情况。所以在 AI 处理链路中建议始终使用enable-auto-commit: false并在所有业务处理成功后才手动ack。4.5 序列化异常的几个实际处理经验使用 JSON 序列化时会遇到一个问题如果消费者端的实体类比生产者端多了一个字段或缺了一个字段反序列化时未必会报错但字段值可能为空。这个在 AI 场景下特别危险因为模型输出的数量、文本长度这些字段如果丢失下游统计分析就会出现偏差。我在实际项目中加了一个消息版本字段在反序列化后做校验不合法就直接投递到死信队列。Kafka 本身没有内置死信队列但我们可以在消费者里捕获异常把解析失败的消息单独发送到一个ai-inference-log-dlqTopic 里。这个 Topic 不需要太多分区主要用于后续人工修复和数据 review。5. 消息延迟高和Lag堆积一条完整的排查链路话说回实战。我们系统的 AI 推理日志接入 Kafka 之后运行了一段时间一切看起来都很正常。直到有一天大促活动流量上来监控群里突然有人贴出一张消费组 LAG 图积压量在肉眼可见地往上跳。这应该是所有用 Kafka 的团队都会遇到的经典问题消息延迟高、消费滞后。5.1 先把问题定义清楚当消息积压发生时第一步不是急着改代码而是确认 LAG 到底积在哪里。用之前的kafka-consumer-groups.sh --describe命令可以看到每个分区的CURRENT-OFFSET和LOG-END-OFFSET。如果所有分区的 LAG 都很大说明消费能力整体不足如果只是某一个分区 LAG 特别高其他分区都很健康那就要重点查看分区分配是否均匀。我们当时的现象是 6 个分区里 5 个消费正常1 个分区积压了几十万条消息。这种情况下最可能的问题有两个某个消费者实例挂掉了导致分区没有被接管或者某个 key 的数据量特别大导致单分区处理不过来。5.2 从监控到根因的完整排查步骤我按以下顺序做了排查这个顺序可以复用到大多数积压问题里先看消费组有多少个消费者实例。Kafka 的规则是一个分区只能被同一个消费组内的一个消费者实例消费如果消费者的数量大于分区数多余消费者会处于空闲状态。如果消费者实例数小于分区数就会有一个消费者同时消费多个分区。我第一件事是检查所有消费者实例是否都活着是否有无限重启的情况。再看分区分配情况。如果某个消费者机器性能比较差那它分到的分区消费速度就会明显慢于其他实例。然后观察消费端的业务逻辑耗时。我们的消费端拿到 AI 推理日志后会调用一个内部接口做文本敏感词分析。上线初期这个接口平均响应 20ms 左右大促期间接口出现毛刺响应时间涨到了 800ms 以上。消费速度立刻从每秒几千条掉到几百条。检查 Kafka Broker 自身的状态。磁盘 IO、网络带宽是不是达到瓶颈。如果 producer 端不断重试也会占用 broker 的资源。排查完之后根因基本清晰了大促期间下游敏感词服务响应退化导致消费端处理一条消息的时间大幅拉长。大量消息来不及消费LAG 自然越积越多。5.3 根因处理的三种手段积压问题的解决办法通常是三种组合第一种是最快的扩消费者实例。但有一个前提消费者数量不能超过分区总数。如果分区数是 6你加 10 个消费者也没用多出来的只会空转。所以在设计 Topic 分区数时就要考虑好未来的扩展空间。我们当时把分区数扩大到了 24然后临时多启动了两倍的消费者实例消费速度一下子提了上来。第二种是优化消费端处理逻辑。把敏感词分析的同步调用改成异步批量调用攒够 100 条或每隔 100ms 合并成一次批量请求发给下游服务显著减少了网络往返带来的开销。这一步对于所有 AI 场景都适用因为模型推理或文本分析服务的单条调用延迟都比较高合并请求是成本最低、效果最明显的优化。第三种是调整 Kafka 消费拉取参数。增加max.poll.records让消费者单次拉取更多的消息同时把max.poll.interval.ms拉长。这套参数它控制的是消费者在两次 poll 之间允许的最大空闲时间如果消费者的处理时间太长超过了这个阈值Kafka 就会认为消费者已经挂了触发 rebalance反而加剧问题。我当时把max.poll.records从 500 调到了 2000max.poll.interval.ms从默认的 3 分钟调到了 10 分钟配合批量处理逻辑整体消费速度提升非常明显。注意max.poll.records不是越大越好。如果你的消费者内存有限一次拉取太多消息可能导致内存溢出。调参之前先评估单条消息的大小和处理时需要的堆内存。5.4 积压之后重复消费的问题接踵而至积压处理完之后还有一个小尾巴就是重复消费。因为我们关闭了自动提交手动提交又是在批量处理完成后才执行。如果消费者在处理完一批消息之后、提交 offset 之前发生了重启这批消息会在重启后被重新消费一遍。这在 AI 日志场景下可以接受因为日志写入数据库支持幂等用 requestId 做唯一键就行。但如果你消费的是发短信、发邮件的任务就必须提前设计好幂等机制否则就会酿成事故。我见过好几个团队就是没有处理好这个问题导致线上重复发送了通知。所以一旦你的下游有非幂等操作务必在业务逻辑里加去重判断或者用消息内的唯一 ID 作为数据库的唯一索引。6. Kafka和RabbitMQ到底怎么选AI场景下的对比决策在 AI 项目里我几乎每一次跟团队讨论消息选型时都会有人问一句为什么不是 RabbitMQ确实如果看传统后端场景RabbitMQ 的普及率相当高它上手简单、功能丰富在任务分发场景里很灵活。但在 AI 链路上两者选择的逻辑区别非常明显。6.1 两者的核心差异点先看两者的定位差异。RabbitMQ 是传统的消息代理它的核心模型是交换器 队列 路由键支持非常复杂的路由规则比如按 topic 去匹配、按 header 去匹配。Kafka 是分布式提交日志它不强调复杂的路由而是把数据组织成有序的日志流让多个消费者各取所需。这个差异带来的直接结果就是吞吐量和扩展性完全不同。Kafka 的横向扩展能力远强于 RabbitMQ单集群支撑百万甚至千万级消息每秒都很常见。RabbitMQ 在单机模式下吞吐能到万级就不错了集群模式的扩展性和运维复杂度也比 Kafka 高不少。还有一个关键点是消息的留存策略。RabbitMQ 默认在消息被消费后即删除消息的生命周期很短。Kafka 则完全不同消息落盘后按照配置的保留时间或大小来保存不因消费而被删除。这意味着 Kafka 可以随时重放历史数据而 RabbitMQ 很难做到这一点。6.2 不同AI场景应该怎么决策从 AI 项目的实际链路来看我做了一个简单的决策表场景推荐方案理由AI网关日志采集Kafka数据量大需要长时间留存支持重放分析Agent事件总线Kafka事件顺序敏感多个Agent订阅同一事件流推理结果批量入库Kafka削峰填谷下游按能力消费数据可回溯轻量任务分发RabbitMQ消息量不大需要灵活路由和优先级控制即时消息通知RabbitMQ延迟更低消息确认机制成熟定时任务下发RabbitMQ支持死信队列和延迟队列开箱即用如果你做的系统是面向用户的实时任务触发比如用户提交了一个修复请求立刻派发给某个工人那 RabbitMQ 的灵活路由和消息确认会更顺手。但如果你处理的是 AI 推理流水线、模型日志、Agent 事件流这种数据密集型的链路Kafka 几乎是无可替代的。6.3 个人经验选型时需要留意的隐性成本最后说一点个人体会。选型时很多人只看吞吐量和功能却忽略了一个重要的隐性成本——运维复杂度。Kafka 的部署运维比 RabbitMQ 重很多它涉及 Zookeeper 或 KRaft 集群、分区副本管理、消费组状态管理。如果你们团队没有专职的基础设施人员小规模业务用 Kafka 可能有点过度设计。我个人的建议是如果单日消息量在几百万以下并且数据不需要长期保存RabbitMQ 完全够用。只有当数据量明显增长到千万甚至亿级或者对数据的回溯、重放能力有硬性要求时再果断上 Kafka。AI 项目里日志分析和训练数据回流通常都是这个量级所以我的默认答案往往就是 Kafka。7. 生产环境的监控、面试高频点与经验补充文章最后这一部分我梳理一些非常有实用价值的内容包括生产环境到底要监控哪些指标、面试里总被问到的几个问题怎么回答以及一些我多年实践下来觉得值得提醒的细节。7.1 生产环境必须盯住的四个核心指标Kafka 在生产环境的监控不能只盯着上下线状态要看真正能反映问题的数据。我日常最关注的有四个第一个是消费组 LAG。它是系统健康度的第一信号积压超过阈值就要触发告警。第二个是 Broker 的请求处理延迟和网络吞吐如果 broker 的 CPU 使用率持续超过 80%很可能已经成为了瓶颈。第三个是磁盘使用率Kafka 依赖磁盘存储一旦磁盘写满整个集群会进入只读状态。第四个是分区副本的健康状态如果某个分区的 ISR 列表中的副本数小于预期就说明有副本同步落后了可能会丢数据。针对这些指标用 Prometheus 抓取 Kafka 的 JMX 指标再配合 Grafana 展示是目前最主流的方式。很多云厂商的托管 Kafka 也会直接提供这些监控能力比自己搭省心不少。7.2 高频面试点Kafka能重复消费吗这个问题几乎每次面试都会被问到。答案是可以重复消费而且重复消费在 Kafka 里属于设计上的特性不是 bug。重复消费的根本原因是消费者在消费消息后会往 Kafka 提交 offset而 offset 代表的是我消费到什么位置了。如果消费者在处理完消息之后、提交 offset 之前宕机了重启后 Kafka 会认为你还没消费过这条消息于是会重新发送。除了宕机还有一个常见原因就是消费者在处理过程中抛了异常代码没有正确提交 offset下一次 poll 时又会拉到同一批消息。所以在真实的 AI 应用里一定要把消费端设计成幂等的也就是不管你处理多少次最终结果都是一致的。7.3 面试加分项分区数、消费者数与性能的关系另一个高频考察点是理解分区如何决定并发上限。Kafka 的分区是并行度的边界同一个消费组内一个分区只能被一个消费者实例消费所以消费者实例的最大并行数等于分区总数。如果你的 Topic 只有 3 个分区你启动 100 个消费者实例也最多只有 3 个能拿到分区其他 97 个全都空闲。这就解释了为什么在设计 Topic 时要预留足够的分区数。分区太少未来扩展消费者实例也提高不了消费速度分区太多又会增加 broker 的元数据管理开销和文件句柄占用。我个人的经验是初始分区数设置为预估峰值消费者数量的 2 到 3 倍比如你最忙时会启动 10 个消费者实例那分区数可以先设置成 24 或 30。这个量级在绝大多数业务场景下都是安全的。7.4 最后的几个实战心得这里分享几个我踩过很多次坑才总结出来的经验不一定在官方文档里看得到但确实很管用。第一发送消息时一定要设置合理的重试参数并且要开幂等。如果不开幂等生产者重试时可能会导致消息重复写入这对下游的统计和模型训练数据都是灾难。开了幂等之后Kafka 会通过序列号机制保证单分区内不重复代价可以忽略不计。第二消费者处理 AI 推理结果时强烈建议把消息的 key 设置为业务唯一 ID。这样就算出现重复消费你也可以在数据库层面用唯一索引去重。如果不设 key消息会随机分布在各个分区后续要做按请求聚合分析会非常头疼。第三不要只看 LAG 这一个指标做告警最好加一个消费速率的指标。有时候 LAG 没有涨但消费速率已经降下来了这时如果流量再涨一点就会瞬间积压提前发现速率下降比事后处理积压要从容得多。第四Kafka 接入 AI 之后你大概率会面临消息格式的演化问题。模型版本更新了输出结果里多加了字段旧消息还是老格式这种新老格式混杂的情况在高保真数据流场景里尤其明显。解决方案是在消息体里加入 schema 版本号或者直接用 Avro 之类的 schema 管理工具。如果只是快速验证那至少要在代码里做字段判空避免反序列化直接报错把整个消费线程搞挂。如果未来的 AI 应用对实时性要求进一步提高Flink Kafka 的组合会是很多 AI 平台的标配能够让模型实时消费 Kafka 里的数据流去做在线推理和特征计算。整体来看Kafka 在 AI 基础设施里站稳脚跟这件事已经不是某个公司的个别现象而是整个领域在数据链路层面达成的新共识。