“Kafka已正式接入AI”，这个标题听起来挺像哪家公司的新闻稿，但落到实际工程里，它说的其实就是一件非常具体的事：AI应用和大模型服务之间，需要一条可靠的数据通道，而Kafka正在成为越来越多团队的首选。这不是什么黑科技，而是架构选型的问题。

这两年大模型应用喷涌式出现，我见过不少团队把AI能力接入业务系统时，第一步都是“加一个HTTP接口调一下模型服务”，等流量一上来、推理链路一长，就开始出各种问题：请求超时、服务雪崩、数据对不上、日志没法追踪。最后绕一圈回来，还是老老实实把Kafka请了回来。这篇文章我会从Kafka在AI链路中的定位讲起，聊透为什么AI场景需要它、怎么设计Topic和消费模型、怎么用Docker快速搭一套可用的环境，再到延迟排查、lag定位、重复消费这些高频问题，把一套能直接落地的方案完整捋一遍。

如果你是后端开发、算法工程师或者正在做大模型应用落地的架构师，这篇文章应该能帮你少踩不少坑。

1. Kafka在AI链路里的定位：不只是消息队列

1.1 先搞清楚“接入AI”要解决什么问题

先说个我自己的观察。很多人一提“Kafka接入AI”，第一反应是“Kafka里存了AI生成的内容”，或者“用AI去分析Kafka里的日志”。这两个方向都存在，但更普遍、更务实的场景是：把Kafka当作AI应用的数据中枢和任务调度管道。

你自己想想一个大模型应用的完整请求链路：用户发来一句话，系统要先把上下文拼好，可能要检索知识库，然后调用大模型接口，等模型生成完，再回传结果，最后还要记录日志、做质量评估、沉淀样本。这条链路里，同步调用最大的问题就是大模型推理太慢了。一次性生成几百个token可能要好几秒甚至几十秒，如果所有业务都同步等这个结果，用户请求会被卡死，后端服务也会被连接数拖垮。

所以AI应用落地时，几乎必然走向异步化：请求先进队列，立刻告诉用户“已受理”，后台慢慢推理，完成后通过回调、轮询或者推送把结果交给用户。这个队列就是Kafka最适合的位置。

另外，AI系统往往不是单打独斗的。用户行为日志、业务数据库变更、模型服务的输入输出、特征数据、反馈数据，这些数据源分散在各个系统里，模型训练和推理都要用到。Kafka能把这些数据统一接进来，形成一条标准化的数据管道，让所有下游消费方按自己的节奏去取数。这就不只是消息队列了，而是整个AI链路的事件总线。

1.2 为什么选Kafka而不是直连、Redis或RabbitMQ

有的朋友可能会问，既然只是异步解耦，用Redis Stream甚至直接塞数据库不行吗？短链路、小流量的场景确实可以，但一旦流量上来，差别就出来了。

我做一个对比，方便你根据场景判断：

为什么很多AI项目最后落到Kafka上，核心就是两点：海量堆积能力和回放能力。大模型推理速度跟不上生产速度时，消息就得在队列里堆着，Kafka能扛住千万级甚至亿级的堆积，而且堆得越多性能下降越不明显。回放能力也用得上——模型出问题的时候，你可能会想拿同一批请求重新推理一遍，Kafka只要重置offset就能做到，其他队列基本做不到这一点。

当然，如果你们团队从来没接触过Kafka，只是内部系统之间做个简单异步通知，RabbitMQ上手更快，没必要为了“高大上”硬上Kafka。但如果你要做的是AI数据管道，或者模型请求量可能有明显峰值，那Kafka基本就是绕不开的选择。

2. 核心细节解析与实操要点：从Topic设计到消费模型

2.1 Topic与分区怎么设计才符合AI场景

Kafka的Topic设计直接决定了后续所有链路的复杂度。我在AI项目里通常会把消息分成几类，每一类对应一组Topic。

第一类是推理请求Topic，比如叫ai-inference-request。业务系统把推理请求发到这里，消息内容包括请求ID、业务类型、上下文文本、参数配置等。请求ID必须带上，这是整条链路追踪的凭证，后面排查问题全靠它。

