# spring-kafka

**Repository Path**: java-study_9/spring-kafka

## Basic Information

- **Project Name**: spring-kafka
- **Description**: kafka生产环境使用
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-21
- **Last Updated**: 2026-05-26

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Kafka Demo

一个基于 Spring Boot 3 + Spring Kafka 的多模块示例工程，采用“生产者应用 / 消费者应用 / 公共组件”分离架构，覆盖以下能力：

- 生产者 REST 接口发送 Kafka 消息
- 统一消息信封 `KafkaEnvelope`
- 消费者监听、手动 ack、指数退避重试、DLT
- 显式 Topic 声明，避免 Topic 契约不确定
- 基于 Redis 的消费端幂等保护
- 基于 Redis ZSET 的可靠延时消息投递

更详细的架构说明见 [docs/ARCHITECTURE.md](docs/架构设计说明.md)。

Kafka 面试题：[大白话详解版（推荐，含答案+排障+配置）](KAFKA-面试题-大白话版.md) · [题目全集](KAFKA-面试题大全.md) · [正式参考答案](KAFKA-面试题大全-参考答案.md)

Agent 开发本实战时引用项目 Skill（支付→Kafka→履约）：

| 文档 | 用途 |
|------|------|
| [SKILL.md](.cursor/skills/kafka-order-payment-fulfillment/SKILL.md) | 主流程、目录铁律、Agent 检查清单 |
| [reference.md](.cursor/skills/kafka-order-payment-fulfillment/reference.md) | 契约、生产加固、Outbox/Saga、配置 |
| [examples.md](.cursor/skills/kafka-order-payment-fulfillment/examples.md) | 联调与测试用例 |
| [runbook.md](.cursor/skills/kafka-order-payment-fulfillment/runbook.md) | 排障 Runbook |

## 1. 项目概览

### 1.1 模块划分

| 模块 | 说明 | 端口 |
|------|------|------|
| `kafka-common` | 公共配置、消息模型、属性绑定、Topic 声明、序列化 | - |
| `kafka-product` | 生产者应用，对外暴露 REST 接口并发送 Kafka 消息 | `8082` |
| `kafka-consumer` | 消费者应用，监听 Kafka、处理消息、重试并写入 DLT | `8081` |
| `docker/` | Kafka KRaft 集群和 Redis 的本地部署脚本 | - |

### 1.2 典型消息链路

```text
HTTP 请求
  -> kafka-product Controller
  -> KafkaProducerService
  -> Kafka Topic
  -> kafka-consumer @KafkaListener
  -> 业务处理
  -> ack

失败时：
  -> DefaultErrorHandler 重试
  -> 超过最大重试后写入 {topic}.DLT
```

### 1.3 当前支持的关键能力

| 能力 | 当前实现 |
|------|----------|
| 发送模式 | 同步发送、异步发送、批量发送、默认 Topic 发送 |
| 领域事件 | `ORDER_CREATED`、`NOTIFICATION_SEND` |
| Topic 治理 | 启动时显式声明业务 Topic 与 DLT Topic |
| 消费确认 | 手动 ack 或按配置切换为记录级提交 |
| 消费失败处理 | 指数退避重试 + DLT |
| 消费幂等 | Redis 完成标记 + 处理中锁 |
| **广播消费（可选）** | `kafka.demo.consumer.broadcast.enabled=true` 时，每实例独立 group，各收全量消息（`KafkaBroadcastMessageListener` 仅日志） |
| 延时消息 | Redis `queue/payload/processing/failed` 四段式可靠投递 |
| **支付→履约实战** | `web/PaymentNotificationController` + `service/payment` 发 MQ；`kafka-consumer/service/order` 消费并履约 |

### 1.4 支付通知 → 订单履约（业务实战）

```text
第三方支付回调 POST /api/payment/callback  (kafka-product:8082)
  -> PaymentNotificationController -> PaymentNotificationService 校验订单/金额
  -> KafkaEnvelope(PAYMENT_SUCCEEDED) + key=orderId
  -> Topic: demo.order
  -> OrderPaymentKafkaListener (kafka-consumer:8081)
  -> Redis 消费幂等 (messageId)
  -> OrderFulfillmentOrchestratorService
        1. 订单状态 CREATED -> PAID -> FULFILLING -> COMPLETED
        2. 扣减库存
        3. 扣减用户余额 (paymentId 幂等)
        4. 发货状态 -> PENDING_PICK
  -> manual ack
```

**代码位置：**

| 端 | 包路径 | 说明 |
|----|--------|------|
| product | `com.kafka.product.web.PaymentNotificationController` | 支付回调 HTTP 入口 |
| product | `com.kafka.product.service.payment` | 支付业务、`DemoOrderCatalog`、发 MQ |
| consumer | `com.kafka.consumer.service.order` | 监听、编排、订单/库存/账户/发货 |
| common | `PaymentSucceededPayload`、`OrderPaymentEventType` | 跨模块消息契约 |

