企业官网建设流程全解析

把一句话撂在这：AI能不能在业务场景里真正跑起来，很多时候不取决于模型有多聪明，而是数据能不能及时、可靠地送到它面前。最近我们团队终于把 Kafka 正式接入 AI 这条链路跑通了，从最开始“用脚本手动喂数据给模型”的野路子，换成了整套基于 Kafka 的实时数据管道。整个过程踩了不少坑，也把生产消费机制、集群部署、可视化监控、消费延迟排查这些老问题重新过了一遍。这篇文章就围绕这次改造，聊聊我理解中的 Kafka 到底在 AI 落地中扮演什么角色、接入架构怎么设计、部署与排查有哪些实操细节，希望对正在做同类事情的你有点参考。

这套方案适合谁看？如果你正在把大模型能力往业务系统里集成，或者想把 AI Agent 的状态流转、工具调用、日志回放做成可审计的事件流，再或者你只是被“消费组一直拉不到数据”“lag 越来越高”折磨过，那这篇文章应该都对得上。下面的内容不是我抄文档写出来的，是这次接入过程中一段一段踩出来的经验，尽量说人话，能直接抄作业的就直接抄作业。

1. AI 落地为什么绕不开 Kafka 这个角色

1.1 AI 不只是在跑模型，更是在跑数据管道

很多团队第一次做 AI 项目时，注意力全在“选哪个模型”“怎么调 prompt”“要不要微调”上，结果真正上线的时候，第一个卡住的地方不是模型效果，而是数据怎么送到模型面前、模型结果怎么送回业务系统。

举个例子：我们对接过一个小场景，用户前端点了一下“分析这份文档”，后端把文档丢给大模型接口，模型返回结果，接口等待几秒钟，用户看到结论。这个场景用普通的 HTTP 同步调用完全够用。但一旦你开始做流式对话、实时内容审核、用户行为意图识别、多 Agent 协作、历史会话回放，事情就变了。每一轮对话、每一次工具调用、每一条推理日志都会源源不断产生，而且上下游系统不止一个，你不可能让“文档分析服务”直接去调“用户画像服务”“消息通知服务”“模型调用服务”各自维护的接口，那样耦合会变成一锅粥。

这时候 Kafka 的位置就出来了。它可以放在业务系统和 AI 系统之间，作为一条统一的数据总线。上游只要把数据写进 topic，下游的 AI 推理服务、存储服务、监控服务各取所需。数据结构和消费速率可以各自演进，互不阻塞。

1.2 Kafka 与 AI 场景天然契合的三个点

我总结下来，Kafka 和 AI 场景能合得来，靠的是三个底层特性：高吞吐、可重放、解耦。

先说高吞吐。AI 场景里数据量往往集中在“日志 + 特征 + 推理请求”这三类，单看一条好像不大，但叠加起来量级很吓人。Kafka 的顺序写盘和批量机制，能以非常稳定的吞吐扛住这种流量，不会因为某个消费者处理慢了就把生产者堵死。群里经常有人问“kafka 和 rabbitmq 有什么区别”，在 AI 管道这个场景下，Kafka 的吞吐和分区并行度就是比 RabbitMQ 更适合做数据总线，它设计上就是为“海量事件流”准备的。

再说可重放。模型结果出问题的时候，你最需要的就是“把某段时间的输入事件重新拉出来再跑一遍”。Kafka 的消息默认会按保留策略存在磁盘上，消费者可以指定从旧 offset 开始重新消费。这个能力在 AI 调试期价值极大。我们在接入过程中，因为 prompt 模板改了几次，就是靠重新消费 topic 里的历史请求，把同一个用户输入在新旧模板下的输出对比了一遍，整个回放过程不用让用户再操作一次。

最后是解耦。AI 团队和业务团队可以各自用各自的语言写服务，业务方往 topic 里丢事件，AI 方从 topic 里拿事件，谁也不需要知道对方服务地址和调用协议。这对多团队协作的项目非常友好。

1.3 什么时候不要硬上 Kafka

但我也得说句公道话：不是所有 AI 项目都得上 Kafka。如果你只是一个小工具，每天几百次请求，模型调用一次几秒甚至几十秒，那加一个 Kafka 除了增加运维负担，没有任何收益。消息队列解决的核心问题是“流量削峰 + 异步解耦 + 多消费者”，你没有这个瓶颈，就别凑热闹。

