企业官网建设流程全解析

1. 这次接入要解决什么问题

1.1 从“数据库轮询”到“事件驱动”

最近把 Kafka 正式接入了 AI 链路，说白了就是把 Kafka 从单纯的“数据管道”升级成 AI 应用的事件中枢。之前的做法比较原始：AI 服务直接查业务库，定时任务每隔几秒拉一次数据，再把数据交给模型去推理。数据量小的时候勉强能跑，等流量一上来，问题就全暴露了。

首先是数据库扛不住。AI 服务要拿数据，业务系统也要拿数据，两边的查询挤在一起，慢查询直接拖垮主库。其次是数据时效性差，轮询周期再短也有空窗，模型拿到的往往是几秒前的“过期数据”。更麻烦的是，一旦某个环节挂了，数据不会自动排队等着，而是直接丢了，你想回放都没地方回放。

换成 Kafka 之后，思路就变了：所有要喂给 AI 的数据，先丢进 Kafka，AI 消费端按自己的节奏去拉。生产端不关心下游谁在消费，AI 服务也不关心上游怎么生产，两边通过一个高吞吐、可重放的消息管道解耦。这正是 Kafka 最擅长的场景——削峰填谷、异步解耦、数据缓冲。

这个项目可以算是一次比较标准的“Kafka + AI”整合实践，适合正在做实时数据分析、智能推荐、异常检测、日志审计这类应用的人参考。无论你是后端开发、数据工程师，还是 AI 应用工程师，只要你的系统里同时存在“大量事件数据”和“模型推理需求”，这套方案都值得看一看。

1.2 整体链路设计

我这次搭建的链路，核心是五个环节：数据生产端、Kafka 集群、AI 消费端、结果存储、监控反馈。

数据生产端负责把业务事件写入 Kafka，比如用户点击日志、订单状态变更、设备上报数据。Kafka 在这里负责暂存和分发，消息按照 key 路由到不同分区，同一 key 的消息天然有序。AI 消费端从指定 topic 拉取数据，组装 prompt，调用大模型做推理，再把结果写回结果存储，同时把处理状态反馈到监控系统。

这条链路里，Kafka 起到了两个关键作用。第一，它是“缓冲池”，生产端瞬间涌入大量数据时，不会直接打到 AI 服务上；AI 消费端可以根据模型推理速度自行控制消费速率，这就是削峰。第二，它是“时光机”，消费端挂了、模型升级了、prompt 调整了，都可以重置 offset 重新消费一遍旧数据，这在调试 AI 应用时太重要了。

后端输出我补一句：如果你的业务数据比较敏感，又不想把数据传到外部模型服务做推理，那 Kafka 这套架构依然成立，只需要把 AI 消费端的模型调用换成本地部署的大模型即可。这个我们后面实操环节会提到。

1.3 为什么是 Kafka 而不是其他中间件

很多朋友第一反应是：我不就是想把数据喂给 AI 吗，直接用 HTTP 回调不行吗？为什么非要上 Kafka？

直接 HTTP 调用在数据量小的时候没毛病，但一旦生产速率超过 AI 服务的处理速率，要么消息积压在内存里，要么直接丢弃，要么把下游打挂。Kafka 的优势在于它是“持久化的消息系统”，数据落盘，可以慢慢消费，还能回溯。这对 AI 场景尤其友好，因为模型推理往往是整个链路里最慢的一环，你总不能让上游等模型算完吧。

还有一个容易被忽略的点：Kafka 支持多消费组。同一个 topic 的数据，A 消费组拿去做了实时风控，B 消费组拿去做了离线特征统计，C 消费组拿去做模型训练样本收集，彼此之间互不影响。这种“一份数据、多处复用”的能力，用 HTTP 回调来实现会非常别扭。

2. 技术选型与核心设计决策

2.1 Kafka 与 RabbitMQ 的选型对比

我知道很多人纠结过 Kafka 和 RabbitMQ 到底选哪个。这次项目选 Kafka，不是因为它“更高级”，而是因为我们的场景就是典型的 Kafka 主场。简单说几个决定性理由：

从吞吐量上看，Kafka 是顺序写盘加批量拉取，单节点能扛住每秒几十万条消息的写入；RabbitMQ 是 AMQP 协议，强在路由灵活，但吞吐量明显低一个量级。我们的业务事件峰值能到几万条每秒，这个量级下 RabbitMQ 会先扛不住。