**联调示例订单（已内置）：**

- `ORD-20260522-001` / 用户 `U-10001` / 金额 `299.00`
- `ORD-20260522-002` / 用户 `U-10002` / 金额 `59.90`

```bash
# 1. 启动 consumer、product（先 consumer）

# 2. 模拟支付成功回调
curl -X POST http://localhost:8082/api/payment/callback \
  -H "Content-Type: application/json" \
  -d "{\"orderId\":\"ORD-20260522-001\",\"userId\":\"U-10001\",\"paymentId\":\"PAY-001\",\"paymentChannel\":\"WECHAT\",\"amount\":299.00,\"currency\":\"CNY\",\"status\":\"SUCCESS\",\"traceId\":\"trace-pay-001\"}"

# 3. 查看履约结果
curl http://localhost:8081/api/order/fulfillment/ORD-20260522-001
```

## 2. 环境要求

- JDK 17+
- Maven 3.6.3+
- Docker / Docker Compose
- Redis

说明：

- 当前仓库父工程使用 `Spring Boot 3.4.5`，因此必须使用 `JDK 17+`
- 如果启用消费端幂等或延时消息，Redis 是必需依赖

## 3. 目录结构

```text
kafka-demo/
├── kafka-common/
│   └── src/main/java/com/kafka/common/
├── kafka-product/
│   └── src/main/java/com/kafka/product/
├── kafka-consumer/
│   └── src/main/java/com/kafka/consumer/
├── docker/
│   └── docker-compose.yml
├── docs/
│   └── ARCHITECTURE.md
└── README.md
```

## 4. 核心设计

### 4.1 统一消息信封

Kafka 消息体统一使用 `KafkaEnvelope<T>`：

```json
{
  "messageId": "8cf1a7ce-4a1d-4f74-b44b-9f1b6f2f66bf",
  "traceId": "2cb7f5f6-d5f3-4124-bf8b-8f51b665a4aa",
  "eventType": "ORDER_CREATED",
  "timestamp": "2026-05-21T10:00:00Z",
  "payload": {
    "orderId": "ORD-10001",
    "userId": "U-1"
  }
}
```

字段含义：

- `messageId`：全局唯一消息 ID，用于消费端幂等与排查
- `traceId`：链路追踪 ID，用于跨 HTTP / Kafka 日志关联
- `eventType`：业务事件类型
- `timestamp`：消息生成时间
- `payload`：业务负载

### 4.2 Topic 约定

默认 Topic 配置：

| 逻辑名称 | 默认值 |
|----------|--------|
| 默认 Topic | `demo.default` |
| 订单 Topic | `demo.order` |
| 通知 Topic | `demo.notification` |
| DLT 后缀 | `.DLT` |

启动时会自动声明：

- `demo.default`
- `demo.order`
- `demo.notification`
- `demo.order.DLT`
- `demo.notification.DLT`

### 4.3 可靠延时消息

延时消息不直接使用 Kafka 原生能力，而是通过 Redis 实现；提交时使用 `delayValue` + `delayUnit`（如 30 + minute = 30 分钟后发送）：

| Redis Key | 类型 | 说明 |
|-----------|------|------|
| `kafka:delay:queue` | ZSET | 待投递任务，`score=executeAt` |
| `kafka:delay:payload` | HASH | 任务详情 |
| `kafka:delay:processing` | ZSET | 已认领未确认任务，`score=认领超时点` |
| `kafka:delay:failed` | HASH | 超过最大重试次数的失败任务 |

设计目标：

- 避免“先删 Redis 再发 Kafka”导致的消息丢失
- 调度器崩溃后，认领超时任务可自动回收
- 超过最大重试后保留失败任务，便于后续人工补偿或离线回放

### 4.4 消费端幂等

消费者通过 Redis 做重复抑制：

| Redis Key | 说明 |
|-----------|------|
| `kafka:consumer:idempotency:processing:{messageId}` | 正在处理中的消息锁 |
| `kafka:consumer:idempotency:completed:{messageId}` | 已完成处理的消息标记 |

处理逻辑：

1. 先检查 `completed` 标记
2. 如果没有，再尝试加 `processing` 锁
3. 处理成功后写入 `completed` 并清理 `processing`
4. 处理失败时清理 `processing`，允许后续重新消费

## 5. 本地启动

### 5.1 启动单节点 Kafka + Redis

在项目根目录执行：

```bash
docker compose -f docker/docker-compose.yml --profile single up -d kafka-single redis
```

如果要启动三节点 Kafka：

```bash
docker compose -f docker/docker-compose.yml up -d kafka-1 kafka-2 kafka-3 redis
```

