2026/05/09/9 分钟/8 阅读/0 评论

Creeks 系统 Kafka 使用全景分析

整个系统是典型的 **微服务 CDC + 事件驱动** 架构，Kafka 在其中扮演了 **"中枢神经"** 的角色。这是所有服务的 **Kafka 基座**，封装了通用的生产者/消费者/工具类。这是整个系统数据同步的 **大动脉**，

分析日期：2026-05-08

一、整体架构

整个系统是典型的 微服务 CDC + 事件驱动 架构，Kafka 在其中扮演了 "中枢神经" 的角色。

graph TB
    subgraph 数据源
    DB[(MySQL)]
    end

    subgraph Maxwell
    MW[Maxwell Binlog Listener]
    end

    subgraph Kafka集群
        subgraph CDC数据Topic
        T1[cdc_api_mall_goods<br/>商品表变更]
        T2[cdc_api_pms_sku<br/>SKU变更]
        T3[cdc_api_pms_spu<br/>SPU变更]
        T4[cdc_api_pms_image<br/>图片变更]
        T5[cdc_inv_inv_stock<br/>库存变更]
        T6[cdc_inv_inv_winit_stock<br/>万邑通库存]
        T7[cdc_inv_inv_winit_declare_stock<br/>报关库存]
        T8[ebay_item<br/>eBay商品]
        T9[other_item<br/>其他平台商品]
        T10[stock-snapshot<br/>库存快照]
        end

        subgraph 业务事件Topic
        B1[package_num_calculate<br/>装箱计算]
        B2[receiving_order_event<br/>收货单事件]
        B3[return_warehouse_order<br/>退仓单]
        B4[winit_self_inspection<br/>验货结果]
        B5[fulfillment_carton_transported<br/>箱运输通知]
        B6[order_center_collect<br/>订单采集]
        B7[static_order_topic<br/>订单统计]
        E1[sku_update_sale_status<br/>SKU销售状态]
        end

        subgraph Amazon同步Topic
        C1[order_topic<br/>Amazon订单]
        C2[data_sync_topic<br/>数据同步]
        C3[finance_group_topic<br/>财务变更]
        end

        subgraph 错误收集Topic
        D1[codo_local_log_v3<br/>错误日志]
        end

        subgraph 基础日志
        L1[base-log topics<br/>基础数据变更日志]
        end
    end

    subgraph 微服务消费端
    CDC[creeks-cdc<br/>数据同步服务]
    ALLOC[creeks-allocation<br/>调拨服务]
    OC[creeks-order-center<br/>订单中心]
    EC[codo-error-collect<br/>错误收集]
    SA[creekshub-sync-amazon<br/>Amazon同步]
    end

    subgraph 存储
    ES[(Elasticsearch)]
    CK[(ClickHouse)]
    end

    DB --> MW
    MW --> T1
    MW --> T2
    MW --> T3
    MW --> T4
    MW --> T5
    MW --> T6
    MW --> T7
    MW --> L1

    T1 -.分流.-> T8
    T1 -.分流.-> T9
    T2 -.转发.-> E1

    T8 --> CDC
    T9 --> CDC
    T5 --> CDC
    T2 --> CDC
    L1 --> CDC

    T10 --> CDC

    B1 --> ALLOC
    B2 --> ALLOC
    B3 --> ALLOC
    B4 --> ALLOC

    B6 --> OC

    D1 --> EC

    C1 --> OC
    C2 --> SA
    C3 --> SA

    CDC --> ES
    CDC --> CK
    EC --> ES
    OC --> ES

二、Kafka 三大核心用途

用途	说明	涉及服务
CDC 数据库变更捕获	Maxwell 监听 MySQL binlog 写入 Kafka，creeks-cdc 消费后同步到 ES/ClickHouse	creeks-maxwell → creeks-cdc
微服务间业务事件解耦	服务 A 发消息，服务 B/C 异步消费，避免同步 Feign 调用	allocation, order-center, sync-amazon
日志采集与错误收集	业务系统错误日志丢 Kafka 缓冲，再写入 ES 防止直接写入被打挂	codo-error-collect

三、各服务 Kafka 使用详解

3.1 creeks-common-services — 公共基础设施

路径：E:\code\creeks-common-services