从消费模型上看，RabbitMQ 是队列模型，一条消息被一个消费者拿走之后就从队列里删掉了，基本上不支持回放。Kafka 是日志模型，消费组通过 offset 记录消费位置，消息本身在保留期内一直存在。AI 场景里，模型版本升级后要拿历史数据重新推理，这种需求 Kafka 天生支持，RabbitMQ 实现起来非常费劲。

从顺序性上看，Kafka 的分区机制能保证分区内有序，合理设置 key 就能做到同一用户的事件有序到达 AI 服务。RabbitMQ 虽然也能实现，但要靠单一队列加单一消费者，并发一上来就废了。

当然，如果你们的场景是任务分发、异步 RPC、需要复杂路由规则，那 RabbitMQ 依然是更好的选择。选型没有绝对的对错，只有适不适合当下的业务。

2.2 消息体结构：JSON 起步，跨团队用 Avro

Kafka 消息本身是字节数组，不关心你往里塞什么格式。但格式选错了，后面维护成本会直线上升。我这次用的是 JSON，因为项目初期团队小、迭代快，JSON 的灵活性最高，改字段不需要重新编译,消费端兼容也比较容易。

等消息量真正上来了，字段变更频繁了，你就会开始后悔没做 Schema 管理。字段删了，老的消费者反序列化直接报错；字段加了名字，新消费者看到的数据缺字段,排查起来很头疼。这时候就该考虑 Confluent Schema Registry + Avro，或者 Protobuf 方案。

Avro 的好处是自带 Schema，消息里只存 schema id，消费端自动拉取对应的 Schema 做反序列化，天然解决字段兼容问题。坏处是生态相对重，需要额外部署 Schema Registry，调试也没 JSON 那么直观。

我的经验是：如果你的 topic 只有自己一个团队消费，JSON 完全够用；如果这个 topic 要跨部门共享，或者要长期存在并持续演进，建议第一天就上 Avro 或 Protobuf，不要等出事故了再迁移，消息格式迁移是出了名的难受。

2.3 消费语义设计：手动提交 + 幂等 + 死信

AI 消费端最忌讳的事情，就是模型已经花了几秒做推理，结果因为消费端崩溃导致 offset 没提交，重启后所有消息重新消费一遍，又把模型调用了一遍。大模型接口按次计费，重复消费一次就是白花一笔钱。

所以消费端我全部采用手动提交 offset，绝对不用 enable.auto.commit=true。流程是：poll 一批消息 → 逐条做 AI 推理 → 推理结果落库成功 → 再手动提交这批 offset。如果推理过程中抛异常，那就不要提交，让消息留在 Kafka 里等待下一次消费。

但手动提交不能完全避免重复消费，因为存在“处理完但还没来得及提交，进程就挂了”的时间窗口。所以业务上必须做幂等。我的做法是在消息体里带上全局唯一的 event_id，AI 推理结果落库时以 event_id 作为唯一键，重复插入时直接覆盖或者跳过。这样就算消息被重复消费，结果也不会产生脏数据。

另外，我专门建了一个死信 topic：ai-events-dlq。普通的消息消费失败重试三次还不行，就不再死磕了，把原始消息和错误原因一起写进死信 topic，由单独的补偿任务去处理。这样主链路不会被单条坏消息卡住，也不会因为无限重试把 AI 服务拖垮。

3. 完整实操过程

3.1 环境准备：Kafka 集群与主题参数规划

先说环境。我这里用的是 Kafka 3.x 版本，单机开发环境直接解压部署，生产环境建议至少三节点集群。Windows 上启动 Kafka 需要先启动 Zookeeper，再启动 Kafka server，配置文件都在 config 目录下。

主题我创建了一个叫 ai-events 的 topic，创建时最核心的参数是分区数和副本数。下面是创建命令：

kafka-topics.sh --bootstrap-server localhost:9092 \ --create \ --topic ai-events \ --partitions 6 \ --replication-factor 1 \ --config retention.ms=86400000

分区数怎么定？我的经验公式是：分区数 ≈ 目标吞吐量 ÷ 单个消费者实际处理能力。比如你的 AI 服务单实例每秒能处理 200 条消息，目标吞吐是 1200 条每秒，那就需要至少 6 个分区。注意，分区数是提前规划的，线上扩容分区虽然 Kafka 3.x 之后支持，但只能增不能减，而且重新分区会带来数据迁移，尽量一次性规划好。