### 5.2 编译

```bash
mvn clean package -DskipTests
```

### 5.3 启动消费者

```bash
mvn spring-boot:run -pl kafka-consumer -Dspring-boot.run.profiles=dev
```

### 5.4 启动生产者

```bash
mvn spring-boot:run -pl kafka-product -Dspring-boot.run.profiles=dev
```

### 5.5 本地端口

| 服务 | 端口 |
|------|------|
| `kafka-product` | `8082` |
| `kafka-consumer` | `8081` |

## 6. 主要接口

### 6.1 kafka-product API

| 方法 | 路径 | 说明 |
|------|------|------|
| POST | `/api/kafka/messages/sync` | 同步发送到任意 Topic |
| POST | `/api/kafka/messages/async` | 异步发送示例 |
| POST | `/api/kafka/messages/batch` | 批量发送到同一 Topic |
| POST | `/api/kafka/messages/default` | 发送到默认 Topic |
| POST | `/api/kafka/events/order-created` | 发送订单事件到 `demo.order` |
| POST | `/api/kafka/events/notification` | 发送通知事件到 `demo.notification` |
| POST | `/api/kafka/messages/schedule` | 提交延时消息任务 |
| POST | `/api/kafka/messages/schedule/demo-order` | 联调：15 秒后向 `demo.order` 投递支付成功履约消息 |
| GET | `/api/kafka/messages/schedule/pending` | 查询待执行延时任务数 |
| DELETE | `/api/kafka/messages/schedule/{taskId}` | 取消延时任务 |
| GET | `/api/kafka/health` | Kafka 可用性检查 |
| GET | `/api/kafka/topics` | 查看当前 Topic 映射 |

### 6.2 kafka-consumer API

| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/kafka/consumer/info` | 查看消费端配置摘要 |
| GET | `/api/kafka/consumer/diagnostics` | 查看 Topic、消费组、offset 等诊断信息 |

## 7. 请求示例

### 7.1 发送订单事件

```bash
curl -X POST http://localhost:8082/api/kafka/events/order-created \
  -H "Content-Type: application/json" \
  -d "{\"orderId\":\"ORD-10001\",\"userId\":\"U-1\",\"amount\":99.90,\"currency\":\"CNY\"}"
```

### 7.2 发送通知事件

```bash
curl -X POST http://localhost:8082/api/kafka/events/notification \
  -H "Content-Type: application/json" \
  -d "{\"userId\":\"U-1\",\"channel\":\"SMS\",\"title\":\"支付提醒\",\"content\":\"订单已支付\"}"
```

### 7.3 发送任意 Topic 消息

```bash
curl -X POST http://localhost:8082/api/kafka/messages/sync \
  -H "Content-Type: application/json" \
  -d "{\"topic\":\"demo.default\",\"key\":\"k1\",\"eventType\":\"USER_REGISTERED\",\"payload\":{\"userId\":\"U-1\"}}"
```

### 7.4 提交延时消息

使用 `delayValue` + `delayUnit`（支持 `second` / `minute` / `hour` / `day`，或 `秒` / `分钟` / `小时` / `天`）：

```bash
# 30 分钟后发送
curl -X POST http://localhost:8082/api/kafka/messages/schedule \
  -H "Content-Type: application/json" \
  -d "{\"topic\":\"demo.default\",\"key\":\"delay-k1\",\"eventType\":\"DELAYED_TEST\",\"delayValue\":30,\"delayUnit\":\"minute\",\"payload\":{\"msg\":\"hello\"}}"