第二类是推理结果Topic，比如叫ai-inference-response。模型服务推理完成后，把结果写回这里，业务系统消费结果再推送或回调给用户。这样生产方和消费方完全解耦，业务方不用关心模型服务是怎么调的、在哪台机器上跑的。

第三类是样本回流Topic，比如叫ai-sample-feedback。用户对模型结果的反馈、人工修正后的答案、质量评分，这些都要记录下来，回流到训练管道里，模型才能持续优化。这一步很多人会忽略，但真正做AI应用的团队都知道，数据飞轮才是长期竞争力。

分区数的设置，我的建议是“按消费者的并发能力和下游处理能力来定”，而不是拍脑袋选个数。一个分区的数据只会被同一个消费组里的一个消费者实例处理，所以分区数决定了某个消费组的最大并行度。比如你的推理服务有6个节点，那分区数至少应该大于等于6，否则一定有节点闲着。但也不要盲目设很多分区，分区越多，ZooKeeper或KRaft元数据压力越大，客户端连接和负载均衡成本越高。一般场景下，按峰值吞吐量除以单个消费者能处理的速度再留30%余量，就够用了。

2.2 生产者参数和消费组机制如何配合AI请求链路

Kafka接入AI系统时，生产者和消费者的参数设置非常关键，直接决定端到端延迟和吞吐。

生产者侧，重点看acks和linger.ms。AI请求通常不允许丢消息，所以acks=all最稳妥，配合enable.idempotence=true还能避免网络重试导致的消息重复。linger.ms的意思是“多攒一会儿再发”，默认0是每条直接发，延迟最低但小消息太多，吞吐上不去。批量推理场景里，把linger.ms调到5到20毫秒，配合合适的batch.size，吞吐能明显上一个台阶。代价是单条消息的延迟多了十几毫秒，对于异步推理链路来说完全可接受。

消费者侧，最核心的是消费组和offset提交方式。AI请求处理有个特点：处理时间长、偶尔失败重试，所以务必使用手动提交offset，而不是默认的自动提交。自动提交是每隔5秒提交一次当前消费位置，如果消息处理到一半进程挂了，重启后会从上次提交的位置继续消费，那部分处理了但还没提交的消息就会被重复消费。手动提交的好处是你可以在消息处理成功后再提交，处理失败的可以单独处理，做到“至少一次”语义下的业务幂等。

消费组还有个容易踩坑的地方：max.poll.interval.ms。Kafka消费者在两次poll之间的时间不能超过这个值，默认是5分钟。如果你的模型推理一次要几分钟，消费者去调模型接口的同时没有再去poll，就可能被判定为“消费超时”，触发Rebalance，分区被分配给其他消费者。这种情况在AI场景里特别容易发生，我记得第一次踩坑的时候，日志里全是“Rebalance”相关的告警，消费者被踢来踢去，消息处理完全乱套。解决办法有两个：要么把max.poll.interval.ms调大，比如调到10到15分钟；要么用Spring Kafka的@KafkaListener配合单独的线程池处理消息，让poll线程始终保持活跃。

2.3 借助Kafka实现批量推理：攒批窗口怎么设计

大模型推理有个典型特点：单条推理要占资源，批量推理能显著提升GPU或CPU利用率。Kafka天然适合做批量数据的中转。

具体怎么做？消费者一次性拉取一批消息，比如max.poll.records=32，积攒到一定条数或等到一定时间窗口，再统一调用一次模型服务做批量推理，结果再批量写回结果Topic。攒批窗口有两个参数要权衡：一是条数阈值，二是时间阈值，谁先到就触发。

我举个实际例子。之前做一个内容审核项目，单条推理的耗时要800毫秒，但32条批量推理只要5秒，平均每条156毫秒，吞吐提升了5倍多。我们把条数阈值设为32，时间阈值设为500毫秒，这样在流量充足时走条数触发，流量稀疏时也不会等太久。这个设计需要你在“延迟”和“吞吐”之间做取舍——攒批时间越长吞吐越高，但单条消息的响应时间也越长。异步场景建议时间阈值控制在1秒以内，尽可能不让用户感知到明显的延迟增加。

