# springboot-kafka-verify **Repository Path**: enzoism/springboot-kafka-verify ## Basic Information - **Project Name**: springboot-kafka-verify - **Description**: Kafka快速验证Demo - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-10-30 - **Last Updated**: 2026-10-30 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # KAFKA 容器化部署与 Spring Boot 全链路验证 > 1-git仓库：https://gitee.com/enzoism/springboot-kafka-verify > 2-一套 Spring Boot 工程适配四种 Kafka 部署形态：单机/集群 × 无鉴权/SASL 认证。 > 3-本文记录四种形态的 Docker 部署设计、验证工程的代码结构、逐形态的全链路实测过程， > 4-以及每一条都真实踩过的坑。所有命令与输出均在本机（Docker Desktop + Apache Kafka 4.2.1）实测通过。 --- @[toc] --- ![](https://i-blog.csdnimg.cn/direct/76ff9f304eb2492aa120332bceee268c.jpeg) ## 1. 项目概览：2×2 部署矩阵 仓库由两部分组成： - `docker/` 下四套互斥的部署方案（同一时刻只启动一套），按「拓扑 × 认证」正交组合： | 目录 | 拓扑 | 认证 | 宿主机端口 | 容器名 | | ------------------------------------ | ------ | --------------- | ----------------------- | ------------------- | | `docker/docker-kafka-noauth` | 单机 | 无（PLAINTEXT） | `9092` | `kafka-noauth` | | `docker/docker-kafka-auth` | 单机 | SASL/PLAIN | `9093` | `kafka-auth` | | `docker/docker-cluster-kafka-noauth` | 3 节点 | 无（PLAINTEXT） | `19091 / 19092 / 19093` | `kafka-noauth-1..3` | | `docker/docker-cluster-kafka-auth` | 3 节点 | SASL/PLAIN | `29091 / 29092 / 29093` | `kafka-auth-1..3` | - `src/` 下一个 Spring Boot 2.7.18 验证工程（端口 8075），通过 Swagger 调试接口对当前连接的 Kafka 做消息收发、自定义分区器、Topic 管理的全链路验证。 四个目录的 README 是各自部署方案的「说明书」；本文是站在工程视角的「合订本」—— 重点讲清**为什么这么设计**、**代码怎么对接**、**怎么验证真的通了**。 --- ## 2. Docker 部署篇 ### 2.1 单机无鉴权：最简起步 `docker-kafka-noauth` 是整个矩阵的基线：KRaft 模式（Kafka 4.x 已彻底移除 ZooKeeper）， 单节点同时担任 broker + controller，只开两个监听器： ```yaml KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 ``` - `PLAINTEXT://9092`：客户端入口，宿主机与容器内同端口直达，无需拆分； - `CONTROLLER://9093`：KRaft 仲裁专用，仅容器内部网络可达，不对宿主机暴露。 宿主机客户端一行即可连通：`localhost:9092`。 ### 2.2 单机 SASL 认证：双监听器 + JAAS `docker-kafka-auth` 在基线上引入 `SASL_PLAINTEXT + SASL/PLAIN`（本地开发不开 TLS， 只做「认人」不做加密）。它暴露了一个关键问题：**一个监听器的 advertised 地址无法同时满足 两个网络视角**。 **为什么必须拆双监听器？** 客户端连上 bootstrap 后，会拿到 broker 的 advertised 地址做后续 通信。宿主机走端口映射（9093），容器内走内部网络（9092）——单监听器无论 advertise 哪个地址， 另一个视角都连不通。所以拆成三监听器： ```yaml KAFKA_LISTENERS: BROKER://0.0.0.0:9092,EXTERNAL://0.0.0.0:9093,CONTROLLER://0.0.0.0:9094 KAFKA_ADVERTISED_LISTENERS: BROKER://kafka-auth:9092,EXTERNAL://localhost:9093 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT KAFKA_INTER_BROKER_LISTENER_NAME: BROKER ``` | 监听器 | 端口 | 协议 | 服务对象 | | ------------ | ---- | -------------- | ---------------------------------------- | | `BROKER` | 9092 | SASL_PLAINTEXT | 容器内/节点间（advertised 用容器主机名） | | `EXTERNAL` | 9093 | SASL_PLAINTEXT | 宿主机（advertised 用 `localhost`） | | `CONTROLLER` | 9094 | PLAINTEXT | KRaft 仲裁，仅内部网络 | **账号管理**：SASL/PLAIN 的用户清单直接写在 JAAS 配置里（`user_<账号>="<密码>"` 语法）， 账密存放在 `.env`（已被 .gitignore 排除，不入库），由 compose 插值注入： ```yaml KAFKA_ADMIN_USER: 管理员（admin / admin123） # 运维脚本 / healthcheck / 建、删 Topic KAFKA_CLIENT_USER: 业务账号（client / client123） # 应用收发消息 ``` ```yaml KAFKA_LISTENER_NAME_BROKER_PLAIN_SASL_JAAS_CONFIG: >- org.apache.kafka.common.security.plain.PlainLoginModule required username="${KAFKA_ADMIN_USER}" password="${KAFKA_ADMIN_PASSWORD}" user_${KAFKA_ADMIN_USER}="${KAFKA_ADMIN_PASSWORD}" user_${KAFKA_CLIENT_USER}="${KAFKA_CLIENT_PASSWORD}"; ``` env 变量名到 Kafka 属性名的映射规则是「下划线转点、小写」： `KAFKA_LISTENER_NAME_BROKER_PLAIN_SASL_JAAS_CONFIG` → `listener.name.broker.plain.sasl.jaas.config`。 `username/password` 是本 broker 作为 SASL 客户端（节点间通信）用的账号，必须是已登记用户。 ### 2.3 三节点集群（无鉴权）：KRaft 仲裁 `docker-cluster-kafka-noauth` 把单机复制为三份：每节点仍兼任 broker + controller， 三节点通过 `CONTROLLER://29093` 组成仲裁： ```yaml KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-noauth-1:29093,2@kafka-noauth-2:29093,3@kafka-noauth-3:29093 ``` 集群形态下端口策略与单机同理再叠加一层：每节点 `INTERNAL://9092`（节点间）+ `EXTERNAL://19092`（对外），宿主机分别映射到 **19091 / 19092 / 19093**，advertised 各自指向 `localhost:1909x`。宿主机客户端 bootstrap 写全三个地址： ``` localhost:19091,localhost:19092,localhost:19093 ``` 集群特有的可用性参数（单机版无意义，集群版是可用性底线）： ```yaml CLUSTER_ID: "yLkS49RrdRghyrqXnQ58-Q" # 三节点必须一致，变量名不是 KAFKA_CLUSTER_ID！ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 # __consumer_offsets 副本 3 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 # 事务日志最小同步副本 2 KAFKA_DEFAULT_REPLICATION_FACTOR: 2 # 普通 Topic 默认副本 2 KAFKA_MIN_INSYNC_REPLICAS: 1 # 本地开发更宽容 KAFKA_NUM_PARTITIONS: 3 # 默认分区数 3 ``` ### 2.4 三节点集群（SASL 认证） `docker-cluster-kafka-auth` = 2.2 的认证机制 × 2.3 的集群拓扑，宿主机端口升到 **29091 / 29092 / 29093**。每个节点都有两段内容相同的 JAAS（`listener.name.internal...` 与 `listener.name.external...`），INTERNAL（节点间）与 EXTERNAL（宿主机）都要求认证， CONTROLLER 仲裁保持 PLAINTEXT（仅 compose 内部网络可达，无公网暴露面）。 ### 2.5 deploy.sh：统一的启停入口 四套目录各带一个同构的 `deploy.sh`，四个子命令覆盖全部生命周期： | 命令 | 行为 | | ------------------------ | ------------------------------------------------------------ | | `./deploy.sh fresh [-y]` | 全新数据启动：停容器 → 清空并重建 `data/kafkaN` → 重启（带交互确认） | | `./deploy.sh start` | 存量数据启动（首次部署也用它）；**改过 `.env` 后也用它重建容器使新账密生效** | | `./deploy.sh stop` | 停止并移除容器（数据保留） | | `./deploy.sh status` | 容器列表 + Broker 连通性 + KRaft 仲裁状态（认证版自动带凭证） | 几个值得抄走的脚本细节： - **端口预检**：启动前用 `lsof` 检查端口占用，且能识别「本 compose 项目自己已发布的端口」， 重建容器时不会被自己的旧容器拦住； - **凭证注入方式**：认证版镜像启动后会消费掉 `KAFKA_*` 环境变量，容器运行时取不到， 所以脚本用 `docker exec -e KU=... -e KP=...` 从宿主机 `.env` 现场注入生成凭证文件； - **跨平台数据目录属主**：macOS 由 Docker Desktop 自动处理，Linux 下自动 `chown 1000:0`。 --- ## 3. Spring Boot 验证工程篇 ### 3.1 设计核心：KafkaProperties 单点配置 工程对接四种形态时，Java 代码**零改动**，靠的是一个配置结构上的决定： **所有 Kafka 客户端通道都从 Spring Boot 的 `KafkaProperties` 构建**。 三个配置类分别覆盖三条通道，全部注入同一个 `KafkaProperties`： | 配置类 | 通道 | 构建方式 | | ------------------------ | ------------------------------------- | -------------------------------------- | | `KafkaBaseConfiguration` | Admin（`KafkaAdmin` + `AdminClient`） | `properties.buildAdminProperties()` | | `KafkaProducerConfig` | Producer（三个 `KafkaTemplate`） | `properties.buildProducerProperties()` | | `KafkaConsumerConfig` | Consumer（JSON 监听容器工厂） | `properties.buildConsumerProperties()` | 而 `KafkaProperties` 的规则是：`spring.kafka.bootstrap-servers` 与 `spring.kafka.properties` （全局属性 map）会**同时追加进这三条通道**。于是「换部署形态」被压缩成改 yaml 一处—— 换端口改 `bootstrap-servers`，上认证往全局 `properties` 里加三行 SASL，完事。 ### 3.2 代码走读 **生产者：三个 KafkaTemplate**（`KafkaProducerConfig`） 由于 Spring Boot 自动配置的 `kafkaTemplate` 是 `@ConditionalOnMissingBean`，自定义任意 Template 后自动配置即退避，所以用到的模板全部显式声明： - `kafkaTemplate`：String/String，普通消息（`/kafka/message/save`）； - `partitionKafkaTemplate`：挂载自定义分区器 `MyPartitioner`（`/kafka/message/save2`）； - `userKafkaTemplate`：`JsonSerializer` 序列化 `UserModel`（`/kafka/user/save`）。 **自定义分区器**（`MyPartitioner`）：有 key 时按 murmur2 哈希取模（与默认 `DefaultPartitioner` 策略一致，用于演示分区器的接入机制），无 key 时随机： ```java if (keyBytes == null) { return ThreadLocalRandom.current().nextInt(numPartitions); } return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; ``` **消费者：安全反序列化 + 错误隔离**（`KafkaConsumerConfig`） `userTopic` 专用容器工厂，两个防御性设计： 1. `JsonDeserializer<>(UserModel.class)` **固定目标类型，不信任消息头里的类型信息**， 杜绝反序列化攻击面；外层再包 `ErrorHandlingDeserializer`，脏消息不会让消费者无限重试； 2. `DefaultErrorHandler(new FixedBackOff(1000L, 3L))`：反序列化/业务异常重试 3 次（间隔 1 秒） 后跳过并记录日志，避免卡死整个分区。 **监听器**（`ConsumerListener`）：两个 `@KafkaListener`， `test-topic` 走 Boot 自动配置的默认工厂（StringDeserializer），`userTopic` 指定 `containerFactory = "userKafkaListenerContainerFactory"`。消费组统一为 `testGroup`， `auto-offset-reset: earliest`，自动提交间隔 100ms。 **调试接口**（`KafkaController`，文档页 `http://localhost:8075/doc.html`）： | 接口 | 验证目标 | | ---------------------------------------- | ------------------------------------------------------------ | | `GET /kafka/message/save?message=` | Producer 通道（String/String） | | `POST /kafka/user/save`（JSON body） | Producer + Consumer JSON 全链路（`JsonSerializer` → `JsonDeserializer`） | | `GET /kafka/message/save2?message=&key=` | 自定义分区器（带 key 走哈希，不带 key 随机） | | `GET /kafka/topic/create?topicName=` | AdminClient 通道（固定 2 分区 1 副本） | | `GET /kafka/topic/query?topicName=` | AdminClient 查询（分区/副本/ISR） | ### 3.3 application.yaml 适配要点 当前 yaml 指向 `docker-cluster-kafka-auth`（3 节点 SASL 集群），完整适配只涉及两个块： ```yaml spring: kafka: # 三节点 EXTERNAL 监听（宿主机入口），三个地址全列入 bootstrap， # 任一节点故障客户端仍可从其余节点获取元数据 bootstrap-servers: localhost:29091,localhost:29092,localhost:29093 properties: # 全局属性：admin/producer/consumer 三通道同时生效 "[security.protocol]": SASL_PLAINTEXT "[sasl.mechanism]": PLAIN "[sasl.jaas.config]": >- org.apache.kafka.common.security.plain.PlainLoginModule required username="${KAFKA_CLIENT_USER:client}" password="${KAFKA_CLIENT_PASSWORD:client123}"; ``` 三个容易踩的细节： 1. **带点号的 key 必须用 `"[...]"` 括号写法**。Boot 2.7 绑定 Map 时，不带括号的 `security.protocol: xxx` 会被当成嵌套结构 `{security: {protocol: ...}}`，绑定结果静默失效； 2. **JAAS 用 `>-` 折叠为单行**，最终等价于 `...PlainLoginModule required username="client" password="client123";`； 3. **账密走环境变量占位符 + 缺省值**（`${KAFKA_CLIENT_USER:client}`），变量名与 docker 侧 `.env` 对齐：默认值保证开箱即跑；`.env` 改账密后 `export KAFKA_CLIENT_USER=xxx KAFKA_CLIENT_PASSWORD=xxx` 再启动应用即可，无需改代码。 切换部署形态时的改动量： | 目标形态 | 改动 | | ----------- | ------------------------------------------------------------ | | 单机 noauth | `bootstrap-servers: localhost:9092`，删除 `properties` 块 | | 单机 auth | `bootstrap-servers: localhost:9093`，保留 SASL 块 | | 集群 noauth | `bootstrap-servers: localhost:19091,localhost:19092,localhost:19093`，删除 `properties` 块 | | 集群 auth | `bootstrap-servers: localhost:29091,localhost:29092,localhost:29093`，保留 SASL 块 | --- ## 4. 全链路验证手册 统一验证流程（以集群 SASL 版为例，其余形态同构替换端口/凭证即可）： **Step 1：启动集群并等待健康** ```bash cd docker/docker-cluster-kafka-auth ./deploy.sh start docker ps --filter name=kafka-auth- --format '{{.Names}} {{.Status}}' # kafka-auth-1 Up 52 seconds (healthy) # 三个节点必须全部 (healthy)—— # kafka-auth-2 Up 52 seconds (healthy) # 认证版 healthcheck 自带凭证，能过即代表 SASL 通道已通 # kafka-auth-3 Up 52 seconds (healthy) ``` **Step 2：启动应用**（`mvn spring-boot:run`；8075 被占用时加 `-Dspring-boot.run.arguments=--server.port=18075`） 启动日志看到分区分配即代表认证通过（未认证时拿不到元数据，只会反复超时）： ``` Started KafkaApplication in 0.804 seconds testGroup: partitions assigned: [test-topic-0, test-topic-1] testGroup: partitions assigned: [userTopic-2] ``` **Step 3：五个接口全打一遍** ```bash curl 'localhost:18075/kafka/message/save?message=cluster-auth-hello' curl -X POST -H 'Content-Type: application/json' -d '{"name":"enzo","age":30}' 'localhost:18075/kafka/user/save' curl 'localhost:18075/kafka/message/save2?message=part-msg-1&key=k1' curl 'localhost:18075/kafka/topic/create?topicName=cluster-sasl-verify' curl 'localhost:18075/kafka/topic/query?topicName=cluster-sasl-verify' ``` **Step 4：三路证据核验** 应用日志（发送 acked + 消费闭环，注意分区器把 key=k1 正确路由到了 partition 2）： ``` 消息发送成功 topic=test-topic, partition=1, offset=0 收到String消息 topic=test-topic, partition=1, offset=0, value=cluster-auth-hello 消息发送成功 topic=test-topic, partition=2, offset=0 # key=k1 → partition 2 收到User消息 topic=userTopic, partition=2, offset=0, value=UserModel(id=null, age=30, name=enzo) ``` Topic 查询接口（集群形态下副本/Leader 应分散在不同节点）： ``` partition=0, leader=localhost:29091 (id: 1) # 节点1 partition=1, leader=localhost:29092 (id: 2) # 节点2 ``` Broker 侧消费位移（全部分区 LAG=0 才算闭环）： ```bash # 容器内生成凭证文件后查询（deploy.sh status 内部就是这么做的） docker exec kafka-auth-1 /opt/kafka/bin/kafka-consumer-groups.sh \ --bootstrap-server localhost:9092 --command-config /tmp/c.properties \ --describe --group testGroup # test-topic 0 offset=16 log-end=16 LAG=0 # test-topic 1 offset=1 log-end=1 LAG=0 # userTopic 2 offset=1 log-end=1 LAG=0 ``` **Step 5（集群加练）：故障演练** `docker stop kafka-auth-3` 后重复 Step 3，宿主机收发不中断（新消费组找 coordinator 的 几秒过渡期内可能报连接错误，重试即可）；`docker start` 后节点重新入列，ISR 恢复完整。 --- ## 5. 踩坑清单（每一条都实践过） 1. **认证版 healthcheck 必须带凭证，且只能用 compose 插值烘焙**：healthcheck 命令里的账密要写成 单美元 `${KAFKA_ADMIN_USER}`（compose 创建容器时从 `.env` 烘焙进命令）。写成 `$${KAFKA_ADMIN_USER}`（运行时容器内取环境变量）会拿到**空值**——Kafka 镜像启动后会把 `KAFKA_*` 环境变量消费掉。症状：broker 正常运行、容器永远 `unhealthy`、日志刷 `Failed authentication ... username not specified`。 2. **无凭证/错误凭证的报错形态不同**：无凭证连接表现为 `TimeoutException`/bootstrap 超时 （broker 拒绝握手后客户端反复重试），**错误密码**才有明确的 `SaslAuthenticationException: Invalid username or password`。排查先看 broker 日志 `data/kafkaN/logs/server.log` 里的 `Failed authentication` 行。 3. **改 `.env` 后必须重建容器才生效**：`./deploy.sh start`（底层 `docker compose up -d` 会因环境变化自动 recreate）。只 `docker restart` 不会重新计算配置。实测中出现过 「healthcheck 通过但脚本带 `.env` 新密码连不上」——就是容器还持着旧账密。 4. **集群 ID 的环境变量名是 `CLUSTER_ID`，不是 `KAFKA_CLUSTER_ID`**：写错不报错但**静默不生效** （镜像用默认 ID 当成 `cluster.id` 忽略掉）。三节点共享同一默认值时集群照样能跑，但与配置 不符；改过 `CLUSTER_ID` 后需 `./deploy.sh fresh`（旧数据按旧 ID 格式化，混用起不来）。 5. **Boot 2.7 的 Map 绑定括号写法**：见 3.3 第 1 条。配置不生效但应用不报错，问题会被 误诊为「broker 侧没配好」，先检查 yaml 写法再查 broker。 6. **容器内客户端禁止用宿主机端口**：`localhost:19091/29091` 这类是宿主机映射端口，容器内 并无监听，误用表现为 AdminClient 反复 `Connection to node -1 could not be established` 后超时。容器内一律走 INTERNAL 监听（`localhost:9092` 或 `kafka-noauth-N:9092`）。 7. **多个应用实例共用同一消费组会分走分区**：本机同时跑 IDEA 实例（8075）和验证实例（18075） 时，两组消费者同属 `testGroup`，3 分区 Topic 的分区会被两边瓜分——表现为「我发的消息 我的日志里没消费记录」。不是配置问题，是消费组负载均衡的正常行为；核验时以 broker 侧 LAG 为准，或只保留一个实例。 8. **端口冲突要提前想好退路**：`docker-cluster-kafka-*` 的 19091-19093/29091-29093 若被旧部署 占用，`deploy.sh` 会预检报错；应用侧 8075 被占用则换端口验证（见 Step 2）。 --- ## 6. 速查表 | | 单机 noauth | 单机 auth | 集群 noauth | 集群 auth | | --------- | --------------------- | ------------------------------------------ | ----------------------------- | ------------------------------------------ | | 目录 | `docker-kafka-noauth` | `docker-kafka-auth` | `docker-cluster-kafka-noauth` | `docker-cluster-kafka-auth` | | bootstrap | `localhost:9092` | `localhost:9093` | `localhost:19091,19092,19093` | `localhost:29091,29092,29093` | | 协议 | PLAINTEXT | SASL_PLAINTEXT + PLAIN | PLAINTEXT | SASL_PLAINTEXT + PLAIN | | 账号 | 无 | admin/admin123、client/client123（`.env`） | 无 | admin/admin123、client/client123（`.env`） | | 应用 yaml | 只改 bootstrap | bootstrap + SASL 块 | 只改 bootstrap | bootstrap + SASL 块 | 一条龙验证（集群 SASL 版）： ```bash cd docker/docker-cluster-kafka-auth && ./deploy.sh start # 等三节点 healthy cd ../.. && mvn spring-boot:run # 应用默认 8075，doc.html 调试 open http://localhost:8075/doc.html ``` > 安全提醒：SASL_PLAINTEXT 只「认人」不加密，账密明文过网络，四套方案均仅限本机开发， > 不要把端口映射到公网。生产环境请上 SASL_SSL。