```

**15 秒后投递到 `demo.order`（触发订单履约，无需请求体）：**

```bash
curl -X POST http://localhost:8082/api/kafka/messages/schedule/demo-order
# 约 15 秒后
curl http://localhost:8081/api/order/fulfillment/ORD-20260522-001
```

### 7.5 查询消费者信息

```bash
curl http://localhost:8081/api/kafka/consumer/info
```

### 7.6 查询消费者诊断

```bash
curl http://localhost:8081/api/kafka/consumer/diagnostics
```

## 8. 配置说明

### 8.1 生产者关键配置

文件：[kafka-product/src/main/resources/application.yml](kafka-product/src/main/resources/application.yml)

重点配置：

- `spring.kafka.bootstrap-servers`
- `spring.kafka.producer.acks=all`
- `spring.kafka.producer.retries=2147483647`
- `spring.kafka.producer.properties.enable.idempotence=true`
- `kafka.demo.producer.default-topic`
- `kafka.demo.delay.*`

### 8.2 消费者关键配置

文件：[kafka-consumer/src/main/resources/application.yml](kafka-consumer/src/main/resources/application.yml)

重点配置：

- `spring.kafka.consumer.group-id`
- `spring.kafka.consumer.auto-offset-reset`
- `spring.kafka.consumer.enable-auto-commit=false`
- `kafka.demo.consumer.concurrency`
- `kafka.demo.consumer.manual-ack`
- `kafka.demo.consumer.retry.*`
- `kafka.demo.consumer.idempotency.*`
- `kafka.demo.consumer.missing-topics-fatal=true`

### 8.3 Topic 治理配置

文件：

- [kafka-common/src/main/java/com/kafka/common/config/KafkaTopicProvisioningConfiguration.java](kafka-common/src/main/java/com/kafka/common/config/KafkaTopicProvisioningConfiguration.java)
- [kafka-common/src/main/java/com/kafka/common/properties/KafkaDemoTopicsProperties.java](kafka-common/src/main/java/com/kafka/common/properties/KafkaDemoTopicsProperties.java)

重点配置：

- `kafka.demo.topics.provision-enabled`
- `kafka.demo.topics.partitions`
- `kafka.demo.topics.replication-factor`

## 9. 消费失败与 DLT

消费端使用 `DefaultErrorHandler`：

1. 监听器抛异常
2. 进入指数退避重试
3. 超过最大重试次数后，写入 `{topic}.DLT`

默认 DLT：

- `demo.order.DLT`
- `demo.notification.DLT`

如果你要新增业务 Topic，也建议同步新增对应 DLT 处理策略。

## 10. 延时消息运维建议

如果启用了延时消息，建议重点关注：

- `kafka:delay:queue`：待执行任务数
- `kafka:delay:processing`：长时间积压说明调度器处理异常
- `kafka:delay:failed`：出现数据说明任务已经超过最大重试次数

建议做的监控：

- `pendingCount` 趋势
- `failed` 数量
- 调度器异常日志
- Kafka 发送失败次数

## 10.1 广播模式（多实例各收一份）

默认是**集群消费**：多个 `kafka-consumer` 实例共用 `kafka-demo-consumer-group`，**一条消息只给一个实例**（分摊分区）。

若要**广播**（A、B、C 三个实例都收到同一条消息）：

1. 在 `kafka-consumer` 的 `application.yml`（或环境变量）中设置：
   ```yaml
   kafka:
     demo:
       topics:
         provision-topics:
           - default
           - order
           - notification
           - test_broadcast
       consumer:
         broadcast:
           enabled: true
           instance-id: node-a   # 每个实例必须不同；不配则用 主机名-端口
           topics:
             - order
             - test_broadcast    # 也可直接写真实 Topic 名
   ```
2. 重启各实例，启动日志应出现 `broadcast ENABLED: group-id=kafka-demo-broadcast-xxx`。
3. 发一条支付回调后，各实例日志应都有 **`【广播消费】`**；订单履约仍只由集群监听器 `OrderPaymentKafkaListener` 执行一次（Redis 幂等）。

查看当前模式：`GET http://localhost:8084/api/kafka/consumer/info`

## 11. 排障清单

### 11.1 生产者发成功但消费者收不到

按顺序排查：

1. 确认 `kafka-product` 和 `kafka-consumer` 都已启动
2. 确认两边 `bootstrap-servers` 指向同一套 Kafka
3. 调用 `/api/kafka/health` 确认生产者能连上 Kafka
4. 调用 `/api/kafka/consumer/info` 和 `/api/kafka/consumer/diagnostics`
5. 确认 Topic 已声明成功
6. 查看消费者日志中是否出现重试或 DLT 相关错误

### 11.2 延时消息不执行

优先检查：

1. `kafka.demo.delay.enabled` 是否为 `true`
2. Redis 是否可达
3. `delayValue` 是否大于 0，`delayUnit` 是否为 second/minute/hour/day
4. `kafka:delay:processing` 是否积压
5. `kafka:delay:failed` 是否已有失败任务

### 11.3 重复消费

优先检查：

1. `messageId` 是否为空
2. `kafka.demo.consumer.idempotency.enabled` 是否为 `true`
3. Redis 是否可达
4. 业务代码是否在成功前就执行了不可逆副作用

## 12. 在新业务项目中复用

推荐方式：

1. 复用 `kafka-common` 中的公共协议、配置和 Topic 声明
2. 以 `kafka-product` 为发送模板
3. 以 `kafka-consumer` 为消费模板
4. 为每个新增业务 Topic 明确：
   - Topic 名称
   - 分区 key
   - 失败重试策略
   - DLT 处理策略
   - 幂等粒度

## 13. 后续扩展建议

如果要继续往生产方向增强，建议优先考虑：

- 接口鉴权与权限控制
- DLT 消息回放工具
- 延时失败任务后台查询与补偿接口
- 更完整的集成测试与端到端测试
- Micrometer / Prometheus 指标监控
- 灰度 Topic、版本化事件协议