这是所有服务的 Kafka 基座，封装了通用的生产者/消费者/工具类。

3.1.1 生产者配置 `KafkaProducerCommonConfig`

配置项                    值                    说明
────────────────────────────────────────────────────────────────
acks                      all                   最强持久化，leader + 所有 ISR 副本确认
retries                   3                     发送失败最多重试 3 次
retry.backoff.ms          1000                  重试间隔 1 秒
key.serializer            StringSerializer      键序列化
value.serializer          StringSerializer      值序列化

3.1.2 消费者配置 `KafkaConsumerCommonConfig`

配置项                    值                    说明
────────────────────────────────────────────────────────────────
auto.offset.reset         latest                从最新偏移开始消费
enable.auto.commit        false                 手动提交偏移量
ack.mode                  MANUAL_IMMEDIATE      手动立即确认
concurrency               1                     单消费者并发

死信队列(DLT)机制：

消费失败 → 重试 3 次（间隔 1 秒）→ 再失败 → 发送到 .DLT 死信 Topic
同时支持单条消费(kafkaListenerContainerFactory)和批量消费(kafkaBatchListenerContainerFactory)两种容器工厂

3.1.3 KafkaHelper 发送工具

// 同步发送（阻塞等待 broker 确认）
kafkaHelper.send(topic, message);              // 默认分区
kafkaHelper.send(topic, partition, message);   // 指定分区
kafkaHelper.send(topic, key, message);         // 按 key 路由分区

// 异步发送（不阻塞）
kafkaHelper.sendAsync(topic, message);
kafkaHelper.sendAsync(topic, partition, message);
kafkaHelper.sendAsync(topic, key, message);

注意： send() 方法是同步的（调了 .get()），发送失败会抛 BusinessException。

3.2 creeks-cdc — CDC 数据变更捕获服务（核心大动脉）

路径：E:\code\creeks-cdc\server

这是整个系统数据同步的 大动脉，监听 Maxwell 写来的 MySQL binlog 变更，分发到不同下游。

graph LR
    A[MySQL Binlog] --> B[Maxwell]
    B --> C[Kafka Topic]
    C --> D[creeks-cdc<br/>Consumer]
    D --> E[(ES)]
    D --> F[(ClickHouse)]
    D --> G[转发到下游 Topic]

消费链路表

Topic	消费者类	groupId	作用	消费模式	max.poll.records
`cdc_api_mall_goods`	`GoodsSyncConsumer`	goods-sync-to-es	商品表变更 → 按平台(ebay/other)拆分转发 → 最终同步到 ES	批量	5000
`ebay_item`	`GoodsSyncConsumer`	ebay-goods-sync-to-es	eBay 商品数据 → 同步到 ES	批量	500
`other_item`	`GoodsSyncConsumer`	other-goods-sync-to-es	其他平台商品数据 → 同步到 ES	批量	500
`cdc_api_pms_sku`	`SkuSyncConsumer`	sku-snapshot-log	SKU 销售状态变更 → 转发到 `sku_update_sale_status`	批量	100
`cdc_inv_inv_stock`	`StockSnapshotConsumer`	stock-snapshot-log	库存表变更 → 写 ClickHouse 做库存快照	批量	100
`base-log topics`	`BaseLogConsumer`	cdc-base-log	基础数据变更日志 → 写入 ES（按月分索引）	批量	100

重点消费逻辑拆解

GoodsSyncConsumer — 商品同步（一个消费者既是消费者又是生产者）：

cdc_api_mall_goods 收到商品变更
  ├─ 判断 platform_name
  │   ├─ eBay → 发送到 ebay_item Topic
  │   └─ 其他 → 发送到 other_item Topic
  │
  ├─ ebay_item Topic 触发 → goodsBatchSyncToES()
  └─ other_item Topic 触发 → goodsBatchSyncToES()

BaseLogConsumer — 基础日志（过滤 + 清洗后写 ES）：

只关注特定表的变更：pms_combo_sku_relation、mall_goods_inventory、inv_stock 等
过滤 update_time 等无意义字段
ES 索引按月拆分：base-log-index-202605
inv_winit_stock 一对多关系，需要为关联货物都记录日志

3.3 creeks-allocation — 调拨服务

路径：E:\code\creeks-allocation\creeks-allocation-server