3. 实操过程与核心环节实现：从零搭一套Kafka+AI示例

3.1 用Docker快速搭建Kafka环境

现在搭Kafka环境已经比以前简单太多了，直接上Docker Compose。我建议直接用Bitnami的镜像，或者Apache官方的Kafka镜像，配置清晰，也支持KRaft模式（不需要单独的ZooKeeper，部署和维护省不少事）。

下面是一份能直接用的docker-compose.yml，单机版，KRaft模式，端口映射到宿主的9092：

version: '3.8' services: kafka: image: bitnami/kafka:3.6 container_name: kafka ports: - "9092:9092" environment: - KAFKA_CFG_NODE_ID=0 - KAFKA_CFG_PROCESS_ROLES=controller,broker - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093 - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true volumes: - kafka_data:/bitnami/kafka volumes: kafka_data:

启动命令很简单：

docker-compose up -d

启动后验证一下是否正常，可以用官方自带的命令行工具。注意容器内的脚本路径，Bitnami镜像通常在/opt/bitnami/kafka/bin/下：

docker exec -it kafka /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic ai-inference-request --partitions 6 --replication-factor 1 docker exec -it kafka /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

很多人喜欢用kafka-console-producer.sh和kafka-console-consumer.sh来快速验证消息是否通。这里有个高频小问题：消费命令启动一次会一直运行吗？答案是会。命令行消费者启动后会持续监听Topic里的新消息，不会自动退出，你想停掉直接用Ctrl+C就行。如果你是带着--from-beginning参数启动的，它会把该Topic里所有历史消息从头消费一遍，所以测试完记得及时退出，不然会误以为消息“又被重复消费了”。

还有一点值得提，本地测试时advertised.listeners要配置成localhost:9092，这样客户端才能从宿主机连通Kafka。如果在服务器上用Docker部署，需要改成服务器的实际IP或域名，否则客户端连不上，这个问题我在一开始搭环境时踩过。

3.2 本地部署AI模型服务并和Kafka打通

搭好Kafka之后，接下来就是重头戏：怎么让模型服务和Kafka真正跑通。现在主流做法是用Spring Boot写业务服务，通过spring-kafka接入Kafka；模型推理服务可以单独部署，通过Kafka消费请求、回传结果。如果你是做Java后端的，这套链路最顺手；如果模型服务用Python写的，也很容易，用confluent-kafka-python或kafka-python库就能对接。

先看Java侧怎么配置。spring.kafka的YAML配置是搜索引擎里出现频率特别高的关键词，我直接给出一份生产实践里验证过的基础配置：

spring: kafka: bootstrap-servers: localhost:9092 producer: acks: all retries: 3 batch-size: 16384 linger-ms: 5 buffer-memory: 33554432 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer consumer: group-id: ai-inference-group enable-auto-commit: false auto-offset-reset: latest max-poll-records: 32 max-poll-interval-ms: 600000 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: "*" spring.json.value.default.type: com.example.ai.model.InferenceRequest

生产者的JSON序列化依赖Spring的JsonSerializer，消费者的spring.json.value.default.type指定了反序列化的目标类，这里我写的是示例包名，你得换成自己项目里的实际类路径。另外max-poll-interval-ms=600000就是我前面说的，给模型推理留足时间，防止消费者在处理长耗时任务时被误判掉线。

生产端发送消息的代码很简洁，用KafkaTemplate即可：

@Service public class InferenceProducer { private final KafkaTemplate<String, InferenceRequest> kafkaTemplate; public InferenceProducer(KafkaTemplate<String, InferenceRequest> kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; } public void submit(InferenceRequest request) { kafkaTemplate.send("ai-inference-request", request.getRequestId(), request); } }

注意send方法的第二个参数是key，这里我用requestId作为key，这样同一个请求ID的消息会被路由到同一个分区，保证顺序。顺序性在AI链路里很重要，尤其是上下文相关的连续对话场景，消息顺序乱了你后面根本没法排查。