我们内部定过一个判断标准：如果系统只有单一消费者、数据量很小、没有回放需求，用 HTTP + 数据库表就能解决，别上 Kafka。如果系统已经出现“多个下游都要同一份数据”“高峰期请求把上游接口打爆”“需要回溯历史消息排查问题”这三个特征里的任意两个，那 Kafka 就是合适合适的选择。这个判断标准也陪你放到别的项目里试用一下。

2. Kafka 接入 AI 的三种典型架构

2.1 数据管道模式：训练、特征、数据回流

第一种模式也是最常见的：把 Kafka 当作数据管道，连接业务数据源和 AI 数据处理层。

大致链路是：业务系统的日志、数据库变更、埋点数据，通过 Canal、Debezium 之类工具同步进 Kafka；下游用 Flink 做清洗和特征计算，再落到特征存储；特征既喂给模型做离线训练，也喂给在线推理。模型打分或生成的结果，再写回另一个 topic，供业务方消费。

这种模式我在项目里用的 topic 命名习惯是：source_db_log（原始数据）、feature_processed（处理后的特征）、model_result（推理结果）。有几个细节值得注意：原始数据 topic 通常设置较长的保留时间，因为你需要回放；特征 topic 可以设置紧凑型保留策略，因为特征只要最新版本；结果 topic 一般只保留几天，因为结果落库后原始内容就不需要长期躺在 Kafka 里。

如果你只是偶尔跑一次离线数据分析，可以不接流处理框架，直接用kafka-console-consumer配合脚本消费到本地文件，再喂给脚本处理。但一旦上了正式环境，Flink 或 Spark Structured Streaming 是更稳的选择。我们这次正式接入 AI 后，就把离线特征补数任务迁移到了 Kafka + Flink 上，数据延迟从原来的“T+1”降到了分钟级。

2.2 请求流模式：在线推理的异步化

第二种模式是把 Kafka 作为请求入口层，解决在线推理的异步化问题。

之前我们做一个 AI 内容分析接口，上游调用方几百个，高峰期每秒几千个请求，而模型服务的吞吐只有几百 QPS。如果直接同步调模型，模型服务必挂，调用方也等不起。我们当时的做法是：请求进来只做参数校验，然后把请求体发到inference_request这个 topic，立刻返回“已受理”；消费端从 topic 拉请求，按批次拼接后调模型服务，拿到结果再写回inference_result，由结果通知服务主动回调给调用方。

这套设计里，Kafka 扮演的是“缓冲池”角色。它能扛住瞬时流量洪峰，模型服务就算处理不过来，请求也不会丢，只是延迟变高。配合分区数量的设置，你可以控制模型消费的并行度，一个分区对应一个消费线程，消费线程数量乘以单线程吞吐，就是整个系统的最大处理能力。想扩容时，不是去改代码，而是调分区数和消费者数量。

2.3 AI Agent 事件流模式：让 Agent 的行为可以被审计和回放

这一种模式，是这次“Kafka 接入 AI”中我个人觉得最值回票价的。AI Agent 应用和传统 CRUD 应用有一个本质区别：Agent 的工作过程不是“一次请求一次响应”那么简单，它会多次观察环境、调用工具、生成中间思考，最后才给用户一个结果。这些过程如果可以完整记录，好处非常多：调试时能看到 Agent 为什么走弯路，线上运行时有完整轨迹可以回溯安全问题，用户问“你为什么这么回答”时，可以拿出证据链。

我们的做法是：Agent 每一步动作都发一个事件到agent_event这个 topic，事件类型包括agent_started、tool_called、llm_replied、human_feedback、agent_finished，消息体里带上 trace_id 串联整条链路。下游有两个消费者：一个负责把事件落到 ES 供检索，一个负责实时分析 Agent 的异常行为。

刚开始做这套事件流时，最大的问题是“事件太多导致 Kafka 堆积”。每个 Agent 任务至少产生几十条事件，如果用户量上到万级，事件量非常大。我们后来加了两个策略：内部步骤事件直接批量写，用户可见事件即时写；低频场景下事件保留一周，高频场景保留三天。这个“按事件价值分级处理”的思路，比单纯靠 Kafka 扛数据更实际。

3. 环境搭建：集群安装与可视化工具选型

3.1 Docker 快速起一个单机开发环境