Kafka 在这里充当 异步任务队列 + 回调通知 的角色。

Topic 配置属性	消费者类	作用
`allocation.queue.package_num_calculate`	`CalculateStockPackageNumConsumer`	调拨单装箱数量计算
`allocation.queue.receiving_order_event`	`ReceivingOrderListener`	收货单上架结果回调
`allocation.queue.return_warehouse_order`	`PlatformReturnOrderConsumer`	平台退仓订单事件（FBA/速卖通/temu）
`allocation.queue.winit_self_inspection`	`WinitInspectionConsumer`	万邑通验货结果回调
`fulfillment_carton_transported`	(生产者)	箱子运输状态通知

典型 Topic 定义（含 DLT 死信队列）：

package_num_calculate       → 业务 Topic
package_num_calculate.DLT   → 死信 Topic（消费失败的消息）

fulfillment_carton_transported       → 业务 Topic
fulfillment_carton_transported.DLT   → 死信 Topic

CalculateStockPackageNumConsumer 的 "延迟+轮询" 模式值得注意：

收到消息先 sleep 2 秒
循环检查装箱信息是否就绪（最多 10 次，每次间隔 1 秒）
超时未就绪则抛异常

本质上是 "等别人发完消息" 的 hack 手段，有竞态风险，后续建议改为事件通知模式。

3.4 creeks-order-center — 订单中心

路径：E:\code\creeks-order-center\server

Topic	角色	作用
`topic.order_center_collect`	消费者	平台订单采集 → 转换 OrderViewDoc → 存 ES
`order.statics_order_topic`	生产者	Amazon 订单额外发一条消息给统计服务

消费链路：

平台订单数据 → topic.order_center_collect
  └─ OrderConsumer.consume()
      ├─ 转换为 OrderViewDoc
      ├─ Amazon 订单 → MqSender.sendOrderCenterMessage() → static_order_topic
      └─ 全部订单 → orderViewRepository.saveAll() → ES

3.5 creekshub-sync-amazon — Amazon 数据同步服务

路径：E:\code\creekshub-sync-amazon\server

纯生产者角色，发送三类消息：

Topic 配置属性	发送方法	作用
`sync.order.order_topic`	`sendOrderCenterMessage()`	发送平台订单数据
`sync.data.sp_topic`	`sendDataSyncTopic()`	数据同步事件
`sync.finance.finance_group_topic`	`notifyFinanceChange()`	Amazon 财务变更通知

3.6 codo-error-collect — 错误日志收集

路径：E:\code\codo-error-collect\server

graph LR
    A[各业务服务] -->|pushV3()| B[Kafka Topic<br/>codo_local_log_v3]
    B -->|LogQueueServiceImpl.consumeV3()| C[写入 ES 索引]
    C --> D[触发告警通知]

版本演进：

版本	Topic	传输对象	状态
V1	`codo_local_log`	`LogRecord`	已废弃
V2	`codo_local_log_v2`	`LogRecordWrapper`	已废弃
V3	`codo_local_log_v3`	原始 JSON 字符串	当前使用

V3 直接写入原始字符串，省掉一次序列化开销，很实在的优化。

消费模式：

批量消费（localLogContainerFactory，concurrency=1）
从 latest 开始消费，手动 ACK
出错后 nack(0, 2500) 重新消费（2.5 秒后重试）

3.7 creeks-maxwell — MySQL Binlog 监听

路径：E:\code\creeks-maxwell

Maxwell 是 MySQL binlog 监听工具，把数据库变更以 JSON 格式写到 Kafka。

# maxwell-kafka-config.properties
kafka.retries=3
kafka.acks=all
kafka.retry.backoff.ms=1000

配置极简，是 creeks-cdc 的 数据源头。

四、完整 Topic 清单

CDC 数据类