retention.ms 我设置为 86400000，也就是一天。为什么不是默认的七天？因为 AI 任务对消息时效性要求高，超过一天的消息基本没有重放价值，留太久只会白白占磁盘。如果你的场景里有离线训练需求，可以单独开一个 topic 设长保留时间，不要用一个 topic 通吃所有需求。

关于顺序性，还有个重要的点：给消息设置 key 时必须谨慎。假设你按 user_id 做 key，那同一个用户的所有事件都会进入同一个分区，AI 消费端就能按时间顺序看到这个用户的完整行为链。如果 key 设置不合理，消息被哈希到不同分区，顺序就乱了，AI 拿到的上下文可能拼接错乱。

3.2 Python 生产者模拟业务数据

为了验证链路，我写了一个简单的 Python 生产者，模拟用户点击行为日志。用的库是 confluent-kafka，API 比 kafka-python 更底层一点，但性能更好、可控性更强。

import json import time from confluent_kafka import Producer conf = { "bootstrap.servers": "localhost:9092", "client.id": "log-simulator", "linger.ms": 5, "batch.num.messages": 500, } producer = Producer(conf) def delivery_report(err, msg): if err is not None: print(f"投递失败: {err}") else: print(f"已写入 topic={msg.topic()} partition={msg.partition()} offset={msg.offset()}") for i in range(10000): event = { "event_id": f"evt-{i:06d}", "ts": int(time.time() * 1000), "user_id": f"user_{i % 200}", "action": "click" if i % 2 == 0 else "view", "payload": { "page": f"/product/{i % 50}", "duration_ms": (i * 37) % 5000, }, } # key 用 user_id，保证同一用户的事件进入同一分区 producer.produce( "ai-events", key=event["user_id"], value=json.dumps(event, ensure_ascii=False), callback=delivery_report, ) producer.poll(0) time.sleep(0.02) producer.flush()

produce 之后为什么要调 poll(0)？因为 confluent-kafka 的回调函数是异步触发的，poll 的作用是让生产者处理待发送消息和触发回调。时间充裕的场景里，flush 会把积压的消息全部发出去，但日志类场景建议在循环里保持 polling，避免回调堆积。

linger.ms 表示生产者最多延迟多少毫秒发送一批消息，调大这个值可以提升吞吐，但会增加少量延迟。我做日志推送时一般设 5-10ms，既能攒一批消息批量发送，又不至于让消费者感受到明显延迟。

3.3 AI 消费端：接入大模型并结构化输出

消费端是整个项目最核心的部分。我的设计是：从 Kafka 拉取业务事件，把事件组装成 prompt，调用大模型做意图判断和行为摘要，最后把结构化结果写回存储。

import json from confluent_kafka import Consumer, KafkaError conf = { "bootstrap.servers": "localhost:9092", "group.id": "ai-processor", "enable.auto.commit": False, "auto.offset.reset": "earliest", "max.poll.interval.ms": 600000, "fetch.max.bytes": 5242880, } consumer = Consumer(conf) consumer.subscribe(["ai-events"]) def llm_complete(prompt: str) -> str: # 这里接你自己的模型服务： # 1. 外部大模型 API # 2. 本地部署的模型，例如通过 Ollama 起的推理服务 # 3. 企业内部的模型网关 # 返回内容要求是 JSON 字符串 raise NotImplementedError("替换成你自己的模型调用实现") def analyze_event(event: dict) -> dict: prompt = f""" 你是一个用户行为分析助手。请分析下面这条用户行为记录，输出 JSON 格式结果，包含： - intent: 用户的意图，只能是 purchase / browse / leave 之一 - summary: 一句话摘要（不超过20个字） - risk_score: 0-100 的整数，表示异常程度 用户行为记录: {json.dumps(event, ensure_ascii=False)} """ raw = llm_complete(prompt) # 部分模型会在 JSON 外面包 markdown 代码块标记，需要清理 raw = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip() return json.loads(raw) def save_result(event_id: str, result: dict) -> None: # 落库时以 event_id 作为唯一键，做幂等写入 print(f"event {event_id} -> {result}") while True: msg = consumer.poll(1.0) if msg is None: continue if msg.error(): if msg.error().code() == KafkaError._PARTITION_EOF: continue print(f"消费错误: {msg.error()}") break event = json.loads(msg.value()) try: result = analyze_event(event) save_result(event["event_id"], result) # 处理成功后手动提交 consumer.commit(asynchronous=False) except Exception as exc: print(f"处理失败: {exc}, 将重试或进入死信处理") # 重试三次仍失败，写入死信 topic