如果你只是本地联调，完全没必要先搭一套完整集群。我用 Docker 起 Kafka 的次数太多了，这里分享一个最省心的组合：Kafka + KRaft 模式（不依赖 ZooKeeper），镜像直接用 Apache Kafka 官方镜像。

version: "3" services: kafka: image: apache/kafka:3.7.0 container_name: kafka-dev ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 volumes: - kafka-dev-data:/var/lib/kafka/data volumes: kafka-dev-data:

启动就一句话：docker compose up -d。然后你可以用官方容器里自带的命令验证一下：

docker exec -it kafka-dev /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic ai-events --partitions 3 --replication-factor 1

这里有个关键配置是KAFKA_ADVERTISED_LISTENERS，很多本地联调连不上的问题，都是因为这里写成了kafka:9092。容器外部访问必须写宿主机可访问的地址，我之前在这上面卡了半小时，后来改成localhost立刻通了。

3.2 生产集群安装要点

开发环境可以单机跑，正式环境还是老老实实搞三个 broker 起步。我们这套 Kafka 集群就是标准的三节点部署，机器配置是 8C16G，磁盘用的是 SSD，分区副本数设 3，topic 默认分区数根据下游消费并行度设。集群安装本身不复杂，重点反而是几个容易被忽略的参数。

log.retention.hours控制消息保留时长，AI 场景下建议按“你是否可能需要回放”来定，需要回放的原始数据 topic 我一般保留 48~72 小时，临时 topic 默认 24 小时。num.partitions默认值建议直接改成 6 或 12，不然后面创建 topic 忘了带分区数，热 topic 的并行度会不够。default.replication.factor建议显式设成 3，防止某些 producer 不指定副本数时，直接拿默认 1 去写，那数据安全就是裸奔。

还要提醒一个事：KRaft 模式确实省掉了 ZooKeeper，但部署时对node.id、controller.quorum.voters的要求更严格，多个节点之间的配置如果有一处不一致，启动会直接拒绝。我第一次搭 KRaft 三节点集群时，就是有两个节点的advertised.listeners写错了，导致集群状态一直不健康。所以生产环境上，不管你是 KRaft 还是 ZooKeeper 模式，第一步一定是先把kafka-configs.sh --describe这套检查脚本跑通，再看业务指标。

3.3 可视化工具到底选哪个

运维 Kafka 不能只靠命令行。排查问题的时候，可视化工具的“一眼看到 topic、分区、lag”能力，能帮你省下大量 SSH 到机器上敲命令的时间。我试过好几款，把体验整理成表格供你参考。

我的个人建议是：本地开发用 Kafka UI 就够了，轻量、直观，缺点是很久没更新，功能相对基础。团队运维阶段可以上 Kafka Map，它对新版 Kafka 的兼容更好，查询消息也方便。如果你习惯桌面客户端操作，Offset Explorer 依然是很多老运维的首选，界面顺手，功能全。

不过要记得，可视化工具只是辅助，真正定位问题还是得回到命令行和监控指标上。我见过不少同事盯着 UI 上的 lag 数字发呆，却不知道这个数字到底是怎么算出来的，下一步我们聊的就是这个。

4. 生产消费命令与延迟排查实战

4.1 那些高频使用的生产消费命令

很多人初学 Kafka 的时候，都会被命令行工具搞晕，尤其是“kafka-console-consumer启动一次会一直运行吗”这种问题。答案是：会，而且这是设计如此。消费者启动后会一直监听分区里的新消息，不会因为读完了就退出，这和普通命令行工具用完就退的习惯完全不同。

生产消息的基本命令是：

# 进入 kafka 安装目录/bin 下执行 kafka-console-producer.sh --broker-list localhost:9092 --topic ai-events

回车后输入一行，回车发送一条。输入Ctrl+C退出。这个命令通常只用来测试通路，生产环境不会用它。

消费消息的命令就要多记几个参数了：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ai-events --from-beginning kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ai-events --group test-group

--from-beginning表示从最早的 offset 开始消费，不带它时，新消费组默认从最新偏移量开始，也就是只能看到启动后新产生的消息。这也是好多人“明明 topic 里有数据，怎么消费不到”的原因。--group指定消费组，同一个组的多个消费者共享分区，不会重复消费（前提是 offset 提交正常）。

要检查一个 topic 当前有多少条消息、lag 是多少，用这个：

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

输出里的LAG列就是消费者落后生产者的消息条数，这个数值在 AI 数据处理场景里是核心监控指标。下面关于延迟的很多排查，都是从这一列开始的。