Topic	分区/副本策略	生产者	消费者
`cdc_api_mall_goods`	可配置	Maxwell	GoodsSyncConsumer
`cdc_api_pms_sku`	可配置	Maxwell	SkuSyncConsumer
`cdc_api_pms_spu`	可配置	Maxwell	(待实现)
`cdc_api_pms_image`	可配置	Maxwell	(待实现)
`cdc_inv_inv_stock`	可配置	Maxwell	StockSnapshotConsumer
`cdc_inv_inv_winit_stock`	可配置	Maxwell	(待实现)
`cdc_inv_inv_winit_declare_stock`	可配置	Maxwell	(待实现)
`ebay_item`	可配置	GoodsSyncConsumer	GoodsSyncConsumer
`other_item`	可配置	GoodsSyncConsumer	GoodsSyncConsumer
`ebay_item.DLT`	可配置	-	-
`other_item.DLT`	可配置	-	-
`stock-snapshot` (配置属性)	可配置	Maxwell	StockSnapshotConsumer
`base-log` (配置属性)	可配置	Maxwell	BaseLogConsumer
`sku_update_sale_status`	-	SkuSyncConsumer	(下游服务)

业务事件类

Topic	分区/副本策略	生产者	消费者
`package_num_calculate`	可配置	AllocationBusiness	CalculateStockPackageNumConsumer
`package_num_calculate.DLT`	可配置	-	-
`fulfillment_carton_transported`	可配置	ShipmentCartonTransportResendMessageService	(下游服务)
`fulfillment_carton_transported.DLT`	可配置	-	-
`allocate_outbound_cancel`	可配置	AllocationOrderCancelServiceImpl	(下游服务)
`receiving_order_event`	-	(上游服务)	ReceivingOrderListener
`return_warehouse_order`	-	(上游服务)	PlatformReturnOrderConsumer
`winit_self_inspection`	-	(上游服务)	WinitInspectionConsumer
`topic.order_center_collect`	-	(采集服务)	OrderConsumer
`order.statics_order_topic`	-	MqSender	(统计服务)

Amazon 同步类

Topic	分区/副本策略	生产者	消费者
`sync.order.order_topic`	可配置	MessageProducer	(订单中心)
`sync.order.order_topic.DLT`	可配置	-	-
`sync.data.sp_topic`	可配置	MessageProducer	(下游服务)
`sync.data.sp_topic.DLT`	可配置	-	-
`sync.finance.finance_group_topic`	可配置	MessageProducer	(下游服务)
`sync.finance.finance_group_topic.DLT`	可配置	-	-

错误收集类

Topic	分区/副本策略	生产者	消费者
`codo_local_log`	可配置	各业务服务	LogQueueServiceImpl (已废弃)
`codo_local_log_v2`	可配置	各业务服务	LogQueueServiceImpl (已废弃)
`codo_local_log_v3`	可配置	各业务服务	LogQueueServiceImpl

五、数据流转链路图

5.1 商品数据同步链路

sequenceDiagram
    participant DB as MySQL
    participant MW as Maxwell
    participant K as Kafka
    participant CDC as creeks-cdc
    participant ES as Elasticsearch

    DB->>MW: binlog 商品表变更
    MW->>K: 写入 cdc_api_mall_goods
    K->>CDC: GoodsSyncConsumer 消费
    alt platform_name = eBay
        CDC->>K: 转发到 ebay_item
        K->>CDC: ebay-goods-sync-to-es 消费
    else platform_name != eBay
        CDC->>K: 转发到 other_item
        K->>CDC: other-goods-sync-to-es 消费
    end
    CDC->>ES: goodsBatchSyncToES()

5.2 订单同步链路

sequenceDiagram
    participant Platform as 电商平台
    participant Hub as creekshub-sync-amazon
    participant K as Kafka
    participant OC as creeks-order-center
    participant ES as Elasticsearch

    Platform->>Hub: 拉取订单数据
    Hub->>K: 发送到 topic.order_center_collect
    K->>OC: OrderConsumer 批量消费
    OC->>OC: 转换为 OrderViewDoc
    opt Amazon 订单
        OC->>K: 发送到 static_order_topic
    end
    OC->>ES: saveAll()

5.3 调拨业务事件链路

sequenceDiagram
    participant Client as 前端/API
    participant Alloc as allocation 服务
    participant K as Kafka
    participant Worker as 消费者

    Client->>Alloc: 创建调拨单
    Alloc->>K: 发送 package_num_calculate
    K->>Worker: CalculateStockPackageNumConsumer
    Worker->>Worker: sleep 2s + 轮询装箱信息
    Worker->>Alloc: calculatePackageStockQuantity()

六、设计亮点