这里有几个细节值得展开。

max.poll.interval.ms 默认是 5 分钟，如果单条消息处理时间太长，超过了这个值，消费者会被判定为“卡死”，触发 rebalance，分区被分给其他消费者。AI 推理是出了名的慢，特别是接了外部模型之后，单次调用可能就要 2-3 秒，一批消息处理下来很容易超过 5 分钟。所以我把它调到 10 分钟，同时控制每次 poll 的最大消息数，避免一次性拉太多导致处理超时。

enable.auto.commit 必须设为 false，这个前面已经强调了。自动提交的时机是 poll 调用时，它提交的是上一次拉取的位置，而不是已经处理完的位置。一旦处理过程中宕机，就会有大量消息被重复消费，白交模型调用费。

还有一个坑是模型输出的格式不稳定。我实测下来，很多模型在输出 JSON 时喜欢加 markdown 代码块标记，直接 json.loads 会报错。所以我加了字符串清理逻辑，把前缀后缀的json去掉再解析。更稳妥的做法是在 prompt 里强制要求“只输出 JSON，不要包含任何其他内容”，但清理逻辑依然要保留，防的就是模型不听话。

3.4 Java 侧快速实现：Spring Kafka 结合 Spring AI

如果你所在团队以 Java 为主，那更常见的做法是用 Spring Kafka 做消费端，再配合 Spring AI 调用模型。这里给一个可以直接抄的简化实现。

首先是 application.yml 里的核心配置：

spring: kafka: bootstrap-servers: localhost:9092 consumer: group-id: ai-processor-java enable-auto-commit: false auto-offset-reset: earliest max-poll-interval-ms: 600000 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer listener: ack-mode: manual concurrency: 4

listener.concurrency 控制并发线程数，也就是同时有几个消费者实例在处理消息。这个值要和你 topic 的分区数匹配，超过分区数的并发是没有意义的，多出来的线程会空闲。我的 ai-events 是 6 个分区，所以 concurrency 可以设 4 到 6。

然后是消费者逻辑：