4.2 重复消费到底是怎么发生的

“Kafka 能重复消费吗”是群里被问烂了的问题。先说结论：能，而且生产环境里几乎一定会遇到。Kafka 自己默认是至少一次语义，不是精确一次。当消费端处理完消息、但还没来得及提交 offset 就挂了，重启后消费者会从上次提交的 offset 重新拉取，这批消息就会被再处理一遍。

解决重复消费的办法，按优先级我建议这么做：第一，让消费逻辑保持幂等，这是治本。比如处理结果写数据库时，用唯一键做约束，重复写只更新不新增；调用模型接口时，往消息体里带 request_id，模型服务可以按这个 id 做结果去重。第二，把enable.auto.commit设为 false，手动确认处理成功后再提交提交 offset。第三，特殊场景可以用 Kafka 的事务 API，但维护成本高，不建议一上来就上。

我们这次接入 AI 时，就被重复消费坑过一次。当时消费者把消息转发给大模型接口，接口超时重试导致同一条消息被模型算了两遍，用户看到了两个一样的分析结果。后来我们给消费处理逻辑加了“基于 trace_id 的分布式锁”，重复消息直接丢弃，问题就消停了。

4.3 延迟高和 lag 居高不下的排查思路

消息延迟高、lag 一直涨，是 Kafka 运维里最磨人的问题之一。它的表象很单一，但根因可能千奇百怪。我把这次实践中“lag 排查”的过程整理成了一套固定动作，你照着做基本能把问题找出来。

第一步，先确认 lag 是不是真的在涨。用kafka-consumer-groups.sh --describe看LAG列的数值变化。如果 lag 在上涨，继续下一步；如果只是偶尔波动，可能是正常流量高峰。

第二步，看消费者是否在正常工作。检查消费进程的 CPU、内存和 GC 日志。AI 场景里最常见的问题是：消费者拿到消息后要去调用大模型接口，接口耗时长，导致单个消费线程处理一条消息要好几秒甚至几十秒。这时候如果单分区只有一个消费线程，上游一直写，lag 自然只涨不跌。解决办法是增加该 topic 的分区数，然后增加消费者实例或线程数。

第三步，检查消费者的max.poll.interval.ms和max.poll.records。Kafka 消费者从 poll 一批消息到处理完，必须在max.poll.interval.ms内发起下一次 poll，否则消费组会认为该消费者已死，触发 rebalance。AI 推理这种耗时操作很容易超过默认的 5 分钟，导致消费者被频繁踢出消费组、不断 rebalance、lag 飙升。我们的调法是：max.poll.records从默认的 500 调低到 50，单批数据变小，处理时间就够；max.poll.interval.ms从 300000 调到 600000，给模型调用留足时间。

第四步，如果消费者和生产者本身都健康，就要检查 broker 侧。磁盘 IO 是否打满、网络带宽是否跑满，都会造成整个集群吞吐下降。这种问题往往不是某一个 topic 的问题，而是所有 topic 都一起变慢，要在监控面板上横向对比。

5. 接入 AI 侧：Spring Kafka 与大模型的无缝对接

5.1 Spring Kafka 配置示例

我们团队大多数服务是 Java 写的，接入 Kafka 用的自然就是 Spring Kafka。配置文件里最关键的几项，直接贴出来。

spring: kafka: bootstrap-servers: kafka01:9092,kafka02:9092,kafka03:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer properties: enable.idempotence: true acks: all consumer: group-id: ai-consumer-group key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: "*" max.poll.records: 50 max.poll.interval.ms: 600000 enable.auto.commit: false auto-offset-reset: latest listener: ack-mode: manual_immediate

这里有几个关键点需要你理解，不只是抄配置。

acks: all配合enable.idempotence: true，保证生产端消息不丢不重。AI 场景下丢一条消息，意味着一次用户请求没有被处理，影响体验；重复消息可以通过消费端幂等消化，所以生产端尽量都往高了配。

enable.auto.commit: false和ack-mode: manual_immediate配合使用，意味着消息处理成功后才手动确认。这样做能最大程度防止“消息还在处理中、consumer 挂了、重启后丢消息”。但代价是重复消费概率增加，所以消费端一定要做幂等。

