1. 项目概述1.1 核心需求解析最近我们团队做了一件挺有争议的事把 Kafka 正式接入 AI。立项会上有人觉得是赶时髦觉得我们一个搞实时数据管道的团队怎么突然要去碰大模型。但等整个改造做完之后我自己反倒觉得这可能是 Kafka 集群从“传输管道”变成“业务大脑神经”的关键一步。我先说清楚这次接入 AI 到底做了什么避免大家产生误解。我们不是把 Kafka 换掉也不是用什么 AI 替代 Kafka而是在现有的 Kafka 集群之上叠加了三层 AI 能力第一层用 AI 辅助 Kafka 集群运维把多年的监控经验沉淀成大模型可调用的诊断逻辑消费 lag 分析、分区异常、消费者组健康度检查这些事从“人肉翻监控”变成“对话式排查”。第二层把 Kafka 作为 AI 应用的事件总线多个 AI Agent 之间通过 topic 异步通信解耦各个模型服务之间的调用关系。第三层用 Kafka 承载 AI 数据管道把业务产生的日志、行为数据、文档内容统一采集进 topic再异步投喂给本地部署的大模型做分析、向量化、知识库构建。这套方案适合谁参考如果你手里已经有一套 Kafka 环境不管规模大小又正好想往 AI 方向走但不知道从哪切入那这篇文章应该能给你一条不算曲折的路线。即便你只是想把 Kafka 的运维水平往上提一档里面关于 lag 排查、监控指标、消费组治理的内容也是可以直接拿来用的。1.2 技术选型背景先交代一下我们现有的环境Kafka 3.5.0三节点集群日常峰值吞吐在 8 万条/秒左右主要承载订单事件、支付回调、用户行为日志三类数据。集群跑了两年多稳定性没问题但痛点也很明显消费 lag 一旦飙起来定位原因很费劲不同团队的同学各查各的最后还得我们去翻监控、看日志、问上下游一来一回半天就没了。这次“Kafka 接入 AI”的契机其实是因为我们团队被拉去支持一个新的 AI 项目。那边要做本地知识库问答数据来源是业务系统里零散的文档和操作日志需要先把数据清洗、格式化、聚合然后才能灌给模型。我们一看这不就是 Kafka 最擅长的事吗——数据管道、解耦、削峰、回放。再加上公司对数据出境有严格要求所有 AI 服务必须本地化部署所以整个方案从一开始就锁定在私有化环境本地大模型 Kafka 事件流 可视化监控平台。我负责的部分是整体架构设计、Kafka 侧的数据接入与消费治理以及和模型服务之间的接口联调。整个项目从立项到上线用了大概五周时间中间踩了不少坑尤其是 Kafka 消费端直接调用模型推理导致阻塞的问题卡了我们差不多一个礼拜。后面我会把这部分单独拿出来详细说这块的教训对任何想在 Kafka 上接 AI 的团队都有价值。2. 整体架构与设计思路2.1 为什么选择 Kafka 而不是直连调用最开始对接 AI 项目时业务方给了一个很“朴素”的方案让业务系统直接把文档和日志通过 HTTP 推送给模型服务。这个方案摆出来我就摇头——不是不能用而是扛不住真实的业务场景。举个例子我们接用户操作日志的时候来源系统有十几个每个系统的峰值时间还不一样有的凌晨跑批有的白天高峰。如果用 HTTP 直连每个来源都要自己去处理重试、限流、补偿不说模型服务那边的负载也会被冲得七上八下。高峰期打过来一堆请求模型来不及处理就超时调用方只能重试重试又加剧模型服务的压力典型的扇入风暴。Kafka 在这里的价值是削峰填谷和异步解耦。所有来源系统的数据先落 topic模型服务按照自己的处理能力去拉取消费处理不过来就积压等高峰期过了慢慢追。数据不会丢模型也不会被打垮。这就像快递柜——寄件人把包裹往柜子里一放就走了快递员按自己的节奏从柜子里取件派送两边互不干扰。所以架构的第一条原则所有 AI 需要的数据一律进 Kafka不允许业务系统绕过 Kafka 直连模型服务。这不是为了炫技而是为了系统稳定性。2.2 三条 AI 接入路线的拆解这次接入我们共规划了三条具体的落地路线每条路线的目标和实现方式都不一样我挨个说清楚。第一条是 AI 辅助运维代号“Kafka 助手”。这个东西解决的是运维痛点。以前消费 lag 高的时候我们的排查路径是先看监控图确认是哪个消费组、哪个分区 lag 在涨然后去查消费者日志看有没有报错再看下游接口耗时最后定位是消费慢还是生产端炸了。整个过程手动操作平均耗时四十分钟到一个小时。现在我们把监控指标全部汇聚到一个内部诊断 topic由本地大模型周期性消费并生成诊断结论遇到异常时直接输出“lag 异常原因分析”报告包括嫌疑点、关联指标、处理建议。虽然还不能完全自动化处理故障但至少能把定位时间压缩到五分钟以内这对我们来说已经是质变了。第二条是 AI 应用事件总线代号“Agent 消息干线”。我们接入了两个 AI Agent一个负责意图识别一个负责执行工单生成。这两个 Agent 之间的通信没有走 HTTP 轮询而是通过 Kafka topic 异步流转。意图识别的结果发到一个 topic工单生成模块订阅这个 topic拿到结果后进行处理再把结果发到下一个 topic。好处是两边可以独立扩缩容意图识别服务压力大了就多开几个消费者工单生成服务慢也不会阻塞上游。第三条是 AI 数据管道代号“知识燃料”。这是给本地知识库用的。业务文档、操作手册、日志模板通过 Kafka Connect 或者自研生产者写入原始数据 topic经过清洗、格式化、切片处理后由向量化消费者拉取并调用本地模型生成向量最后写入向量数据库。RAG检索增强生成的效果好坏很大程度取决于数据管道的稳定性和实时性Kafka 在这里保证了数据从产生到进入知识库的路程是可追踪、可回放的。2.3 可视化监控体系同步升级Kafka 接入 AI 之后我们顺手把 Kafka 的可视化监控也升级了一遍。以前用的是自研的简易监控页面只展示集群流量和 topic 积压量粒度太粗。这次换成了 AKHQ 加 Prometheus 加 Grafana 的组合。AKHQ 用于日常管理界面直观能查看 topic 列表、消息内容、消费组状态还能直接查看 Kafka Connect 的任务执行情况。工欲善其事必先利其器。以前排查消息格式问题还要上服务器用命令行消费现在直接在 AKHQ 里看消息体效率提升非常明显。Prometheus 负责采集 Kafka 的 JMX 指标和消费者 lag 指标Grafana 负责展示大盘。我们重点盯五个指标消息吞吐量、消费 lag、分区 leader 分布、请求处理耗时、网络吞吐。这五个指标直接对应了 Kafka 集群的健康状况也是后面 AI 辅助诊断的基础数据来源。3. 核心环节实操与细节实现3.1 Kafka 接入 AI 的 topic 设计规范topic 设计在 Kafka 架构里是最基础也最容易翻车的一环。这次接入 AI我们重新梳理了 topic 的命名和分区策略经验不多但很实用。首先是命名规范。格式统一为“域.数据类型.用途”比如ai.feedback.diagnosis、ai.log.rawdata、ai.agent.intent。域表示业务归属数据类型表示消息的性质用途表示这个 topic 在管道里充当什么角色。这套命名看起来简单但对后面做权限控制和监控告警非常有帮助正则匹配的时候一目了然。其次是分区数设置。Kafka 的分区数是并发度的上限也是消息顺序性的边界。AI 数据管道里的 topic分区数和模型服务的消费者并发数保持对应关系。比如向量化服务我计划开 3 个消费者实例那 topic 分区数就设 6 个预留一倍余量。如果分区数太少消费者实例再多也没用多余的实例闲着。如果分区数太多一个消费者要处理多个分区单分区消息量不大时问题不大但全局 rebalance 的开销会变大。再次是消息格式统一。我们所有 topic 的消息体统一使用 JSON并且约定必须包含三个公共字段event_id、event_time、source_type。event_id是全局唯一的消息 ID消费者做幂等处理的时候要用event_time是业务事件的发生时间而不是 Kafka 的写入时间做延迟分析和数据回放的时候这个字段不可缺失source_type标注消息的来源系统。这三个字段在 AI 辅助诊断的时候也很有用——模型可以根据消息的元数据判断数据链路中哪个环节出了问题。3.2 本地大模型的部署与 Kafka 对接参数AI 能力的核心是模型服务。考虑到数据合规和网络隔离要求我们选择了本地部署方案用 Ollama 跑开源模型。模型选型上我们对比过两个方案Qwen2.5-7B-Instruct 和 Qwen2.5-14B-Instruct。7B 模型的优势是推理速度快对机器的要求低普通双卡 24G 显存的机器就能跑得动单条消息的推理延迟在 300 到 800 毫秒之间。14B 模型的优势是理解能力和生成质量更好但推理速度明显下降延迟变成 1.5 到 3 秒对硬件的要求也更高。经过两轮测试我们最终选择了 7B 模型作为主力原因很简单Kafka 消息体通常比较短任务大多是分类、抽取、总结7B 模型的精度完全够用而且消费端的瓶颈在推理速度越快越好。Ollama 的部署配置有几个参数值得单独拿出来讲。默认配置下Ollama 会限制并发请求数如果 Kafka 消费者一次性拉取大量消息并发的调用模型会出现大量请求排队甚至超时。我们调整后的配置如下# 并发处理线程数 OLLAMA_NUM_PARALLEL4 # 最大加载模型数量 OLLAMA_MAX_LOADED_MODELS1 # 模型常驻内存时间单位秒 OLLAMA_KEEP_ALIVE-1OLLAMA_NUM_PARALLEL4表示同一个模型可以同时处理 4 个请求这个值不能盲目调大因为并发数越高单请求的响应时间反而会变长——GPU 的计算资源是固定的并发争抢会让每个请求都变慢。OLLAMA_KEEP_ALIVE-1是让模型常驻显存避免每次请求都重新加载模型。如果模型频繁被换入换出光加载模型的时间就够让消费 lag 涨到天文数字。3.3 Kafka 消息消费与 AI 分析的对接实现核心链路是 Kafka 消费者从 topic 拉取消息调用本地模型进行分析然后把结果写回结果 topic。这一段是整个项目里最需要精细打磨的部分我直接贴我们生产环境的核心代码。KafkaListener(topics ai.log.rawdata, groupId ai-analysis-group, containerFactory kafkaListenerContainerFactory) public void handleLogMessage(ConsumerRecordString, String record) { // 1. 解析消息体 LogMessage message JSON.parseObject(record.value(), LogMessage.class); // 2. 幂等校验防止重复消费 if (idempotentService.isProcessed(message.getEventId())) { return; } // 3. 调用本地 Ollama 模型分析超时控制是关键 String prompt buildAnalysisPrompt(message); String analysisResult ollamaClient.chat(qwen2.5:7b, prompt, 10); if (analysisResult null) { // 分析超时抛出异常交给 Kafka 重试机制处理 throw new RuntimeException(model analysis timeout); } // 4. 结果写入结果 topic AnalysisResult result new AnalysisResult(message.getEventId(), analysisResult); kafkaTemplate.send(ai.feedback.diagnosis, result.getEventId(), JSON.toJSONString(result)); // 5. 记录处理状态 idempotentService.markProcessed(message.getEventId()); }这段代码里有两个细节特别重要。第一个是 ollamaClient.chat 的超时参数我们设的是 10 秒。模型推理在并发高的时候会变慢给 10 秒是合理的但如果超过 10 秒还拿不到结果大概率是模型服务挂了或者队列堵死了再等下去没有意义直接抛异常让 Kafka 消费者触发重试更合理。第二个是幂等处理Kafka 默认的消费语义是至少一次消费者挂了重启后很可能会重复消费到同一条消息如果没有幂等保护同一条日志会被 AI 重复分析两次浪费算力不说结果 topic 里还会出现重复数据。配套的 Spring Kafka 配置如下spring: kafka: bootstrap-servers: kafka-01:9092,kafka-02:9092,kafka-03:9092 consumer: group-id: ai-analysis-group enable-auto-commit: false auto-offset-reset: earliest max-poll-records: 50 fetch-max-wait: 2000 listener: ack-mode: manual concurrency: 6 producer: acks: all retries: 3这里我重点解释三个配置项。enable-auto-commit: false关闭自动提交位移改为手动提交保证消息处理成功后才提交偏移量防止消息丢失。max-poll-records: 50控制一次 poll 拉取的最大消息数这个值不能设太大因为每条消息都要调用模型推理拉太多处理不完反而会触发消费者再平衡。concurrency: 6是消费者线程数我们集群一共两个消费者实例每个实例 3 条线程和 topic 分区数匹配。3.4 消息回流与可视化连接器验证数据经过 AI 分析之后结果回流到ai.feedback.diagnosistopic供 Grafana 面板展示和后续诊断使用。这里额外说一个 AKHQ 的实用操作如何在 AKHQ 中查看 Kafka Connect 任务的执行状态。在 AKHQ 的界面里左侧导航栏先看“Connect”专区主页面会显示所有已经注册的 Connect 连接器列表。每一个连接器条目上都有运行状态绿色表示正常运行红色表示任务失败。点进具体的连接器之后能看到更细致的任务分解信息包括每个任务的运行节点、配置参数、错误日志和偏移量提交记录。重点是看“Tasks”标签如果任务的运行状态是“FAILED”点开任务的详细页面AKHQ 会直接展示当前轮次的异常堆栈这个信息在排查数据管道问题的时候利用率极高。进入生产环境的第三天我们就靠 AKHQ 定位了一起向量化任务的连续失败问题。提示信息和数据源字段格式有关模型消费消息时对日期字段的格式有严格要求AKHQ 的任务日志里打出了完整的报错堆栈我们顺着堆栈定位到是数据格式不兼容改了配置后任务马上恢复前后不到二十分钟。放在以前这种问题少说要摸索两三个小时。4. 常见问题与排查思路实录4.1 lag 持续上涨从哪几个维度排查Kafka 接入 AI 之后最常见的故障就是消费 lag 上涨。AI 分析类的消费者和普通消费者不一样它会调用模型推理单条消息的处理时间可能从几毫秒变成几百毫秒甚至几秒lag 稍微涨一下其实正常。但如果是持续上涨、涨到告警线就说明消费能力跟不上生产速度了需要及时介入。我建议的排查顺序是先看模型服务再看消费者线程最后看分区分配。第一模型服务是不是慢了。直接看 Ollama 的推理日志和 GPU 利用率。如果 GPU 满载平均推理耗时翻倍那问题在模型端。常见诱因是并发打满或者有特别长的消息导致推理时间异常增加。第二步看消费者线程是不是阻塞了。用jstack导出消费者实例的线程堆栈如果发现大量线程卡在 HTTP 调用的等待响应上那就是下游模型服务响应太慢把 Kafka 消费线程拖死了。第三步看分区分配有没有倾斜。用命令查看消费组的分区分配情况kafka-consumer-groups.sh --bootstrap-server kafka-01:9092 --describe --group ai-analysis-group这条命令的输出里每个分区的 CURRENT-OFFSET 和 LOG-END-OFFSET 的差就是 lagLAG 列数字。重点看是不是某一个分区的 LAG 特别大其他分区都很正常。如果是说明消息分布不均匀某个分区里可能堆积了大量长文本消息每条都要花更长的推理时间。我们线上就出过这个问题用户操作日志里混入了一批导出的 PDF 文本单条消息体有十几 KB模型处理一条要将近 5 秒而普通日志消息只要 300 毫秒。尾部消息把其中一个分区的消费速度拖慢导致那个分区 lag 持续增长。解决办法是写一个消息分发逻辑按消息体大小区分投递到不同的 topic长文本走单独的慢消费通道避免堵住普通消息的处理。4.2 模型推理超时导致消息消费阻塞的治理这是我们这次接入遇到的最大坑单独拿出来讲。一开始我们直接把模型调用放在 Kafka 消费者的消息处理方法里问题马上就暴露了消费线程被模型推理阻塞max-poll-records设成 50一次拉取 50 条消息每条模型推理 2 秒一轮处理就是 100 秒。Kafka 消费者的max.poll.interval.ms默认是 5 分钟虽然 100 秒还没超限但线程池的并发度和处理效率完全达不到预期lag 蹭蹭往上涨不说整个消费组的心跳都有问题。后来我们改成了异步化方案把模型调用从消费线程里拿出去丢给单独的线程池处理KafkaListener(topics ai.log.rawdata, groupId ai-analysis-group, containerFactory kafkaListenerContainerFactory) public void handleLogMessage(ConsumerRecordString, String record) { // 异步提交任务消费线程立刻返回不阻塞 analysisExecutor.submit(() - { try { LogMessage message JSON.parseObject(record.value(), LogMessage.class); String prompt buildAnalysisPrompt(message); String analysisResult ollamaClient.chat(qwen2.5:7b, prompt, 10); if (analysisResult ! null) { kafkaTemplate.send(ai.feedback.diagnosis, message.getEventId(), analysisResult); } } catch (Exception e) { log.error(异步分析消息失败: {}, record.value(), e); } }); }异步化之后Kafka 消费线程不再阻塞一次拉取 50 条消息很快就执行完lag 问题大幅缓解。但随之而来一个新的问题手动提交 offset 的时机变了。消费线程已经把消息丢进线程池了如果用消费线程执行完就提交 offset线程池里的任务可能还没执行完一旦消费者进程崩溃那些还没处理的消息就丢了而且不满足我们“至少一次”的可靠性要求。权衡之下我们决定在异步任务完成之后单独提交 offset。这个提交粒度更细但实现复杂度更高需要对 Kafka Acknowledgment 做线程安全的包装。我们最终采用了带状态追踪的提交方案每条消息处理完成后记录 offset由一个定时任务批量提交已经完成处理的 offset。这样既保证消息不丢也不影响消费线程的处理速度。4.3 重复消费和消息顺序性AI 应用场景下要小心Kafka 接入 AI 之后业务方问得最多的两个问题是Kafka 能重复消费吗能保证消息顺序吗这两个问题的答案其实都不复杂但在 AI 场景下要注意的地方更多。重复消费的问题答案是能。Kafka 的消费语义默认是至少一次消费端处理完消息之后如果还没提交 offset 就崩溃恢复后会从上次提交的位置重新消费这条消息就被处理了两遍。在普通场景下重复消费的代价只是多算一次问题不大。但在 AI 场景下重复消费意味着同一份数据被重复调用模型白花算力和钱还有可能在结果 topic 里产生冗余数据。解决办法就是幂等按消息 ID 判重。我在前面的代码示例里已经实现了这里再强调一遍event_id必须在消息进入 Kafka 之前就生成好消费者在处理之前查一次去重表处理完再标记一次两步缺一不可。消息顺序性问题在 AI 场景下更隐蔽。Kafka 的分区机制保证的是单个分区内的消息有序如果 AI 任务依赖消息之间的先后关系比如先有“开始训练”事件、后有“训练完成”事件这两个事件必须落在同一个分区并且由同一个消费者线程处理顺序才有保障。我们在工单生成链路里就踩过这个坑。意图识别 Agent 发出的事件和后续状态更新事件因为 key 设计得不够精细被 hash 到了不同的分区导致状态更新先被消费、意图识别结果后到生成出来的工单内容错乱。后面我们把消息 key 统一改成业务单号保证同一个业务单号的所有事件都进入同一个分区顺序问题就解决了。4.4 本地模型显存不足的量化压缩方案本地部署大模型显存永远是不够用的。我们最初在 24G 显存的机器上跑 Qwen2.5-7B 的 FP16 版本一个模型就占掉 14G 显存再叠加并发负载高峰期偶尔会报显存溢出。解决办法是做模型量化用 4-bit 量化版本替代 FP16。Ollama 对量化的支持比较友好直接拉到带q4_K_M后缀的模型即可。这个量化版本的模型大小从 15G 降到了 4.4G显存占用大幅下降。代价是推理精度会有轻微下降我们测试了一些文本分类和实体抽取任务精度从 92% 降到了 89%在可接受范围内。如果业务方对精度要求更高可以考虑 8-bit 量化版本精度损失更小但显存占用会多 2 到 3G。除了量化显存不够还可以调整 Ollama 的并发参数。把OLLAMA_NUM_PARALLEL从 4 降到 2单次加载的模型副本少了每个副本能分到的显存就宽裕了。处理并发高的时候宁可排队也不要把模型服务拖死。4.5 高频面试题Kafka 与 RabbitMQ 的本质区别这个题不在项目排期里但是项目过程中几乎逢人就问。既然 Kafka 已经接入 AI 了我顺带把这两个消息中间件的本质区别说说不管是用到消息选型、还是准备面试都省得再翻资料。Kafka 是为流处理和大数据吞吐设计的。它的核心抽象是日志消息按分区顺序追加消费者通过维护 offset 来决定从哪里继续消费数据有保留期可以重复消费。吞吐量高水平扩展容易同时提供强大的消息回溯能力。RabbitMQ 是面向业务系统消息通信设计的。它的核心抽象是队列消息被某个消费者取走之后就从队列中删除不存在“重复消费同一份数据”的概念。它的优势是消息路由灵活、功能复杂、延迟低适合业务系统之间的点对点任务分发。放到本次 AI 接入的场景里看就很清晰了。AI 数据管道需要消息能被多个服务消费、能回放、能根据 consumer group 管理不同消费者的进度这是 RabbitMQ 做起来很吃力的Kafka 则是原生的能力。而 AI Agent 之间如果只是简单的请求-响应调用没有回溯需求那 RabbitMQ 也是合适的选择。选型从来不是哪个好哪个坏而是看场景需要什么能力。5. 项目落地经验与后续扩展思路5.1 从这次接入里提炼的几条实战建议整个项目做下来我最想强调的是“渐进式接入”。Kafka 接 AI 不是一个革命性的改动而是一个逐步叠加的过程。我们第一步只是接了一个 AI 辅助消费分析模拟人工查看 lag 和异常第二步才把 AI Agent 的通信迁到 Kafka 上第三步才上数据管道。每走一步都有回退方案出问题能立刻切回旧方案风险完全可控。如果你也要做类似的改造我的建议是先把监控做好再碰 AI。没有完善的 Kafka 监控体系接入了 AI 也只是黑盒套黑盒出了问题根本无从定位。监控、日志、链路追踪这三件事必须在接入 AI 之前就做扎实。AKHQ 和 Prometheus 加 Grafana 这套组合小规模集群完全够用了先搭起来后面再扩展。另外一个教训是AI 服务必须独立于 Kafka 消费者部署。一开始我们把 Ollama 和消费者进程部署在同一台机器上模型推理吃满 CPU 的时候消费者进程的 GC 都被拖慢了弄得两头都崩。后来把模型服务拆到单独的 GPU 机器上消费者实例只做数据搬运和调度问题才解决。资源的物理隔离在 AI 接入工作中不是可选项是必选项。5.2 关于 Agent 消息通信的扩展想法第二部分的 Agent 消息干线目前还只是两个 Agent 之间互通消息。后续我计划把更多的 AI 能力接入这个总线比如多模态分析、定时任务调度、用户意图跟踪等。Kafka 作为 Agent 通信底座的好处在于Agent 之间的所有通信都有日志可以审计、可以回放、可以离线分析。这在 AI 越来越深入业务核心的背景下是非常有价值的合规能力。出了问题打开 topic 就能看到完整的对话链和任务流转记录定位责任方非常方便。具体到实现上我准备给 Agent 消息增加消息协议版本号和回复关联 ID类似 HTTP 的 request-id 机制。这样多个 Agent 之间的异步链路就可以串联起来从发起方到处理方到最终结果一条链路完整可追踪。后续还能把追踪数据直接投给 AI 分析模型做 Agent 运行质量评估。5.3 Kafka 监控与 AI 分析联动的优化方向现在我们的 AI 辅助诊断是半自动的数据先进诊断 topic模型消费后生成结论人工再去看结论决定是否要处理。下一步的优化方向是做成自动闭环AI 诊断出消费 lag 异常的原因之后直接触发预设的处理动作比如动态调整消费者线程数、通知下游系统降级、对超时任务做补偿处理。这个方向的实现路径我们已经在规划了。核心是给 AI 诊断模块增加“动作输出接口”模型不再只输出文本分析而是输出结构化的处理建议比如“增加消费者线程”“暂停消费非关键消息”“通知上游限流”。这些建议经过规则引擎校验后自动执行只有在风险较高时才转人工审批。说白了就是让 Kafka 的运维从“发现问题-人肉分析-手动处理”变成“自动发现问题-AI分析-自动执行预案”人只需要处理那些 AI 判断不了的新问题。以上是我这次把 Kafka 接入 AI 的全部实操记录。踩过的坑都是真金白银换来的如果你也在做类似的事情希望这些内容能帮你少走一点弯路。当前阶段Kafka 往 AI 方向靠拢已经是大势所趋但不管外部怎么炒概念落地的时候还是要回归到消息系统的根本数据别丢、消费不被堵、链路能追踪。把这三点做好了AI 只是 Kafka 上的一层新能力仅此而已。