消费端用@KafkaListener监听Topic，配合Acknowledgment手动提交offset：

@Component public class InferenceConsumer { private final ModelService modelService; private final KafkaTemplate<String, InferenceResult> kafkaTemplate; public InferenceConsumer(ModelService modelService, KafkaTemplate<String, InferenceResult> kafkaTemplate) { this.modelService = modelService; this.kafkaTemplate = kafkaTemplate; } @KafkaListener(topics = "ai-inference-request", groupId = "ai-inference-group") public void onMessage(ConsumerRecord<String, InferenceRequest> record, Acknowledgment ack) { InferenceRequest request = record.value(); try { InferenceResult result = modelService.infer(request); kafkaTemplate.send("ai-inference-response", request.getRequestId(), result); ack.acknowledge(); } catch (Exception e) { // 记录失败，发送到异常Topic或进入重试队列 // 不要在这里ack，否则消息会丢失 kafkaTemplate.send("ai-inference-deadletter", request.getRequestId(), request); ack.acknowledge(); } } }

这段代码里有一个关键点：处理失败时不要把消息丢掉。如果你的模型服务暂时不可用，消息被poll出来后如果既没提交offset、也没做人工处理，服务一直重试也不是办法。更合理的做法是把失败的消息送到一个专门的死信Topic（dead letter topic），等模型服务恢复后再人工回放。这样主链路不会卡死，失败数据也不会丢。

如果模型服务是Python实现的，消费逻辑也类似。下面是一个批量推理消费者的伪代码框架：

from kafka import KafkaConsumer, KafkaProducer import json, time consumer = KafkaConsumer( 'ai-inference-request', bootstrap_servers='localhost:9092', group_id='python-inference-group', enable_auto_commit=False, max_poll_records=32, max_poll_interval_ms=600000, value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) buffer = [] window_size = 16 last_trigger_time = time.time() for msg in consumer: buffer.append(msg) if len(buffer) >= window_size or time.time() - last_trigger_time >= 0.5: batch = [item.value for item in buffer] results = batch_infer(batch) # 调用模型的批量推理接口 for item, result in zip(buffer, results): producer.send( 'ai-inference-response', key=item.value['request_id'].encode('utf-8'), value=result ) consumer.commit() buffer.clear() last_trigger_time = time.time()

注意消费者组id换了，因为Java服务和Python服务如果消费同一个Topic，一般不应该共用同一个group id——除非你就是想让它们分摊同一个Topic的消息负载。如果两边各干各的，必须用不同的消费组。

3.3 可视化监控工具怎么选：把Lag和吞吐看明白

Kafka接入AI之后，监控是刚需。没有可视化工具，你根本不知道消息堆积了多少、消费端是不是已经到了瓶颈。这也是为什么“Kafka可视化工具”相关的搜索热度一直很高。

常用的开源工具我整理了一张对比表：

我个人习惯是本地开发和测试用Kafka UI，部署到测试环境或生产环境之后，用Prometheus配合Grafana做指标采集和告警，因为生产上关键指标必须能趋势化回溯，出了事能往前查。Kafka自身暴露了大量JMX指标，比如kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec、BytesInPerSec、BytesOutPerSec，还有消费者端Lag指标，用JMX Exporter把指标拉出来接到Prometheus，再配上Grafana模板，一套基础监控几分钟就能跑起来。

另外一个细节：每次上线前先把Lag监控看一遍，比任何代码review都更能发现问题。有一次我们一个消费服务升级后，CPU突然跑满，Lag指数级上涨，全靠监控曲线定位到是反序列化逻辑写了个死循环，这种问题没有监控根本发现不了。

4. 常见问题与排查技巧实录

4.1 消息延迟高，问题可能出在哪几个环节

“Kafka消息延迟高”是个高频搜索词，但答案往往不是一个点，而是整条链路。根据自己的排查经验，我建议按这四个环节逐一排查。

第一是生产端延迟。如果生产者发了消息但很快被broker确认，说明发送链路没太大问题；但如果消息丢在生产者本地缓冲里迟迟发不出去，大概率是linger.ms设置过大，或者buffer.memory满了在阻塞。把生产端的发送耗时和回调耗时打点看一下，立刻就能定位。

第二是Broker端延迟。Kafka写消息要落盘，磁盘IO慢、Page Cache压力大、分区副本数设置不合理都会导致写延迟。查看broker的kafka_server_NetworkProcessorAvgIdlePercent和磁盘await指标，如果长时间低于阈值，该加机器加固态了。

第三是消费端延迟。这是最常见的瓶颈。消费一条消息要多久？处理逻辑里如果调用了外部模型服务，单条消息处理时间可能就是几百毫秒到几秒。你可以先调整max.poll.records和fetch.min.bytes，用批量拉取来摊薄开销，如果还不行，就要考虑横向扩容消费者实例。

第四是端到端的业务链路延迟。比如从业务发送消息到消费处理完成，中间经过了Kafka，但也可能经过了一堆中间件。建议全链路给每条消息打上时间戳：发送时间、到达时间、开始处理时间、处理完成时间。这四个时间节点的差值，能帮你快速划清延迟到底出在生产、传输、排队还是处理上。

4.2 Kafka Lag一直增长怎么办

Lag就是消费者还没消费完的堆积消息数，Lag只增不减，说明消费速度跟不上生产速度，或者消费者已经彻底卡死了。

排查思路比较简单粗暴：

1. 看消费者组里每个成员的状态：是Active还是Dead，如果消费者掉线了，先解决掉线问题；如果还在运行但Lag不动，那就是处理逻辑卡住了，可能是死锁、无限等待外部接口、OOM等。
2. 看单条消息的处理耗时：加日志或打点，计算每秒能消费多少条。如果你每秒消费100条，但生产速度是每秒500条，Lag一定会涨。解决办法是扩容消费者实例，但注意分区数要大于等于消费者实例数，否则扩了也白扩。
3. 看有没有Rebalance风暴：消费者频繁加入、退出、再平衡，会导致Lag出现锯齿状波动。日志里如果频繁出现Rebalance相关记录，要重点检查max.poll.interval.ms是否设置合理，以及消费者是否有线程卡在长任务里。
4. 看下游能力：如果消费端把消息发给模型服务，模型服务的吞吐瓶颈也会传导到Kafka消费端。这时候Kafka只是“背锅的”，真正的瓶颈在模型服务。扩容模型服务、做批量推理、或者做限流保护，才是解决方向。

4.3 消费者为什么会重复消费？能重复消费吗

这个问题也是面试和实战里的高频条目。Kafka默认的消费语义是至少一次（at least once），也就是说消息不会丢，但在极端情况下可能重复。什么情况下会重复？消费者处理完消息后，还没来得及提交offset，进程就挂了，重启后会从上次提交的位置重新消费，刚才那条消息自然就再被处理了一遍。

所以答案很明确：能重复消费，而且默认情况下确实可能重复消费。解决方案不是让Kafka不重复，而是在业务层做幂等。AI推理场景里，如果重复调用一次模型服务，结果通常是无害的，但如果是扣费、发消息这类有副作用的操作，就必须做幂等控制。常见做法是消费端维护一个已处理RequestId的集合，每次处理前先查一下；或者把处理结果写入带唯一约束的存储，重复消费时靠数据库约束去重。

我还有个小技巧：如果重放了大量历史消息做模型复盘，建议专门用一个debug消费组，指定auto.offset.reset=earliest从头消费，跑完直接删除这个消费组，不影响正常链路的offset位置。

4.4 Kafka和RabbitMQ到底怎么选

这是一个老生常谈的问题，但在AI场景下有一个比较清晰的判断标准。

如果你的核心诉求是“让业务系统之间松耦合、异步通知、简单任务分发”，RabbitMQ更合适，它上手门槛低，路由规则灵活，社区资料也多。但如果你要处理的是“海量数据管道、日志采集、AI样本回流、需要历史回放、消费端需要高吞吐”，Kafka几乎是不二之选。

我自己见过一些团队在选型时不看场景，单纯因为“Kafka更牛”就硬上，结果搞出一堆运维负担。反过来也有团队用RabbitMQ硬扛每天上亿条日志，最后各种调优都救不回来。选型不是比技术先进性，而是匹配业务特征。AI链路里，离线训练和在线推理都依赖数据的完整性和可回放性，这是Kafka的主场。

4.5 命令行消费、Topic数据查看这类基础问题集中回答

热搜词里有一堆关于Kafka命令行的查询，比如“kafka查看topic中的数据”、“kafka生产消费命令启动一次会一直运行吗”。这些确实是新手最容易卡住的地方，顺手写几个我常用的命令。

查看Topic列表：

kafka-topics.sh --bootstrap-server localhost:9092 --list

查看某个Topic的分区和偏移量情况：

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic ai-inference-request

查看某个消费组的Lag：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group ai-inference-group

这条命令会列出每个分区当前消费到的offset和最新offset，Lag就是两者的差值。排查消息堆积时，这条命令比任何监控工具都直观。

从最早的消息开始消费一个Topic：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ai-inference-request --from-beginning

提醒一下，这条命令消费完不会自动退出，会一直挂着等新消息，要退出按Ctrl+C。如果你只想看看有没有新消息，不加--from-beginning，它只会从启动时刻之后的新消息开始打印。

5. 把Kafka接入AI后的扩展经验

5.1 Kafka作为AI Agent的“事件总线”

现在很多团队在搞AI Agent，本质上是多个模型、多个工具、多步任务编排。Agent系统有个很现实的问题：多个组件之间怎么协作？如果每个组件都互相直连调用，编排逻辑会乱成一团，而且每一步的状态、日志、输入输出都要有记录，出了问题才能回溯。

Kafka在这里可以作为Agent系统的事件总线。每个Agent节点把自己的动作事件、中间结果、工具调用记录发到对应的Topic，编排引擎消费这些事件决定下一步做什么；同时所有事件天然被持久化下来，随时可以复盘一次完整的任务链路。这种“事件驱动 + 事件溯源”的架构，比层层嵌套的函数调用清晰很多。

比如一个简单的任务：“根据用户问题查询数据库并生成报告”。用户问题进来后先发到agent-task，检索Agent消费后把SQL查询结果发到agent-result，报告Agent再消费并生成文件。整个过程每一步都有事件记录，任何一步出了问题都能定位到具体Topic和具体消息。

5.2 样本回流和模型迭代链路

Kafka在AI链路上的价值不只是在线推理，还有数据飞轮。用户对模型输出的每一次反馈、每一次人工修正、每一次质量评分，都可以通过Kafka回收到样本池。这些样本经过清洗和筛选后，进入模型训练管道，训练好的新模型再通过模型服务上线。

这条闭环链路里，Kafka是承上启下的那一环。没有这个数据管道，反馈数据散落在各业务系统的数据库里，采集合规和清洗成本会非常离谱。有了统一的Topic和标准消息格式，新模型训练时只需要拉取对应时间段的反馈数据，整个迭代节奏都能加快。

5.3 监控告警：接入AI后必须补上的“安全网”

最后必须提醒一点：把AI接入Kafka之后，监控告警的等级要往上提。AI模型服务有时会被外部供应商限流，有时本地GPU显存溢出，这些问题都会直接体现在Kafka的Lag和消费失败率上。你最好提前配好三个基础告警：Lag超过阈值告警、消费组Rebalance告警、死信Topic有消息告警。这三个告警能帮你第一时间发现问题，而不是等到用户投诉“为什么我的请求一直没返回”的时候才去排查。

基于多年的实操经验，把Kafka接入AI，最花时间的从来不是消息队列本身，而是整条链路上的数据契约设计和异常处理策略。消息格式没定好，后面所有消费者都要跟着改；消费者没做幂等，一次重复消费可能就要赔钱。所以建议先把Topic划分、消息字段规范、消费失败策略这“三件套”定清楚，再动手写代码。这套基础打好了，Kafka+AI这条链路你会用得相当顺手。