auto-offset-reset: latest表示没有提交记录时，从最新消息开始消费。如果你希望消费者启动后把积压的消息全部处理一遍，就改成earliest。我们在做数据回放时会把消费组的这个配置临时改成earliest，跑完再改回来。

5.2 消费消息并调用大模型的核心代码

在此基础上，消费端的代码骨架大概是这个样子：

@Component public class AiRequestConsumer { @Autowired private AiModelClient aiModelClient; @Autowired private KafkaTemplate<String, Object> kafkaTemplate; @KafkaListener(topics = "inference_request", groupId = "ai-consumer-group") public void onMessage(String message, Acknowledgment ack) { try { // 1. 解析请求消息，提取 requestId、用户输入、上下文 AiRequest request = JsonUtils.parse(message, AiRequest.class); // 2. 调用大模型接口 AiResult result = aiModelClient.invoke(request); // 3. 把结果写回结果 topic，由下游服务回调给调用方 kafkaTemplate.send("inference_result", request.getRequestId(), result); // 4. 处理成功，手动提交 offset ack.acknowledge(); } catch (Exception e) { // 记录失败消息到本地文件或死信 topic，后续人工处理 log.error("ai request process failed, message: {}", message, e); } } }

这段代码的核心设计思想就是“先记录下来，再处理；宁可后面重试，也不让消息悄悄丢掉”。调用大模型接口时，设置超时非常重要。我们线上模型接口 P99 要 5 秒，超时设成了 8 秒，但消费端的 poll 循环间隔还要留足够余量，这就是前面配置里max.poll.interval.ms调大的原因。

实际运行下来，我还发现一个现象：模型服务偶发超时，如果直接 catch 异常不提交 offset，Spring Kafka 会不断重试同一批消息，导致整个消费进度卡死。解决办法是：捕获到单个消息的异常后，不要直接抛出让 listener 容器重试整批，而是记录下来，继续ack，然后用一个单独的“失败重试表”去跟踪这些消息，定时重放。我们内部叫它“旁路重试”，比黏在 Kafka 消费链路里重试要优雅得多。

5.3 别让流量直接打爆模型服务

Kafka 接入 AI 后还有一个很容易被忽略的问题：模型服务不是无限吞吐的。你往inference_requesttopic 里塞了一万个请求，消费端确实会老老实实拉出来，但如果消费端不加控制地疯狂调用模型接口，很可能把模型服务打挂。

我们用的办法是在消费端内置一个简单的并发控制。比如用Semaphore限制同时调用模型接口的并发数：

private final Semaphore semaphore = new Semaphore(32); public AiResult invokeWithLimit(AiRequest request) { semaphore.acquire(); try { return aiModelClient.invoke(request); } finally { semaphore.release(); } }

这比靠 Kafka 消费者数量控制并发要直接。消费者线程数再高，模型服务的并发上限还是掌握在你手里的。另外，如果模型服务支持批量预测，尽量改成批量调用。Kafka 消费端可以攒够 N 条消息或者超时阈值后再统一调用模型接口，这种“攒批”模式能大幅提高模型服务的吞吐，但会增加单条消息的处理延迟。具体怎么取舍，取决于你的业务对实时性要求高不高。

6. 常见问题与踩坑实录

6.1 一张表解决典型问题定位

把最近项目里大家问得最多、我们实际踩过的坑整理成一张速查表，照着查比翻文档快。

那张表是按“症状 -> 原因 -> 手段”来组织的，但实际排查时不能第一反应就去改参数，要先看现象，再看监控，最后改配置。我之前有同事一看到重复消费就调max.poll.interval.ms，结果 rebalance 少了，但重复消费还是没解决。最后发现是消费端 catch 异常后抛出，导致 Spring 容器不断重试同一批消息。所以先分清是“消费推进问题”还是“offset 提交问题”，很关键。

6.2 踩坑一：session.timeout 和 poll.interval 不是一回事

很多人分不清session.timeout.ms和max.poll.interval.ms，觉得都是“超时”，其实完全两码事。

session.timeout.ms是消费者和 broker 之间的心跳超时，如果消费者在这个时间内没有发送心跳，broker 就认为它挂了，发起 rebalance。它的意义在于“这个消费者还活着吗”。

max.poll.interval.ms是消费者两次 poll 之间允许的最大间隔，如果消费者在处理消息上花了太长时间、迟迟不去拉下一批，broker 也会认为它卡住了，把它移出消费组。它的意义在于“这个消费者还有能力处理新消息吗”。

在 AI 场景里，最需要关注的是后者。模型调用耗时不稳定，有时候一次推理就要十几秒，如果消费线程还按默认的 500 条一批去拉数据，一批处理完可能要几分钟，远超max.poll.interval.ms，消费者就被踢出了。我把这两个值理解成：前者看你心跳是否正常，后者看你干活是否来得及。干活来不及，光调整心跳参数没用。

6.3 踩坑二：Kafka 和 RabbitMQ 选型别选错了再回炉

团队里聊“Kafka 和 rabbitmq 的区别”非常多。我们的选型结论很简单：消息路由、复杂交换机、按 key 定向发送，选 RabbitMQ；海量日志、事件流、数据管道、流式处理、消费回放，选 Kafka。

但实际项目里最怕的是“用 Kafka 的架构去解决 RabbitMQ 的诉求”。比如你只是想让“订单创建后通知三个服务”，这用 RabbitMQ 的 topic 交换机一套就能搞定；但你非要用 Kafka，就得自己建 topic、自己管理消费组和 offset，还要处理消费位点的问题。反过来，如果你的是“每天数亿条埋点日志，要实时清洗、离线回放”，RabbitMQ 就撑不住了，它没有 Kafka 那样的分区并行消费和日志保留机制。

我个人建议：拿不准的时候，优先问一句“我需要回放历史消息吗？”。需要回放，选 Kafka；不需要回放，且偏实时短生命周期的任务，用 RabbitMQ 更顺手。

6.4 踩坑三：可视化工具显示不了 topic

Docker 起的 Kafka，在 Kafka UI 里连上了，却看不到 topic，这也是常见问题。这个坑的根源不在工具，而在KAFKA_ADVERTISED_LISTENERS配置。容器内部的 broker 有它自己的地址，外部工具用localhost:9092访问时，broker 返回给自己的元数据如果还是容器内网地址，工具就没法建立真正的连接。解决办法就是在 broker 配置里把ADVERTISED_LISTENERS设置为外部可访问的地址，让 broker 在返回元数据时告诉客户端“你该连这个地址”。

如果你用了 Docker Compose 起 Kafka，又想在宿主机上用桌面版 Kafka Tool 连接，尽量把端口映射和ADVERTISED_LISTENERS都改成宿主机 IP 或localhost，不要图省事用服务名。这个问题排查难度不大，但很典型，因为它在容器环境里出现的频次实在太高了。

6.5 踩坑四：消费端线程和分区数的配比关系

Kafka 的性能调优里，讨论最多的问题是“消费端到底开几个线程”。我的认知是：一个分区最多只能被同一个消费组里的一个消费者实例消费，所以消费者线程的并行上限就是分区数。如果你 topic 只建了 3 个分区，就算你开了 30 个消费者线程，也只有 3 个在干活，其余 27 个完全是空闲的。

AI 场景里，我们消费inference_request这个 topic 时就吃过这个亏。一开始 topic 分区数是 3，消费端起了 10 个线程，以为并行度拉满了，结果监控显示 lag 一直涨。后来才意识到问题出在“分区数不够”。我们把分区数扩到了 12，消费者进程也扩到了 6 个，每个进程 2 个线程，吞吐立刻上来了。

所以建 topic 之前，先想清楚你的目标并行度。一般来说，分区数可以按“预估峰值 QPS / 单消费者处理能力”来估算，再留出 1.5 到 2 倍冗余。但也要注意，分区数也不是越多越好，分区太多会带来 broker 端文件句柄和 rebalance 成本的增加，要平衡。

这次把 Kafka 正式接入 AI 的过程，让我重新想清楚了一件事：很多人以为“AI 接入”就是写个接口调大模型，其实真正的难点全都藏在数据链路里。消息怎么可靠地进来、怎么不重不丢地送出去、消费速度怎么匹配模型服务的吞吐、出问题之后怎么回溯和重放，这些才是决定一个 AI 应用能不能稳定在线上跑起来的底子。Kafka 在中间扮演的，恰恰就是这个“数据底层”的角色。

最后再分享一个我在实测里悟出来的小技巧：接手一个 AI 数据管道之前，先把 topic 的消息存活时间、消费组 lag 和最大消费间隔这三个参数打到一个监控面板上。很多问题不用等你查日志，光看这三个数字就能猜个八九不离十。把底子打好，AI 才不会在关键时刻掉链子。