@Component public class AiEventConsumer { @KafkaListener(topics = "ai-events", groupId = "ai-processor-java") public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) { try { String result = analyzeWithAi(record.value()); saveResult(record.key(), result); ack.acknowledge(); } catch (Exception e) { // 重试或写死信 throw new RuntimeException(e); } } }

ack-mode 设为 manual 后，只有手动调用 ack.acknowledge() 才算处理完成。这跟 Python 端的 consumer.commit() 是一个道理。

Spring AI 的接入也很顺，它在 starter 里封装了各种模型厂商的客户端。如果你用的是 OpenAI 兼容接口，直接注入 ChatClient 就能调用。Java 这套方案的好处是跟现有 Spring Boot 生态无缝集成，适合已经重度使用 Spring 的团队。

我自己平时 Python 和 Java 两边切换，感觉是：如果团队主要是算法出身，Python 端更灵活，写 prompt 和调试都更快；如果是业务系统团队主导，Java 的维护成本更低。

3.5 用 AKHQ 做可视化与任务查看

Kafka 的命令行工具很强，但日常调试和排查问题，还是可视化工具更舒服。我这次用 AKHQ 比较多，它最大的价值是能把 topic、消费组、Connect 任务集中在一个界面里看。

AKHQ 的部署比较简单，我直接用 Docker 起了一个：

docker run -d \ --name akhq \ -p 8080:8080 \ -e KAFKA_BROKERS="tcp://localhost:9092" \ tchiotludo/akhq

打开页面后，左侧能看到 Topics、Consumers 等菜单。Topics 页面可以查看消息内容和分区分布，Consumers 页面可以直观看到每个消费组的 lag 情况。

如果你使用了 Kafka Connect，想知道某个 connector 任务的运行状态，可以在 AKHQ 左侧找到 Connect 菜单，里面会列出所有 connector，点进去能看到 task 级别的事件日志和状态。有 task 挂掉的时候，界面上会直接标红并显示错误信息，比自己去查 Connect 日志快得多。

除了 AKHQ，Kafka UI 也是不错的开源选择，UI 更现代化，功能大同小异。建议两个都试一下，选自己顺手的。工欲善其事，必先利其器，排查故障时一个趁手的可视化工具能节约一半时间。

4. 遇到的坑与排查实录

4.1 消息延迟高的排查思路

接入 AI 之后最常遇到的现象就是“消息生产端明明秒发，AI 消费端却迟迟没反应”。这种延迟问题，我建议按四个时间段来拆解：

首先是生产端到 Broker 的写入耗时。可以在生产者回调里记录 send 时间，如果发现生产延迟本身很大，优先查网络和 Kafka 服务端负载。

其次是消息在 Kafka 中等待被消费的时间。正常情况下消息写入 broker 后，消费者几毫秒内就能拉到。如果你发现消息已经写入但很久没被消费，那就要看消费端是不是卡住了。

然后是消费处理耗时，这也是 AI 场景下最容易出现瓶颈的地方。大模型推理是串行且耗时的，跟传统的消息处理完全不是一个量级。建议在消费端对“poll 到处理完成”做埋点，统计平均耗时和 P99 耗时，看是不是模型调用拖了后腿。

最后是结果写回的耗时。如果 AI 推理完结果，但落库的数据库响应慢，整体延迟一样会飙升。

我把排查经验整理成一张表，方便你对照处理：

这个“四段式时间分解法”是我排查线上问题最常用的手段，比到处乱看日志高效得多。

4.2 Lag 不断上涨怎么办

Lag 就是消费者落后生产者的消息条数。如果 lag 持续上涨，说明消费速度跟不上生产速度，消息在 Kafka 里越积越多。

导致 lag 上涨的原因主要有三种。一是消费者并发不够，分区数很多但 concurrency 设得小，大量分区处于闲置状态；二是单条消息处理太慢，模型推理耗时高，整体吞吐上不去；三是消费者逻辑里有阻塞操作，比如同步调用外部 API 超时、数据库连接池打满，导致整个消费线程被卡死。

处理思路也要分情况。如果分区数还有余量，直接调大 concurrency 让更多消费者实例并行处理。如果分区数已经被占满了，那只能从提高单消费者吞吐入手，比如把批量消息合并成一次模型推理（前提是业务允许），或者给模型调用加缓存，相同类型的事件直接命中缓存不再推理。

关于查看 lag，最直接的方式还是命令行：

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

输出里 LAG 列就是积压量。如果发现某个分区的 lag 特别大，而其他分区正常，多半是消息 key 分布不均，也就是所谓的热点分区问题，这时候要检查生产端 key 的设计是否合理。

正常情况下，消费端的理想状态是 lag 接近 0。如果 lag 长期维持在一个稳定数值，说明生产速率和消费速率刚好平衡，系统还能维持；一旦 lag 持续增长，那就要立刻扩容或优化，否则堆积到影响消息保留时间后，开始有消息被清理，数据就丢了。

4.3 重复消费和多实例并发问题

“Kafka 能重复消费吗”这个问题经常被问到。能，而且必须假设它会重复消费。引起重复消费的典型原因有三个。

第一是消费端处理完消息但还没来得及提交 offset，进程就崩溃了，重启后从上次提交的位置继续消费，这批消息被重新处理。第二是网络分区或者心跳超时，Broker 认为消费者挂了，触发 rebalance，分区被分配给其他消费者，新消费者从旧的 offset 开始消费。第三是手动提交 offset 时提交失败，比如 commit 超时，下次消费还是会拉到同一批数据。

AI 场景下重复消费的代价比普通业务更高，因为每一次消费都是一次模型调用。我在前面的幂等设计已经聊过，这里再强调一下：消息里带上全局唯一 ID，推理结果落库时用这个 ID 做去重，是性价比最高的方案。

多实例部署时，还要注意一个问题：如果多个消费者实例都属于同一个消费组，同一个分区只会被其中一个实例消费，这是 Kafka 的分区分配机制保证的。不用担心同一批消息被多个实例同时抢到。但如果你开了多个消费组，那每个消费组都会独立拿到全量数据，这是特性，不是 bug，要确认自己到底想用哪种模式。

4.4 AI 返回格式不稳定与失败重试

接入大模型之后，你很快会发现一个事实：模型的输出是不稳定的。同样的 prompt，这次输出合法 JSON，下次可能输出一大段废话，再下次干脆超时。对于 Kafka 消费链路来说，这种不稳定性必须提前设计好防御机制。

首先是格式解析。模型输出内容中掺杂 markdown 标记、解释性文字、额外字段，都是常态。消费端在 json.loads 之前先做一次字符串清理，同时用异常捕获兜底。解析失败的消息不要直接丢弃，单独放到死信 topic，方便事后分析是模型问题还是 prompt 问题。

其次是超时和重试。外部模型接口必须设置超时时间，我一般设 30 秒。超过 30 秒直接判定失败，等一下轮重试，不要无限等待。重试次数控制在 3 次以内，超过就进死信队列。

最后是重试带来的幂等问题。每次重试都是一次额外的模型调用，而且重试的结果可能跟第一次不一样，因为模型本身是概率性的。如果你的业务对结果一致性有要求，第一次成功就不需要重试；如果失败后必须重试，建议把第一次的原始响应缓存起来作为兜底。

4.5 故障排查速查表

最后整理一份速查表，把我这次遇到的典型问题都放进去：

5. 后续演进：Kafka 与 AI Agent 的事件闭环

5.1 从“被动分析”到“主动决策”

这次接入跑通之后，我开始琢磨一个更深层的问题：如果 AI 不再只是消费 Kafka 里的数据做分析，而是消费事件之后直接做出决策，再把决策结果以事件的形式写回 Kafka，让下游系统去执行，那整个架构就变成了一个真正的“智能事件闭环”。

举个例子：AI 消费端从一个叫 order-events 的 topic 里读订单数据，发现某个用户短时间多次下单但都没支付，判断可能是异常行为，于是生成一条 risk-control-action 事件写到另一个 topic，风控系统消费这个事件，自动触发限流或人工审核。

这就是 AI Agent 和 Kafka 结合的方向。Kafka 在这里扮演的是 Agent 的“记忆”和“行动总线”。所有事实、决策、行动结果都变成事件流，AI Agent 不再是孤立的接口调用，而是整个业务系统里一个持续运转的事件处理器。

从工程实现上看，这个演进没有引入新的中间件，还是那套 Kafka 消费生产模型，只是把“AI 分析结果”从单纯落库变成了“落库 + 产生新事件”。架构的扩展成本很低，但业务价值提升了一个量级。

5.2 事件数据反哺模型训练

Kafka 日志保留策略默认会清理旧数据，但对 AI 团队来说，这些历史事件恰恰是模型迭代的宝贵素材。用户在真实业务中产生的事件序列，经过 AI 分析后的意图标签、风险评分、决策结果，把这些数据整理成训练集，比人工标注高效得多。

我的做法是，单独建一个 training-events topic，把 AI 处理完的事件和结果原样写一份进去，保留时间设长，比如 30 天。离线训练任务定期从里面拉数据，经过清洗和标注，生成新的训练样本。这样等于搭了一条“业务事件 → AI 推理 → 样本回流 → 模型再训练 → 新模型上线”的闭环数据管道，而且全程不用人肉搬运数据。

5.3 我的几点体会

这次把 Kafka 接入 AI，踩了不少坑，也有几个比较深的体会。

第一，技术选型前先想清楚自己的场景是重吞吐还是重路由。我们这种海量事件喂给 AI 的场景，Kafka 就是最优解，但如果你只是想把任务分发给几个 worker，那它反而不如 RabbitMQ 顺手。

第二，AI 消费端跟传统消费端完全是两回事。传统消费端处理一条消息只要几毫秒，AI 消费端动辄几秒，消费超时、重复消费、重试策略这些设计要提前考虑，否则上线就跑飞。

第三，幂等设计不是可选项，是必选项。不要抱侥幸心理觉得“我们的消费端应该不会重复”，只要手动提交 offset 存在窗口期，重复消费就一定会发生。

最后，如果要给一个人做入门建议，我会说：不要一上来就搞流处理框架，先把生产者、消费者、主题、消费组、offset、lag 这些最基础的概念用熟，把 Kafka 和 AI 的链路亲手摸一遍，再往上加复杂的东西。基础打牢了，后面怎么扩展都不慌。