1. 统一公共配置

creeks-common-services 封装了：

生产者配置（acks、重试、序列化）
消费者配置（ACK 模式、偏移策略）
死信队列自动恢复 — DeadLetterPublishingRecoverer 消费失败 3 次后自动发 .DLT Topic
KafkaHelper 统一发送入口，各服务不重复造轮子

2. CDC + Kafka 解耦数据同步

MySQL → Maxwell → Kafka → CDC Service → ES / ClickHouse / 下游 Topic

数据库变更不直接依赖下游系统，中间通过 Kafka 缓冲，新增下游只需要新增消费者，不需要改上游。

3. DLT 死信队列兜底

重要 Topic 都配套定义了 .DLT 死信队列：

消费失败不丢数据
失败记录后续可追溯、可重放
避免单条脏消息阻塞整个消费流程

4. 分区/副本可配置

所有 Topic 的分区数、副本数、最小 ISR 都通过 YAML 配置项控制，不同环境灵活调整。

5. 批量消费提升吞吐

CDC 类消费者普遍使用 kafkaBatchListenerContainerFactory，max.poll.records 从 100 到 5000 不等，大批量写入 ES 性能更好。

七、潜在问题与建议

7.1 `acks=all` 一刀切

KafkaProducerCommonConfig 中生产者统一设置 acks=all，对于日志采集等场景性能损耗过大。

建议： 根据业务重要性分级：

订单状态流转、库存变更 → acks=all
日志采集、监控数据 → acks=1

7.2 CalculateStockPackageNumConsumer 的轮询 hack

TimeUnit.SECONDS.sleep(2L);
int count = 1;
while (!isReceivePackage(allocationOrderNo)) {
    TimeUnit.SECONDS.sleep(1L);
    count++;
    if (count >= 10) {
        throw new AllocationBusinessException("重试次数过多");
    }
}

这种 "sleep + 轮询等别人发消息" 的方式有竞态风险。

建议： 改为纯事件驱动 — 装箱完成事件直接触发装箱数量计算，去掉 sleep 和轮询。

7.3 OrderConsumer 吞异常

} catch (Exception e) {
    log.error("消费订单失败!", e);
}
ack.acknowledge();

异常被 catch 后仍然 ack 了，相当于确认了一条失败消息，数据会丢失。

建议： catch 后不 ack，让 Kafka 重试；或者直接 throw 让死信队列接管。

7.4 Topic 命名不规范

风格	示例
下划线分隔	`cdc_api_mall_goods`, `package_num_calculate`
点分隔	`codo_local_log_v3`
带 DLT 后缀	`package_num_calculate.DLT`
无 DLT	`receiving_order_event`, `return_warehouse_order`

建议： 统一命名规范，如 creeks.{服务名}.{业务}.{版本}，DLT 统一后缀。

7.5 无全局 Topic 注册表

Topic 散落在各个服务的配置类、YAML 文件中，没有一个统一的地方查看系统中所有 Topic。

建议： 在 creeks-common-services 中建立 KafkaTopicConstants 常量类，统一管理所有 Topic 名称。

八、关键文件索引

文件	路径	作用
`KafkaProducerCommonConfig.java`	`creeks-common-services/src/.../kafka/`	统一生产者配置
`KafkaConsumerCommonConfig.java`	`creeks-common-services/src/.../kafka/`	统一消费者配置 + DLT
`KafkaHelper.java`	`creeks-common-services/src/.../kafka/`	统一发送工具类
`KafkaConfig.java` (CDC)	`creeks-cdc/server/src/.../config/`	CDC Topic 定义
`KafkaTopicConfig.java` (Allocation)	`creeks-allocation/.../config/`	调拨 Topic 定义
`KafkaConfig.java` (Error Collect)	`codo-error-collect/server/.../springconfig/`	错误日志 Topic 定义
`KafkaTopicConfig.java` (Amazon)	`creekshub-sync-amazon/server/.../config/`	Amazon Topic 定义
`maxwell-kafka-config.properties`	`creeks-maxwell/`	Maxwell Kafka 配置
`GoodsSyncConsumer.java`	`creeks-cdc/server/.../consumer/goods/`	商品同步消费
`BaseLogConsumer.java`	`creeks-cdc/server/.../consumer/log/`	基础日志消费
`StockSnapshotConsumer.java`	`creeks-cdc/server/.../consumer/stock/`	库存快照消费
`SkuSyncConsumer.java`	`creeks-cdc/server/.../consumer/products/`	SKU 同步消费
`CalculateStockPackageNumConsumer.java`	`creeks-allocation/.../consumer/`	装箱计算消费
`ReceivingOrderListener.java`	`creeks-allocation/.../consumer/`	收货单事件消费
`PlatformReturnOrderConsumer.java`	`creeks-allocation/.../consumer/`	退仓单事件消费
`WinitInspectionConsumer.java`	`creeks-allocation/.../consumer/`	验货结果消费
`OrderConsumer.java`	`creeks-order-center/server/.../consumer/`	订单采集消费
`LogQueueServiceImpl.java`	`codo-error-collect/server/.../service/log/impl/`	错误日志收发
`MessageProducer.java`	`creekshub-sync-amazon/server/.../mq/`	Amazon 消息发送

Creeks 系统 Kafka 使用全景分析

一、整体架构

二、Kafka 三大核心用途

三、各服务 Kafka 使用详解

3.1 creeks-common-services — 公共基础设施

3.1.1 生产者配置 `KafkaProducerCommonConfig`

3.1.2 消费者配置 `KafkaConsumerCommonConfig`

3.1.3 KafkaHelper 发送工具

3.2 creeks-cdc — CDC 数据变更捕获服务（核心大动脉）

消费链路表

重点消费逻辑拆解

3.3 creeks-allocation — 调拨服务

3.4 creeks-order-center — 订单中心

3.5 creekshub-sync-amazon — Amazon 数据同步服务

3.6 codo-error-collect — 错误日志收集

3.7 creeks-maxwell — MySQL Binlog 监听

四、完整 Topic 清单

CDC 数据类

业务事件类

Amazon 同步类

错误收集类

五、数据流转链路图

5.1 商品数据同步链路

5.2 订单同步链路

5.3 调拨业务事件链路

六、设计亮点

1. 统一公共配置

2. CDC + Kafka 解耦数据同步

3. DLT 死信队列兜底

4. 分区/副本可配置

5. 批量消费提升吞吐

七、潜在问题与建议

7.1 `acks=all` 一刀切

7.2 CalculateStockPackageNumConsumer 的轮询 hack

7.3 OrderConsumer 吞异常

7.4 Topic 命名不规范

7.5 无全局 Topic 注册表

八、关键文件索引

Kafka 生产消费工具封装与死信队列处理

评论 0

一、整体架构

二、Kafka 三大核心用途

三、各服务 Kafka 使用详解

3.1 creeks-common-services — 公共基础设施

3.1.1 生产者配置 KafkaProducerCommonConfig

3.1.2 消费者配置 KafkaConsumerCommonConfig

3.1.3 KafkaHelper 发送工具

3.2 creeks-cdc — CDC 数据变更捕获服务（核心大动脉）

消费链路表

重点消费逻辑拆解

3.3 creeks-allocation — 调拨服务

3.4 creeks-order-center — 订单中心

3.5 creekshub-sync-amazon — Amazon 数据同步服务

3.6 codo-error-collect — 错误日志收集

3.7 creeks-maxwell — MySQL Binlog 监听

四、完整 Topic 清单

CDC 数据类

业务事件类

Amazon 同步类

错误收集类

五、数据流转链路图

5.1 商品数据同步链路

5.2 订单同步链路

5.3 调拨业务事件链路

六、设计亮点

1. 统一公共配置

2. CDC + Kafka 解耦数据同步

3. DLT 死信队列兜底

4. 分区/副本可配置

5. 批量消费提升吞吐

七、潜在问题与建议

7.1 acks=all 一刀切

7.2 CalculateStockPackageNumConsumer 的轮询 hack

7.3 OrderConsumer 吞异常

7.4 Topic 命名不规范

7.5 无全局 Topic 注册表

八、关键文件索引

Kafka 生产消费工具封装与死信队列处理

评论 0

3.1.1 生产者配置 `KafkaProducerCommonConfig`

3.1.2 消费者配置 `KafkaConsumerCommonConfig`

7.1 `acks=all` 一刀切