多渠道 SKU 事件 Kafka 消费方案

结论

方案结论：PIM 与 Backend 共用一条 SKU 事件消费路径：按 store_code 取渠道配置，匹配商品一次，再保存两张表。无内部 Kafka 通知。京东首批接入；其他渠道按统一契约映射后复用。

channelCode 从 auth_config 获取；消费者不写渠道名分支。
各渠道外部 SKU / item ID 映射到 channel_sku_id / 可空 page_id。
page_id 为空时，两表 item_id 都使用 channel_sku_id。
当前仍为设计稿；每个渠道上线前需协调其旧写入任务，首批京东需关闭 ODTS JD Job。

适用于多渠道共用的 SKU 事件消费京东首批接入尚未实施

事件如何从 PIM 到达数据库？

PIM 发布事件。Backend Listener 解析后调用 Service。Service 按 store_code 查询授权配置，处理可空 page_id 并匹配一次商品，然后直接保存两张表和清理缓存；不发布内部 Kafka 消息。

PIM → Backend → 共享数据库

图的结构

PIM pim
Kafka topic kafka
Backend Consumer backend
odts 共享库 database
pim → kafka · 发布
kafka → backend · 消费
backend → database · 查配置 / 匹配 / 保存两表

消息携带什么？

Topic 为 offer-pim-channel-sku-basic-publish。分区键 store_code-channel_sku_id 由 PIM 控制；消费者不生成消息 Key。

字段	用途	落库或处理
`schema_version`	消息字段	不落库，不做版本校验，也不作为拒收消息的条件
`event_id`	日志关联	不做幂等去重；重复消息按最后写入覆盖
`event_type`	事件路由	`UPSERT`、`DELETED`、`ON_SHELF`、`OFF_SHELF`
`event_time`	上游事件时间	转成 `Instant` 写入 `modified`
`store_code`	店铺识别码	查询 `auth_config`，取得 `storeId`、`channelCode`、`authCode`
`page_id`	渠道商品 / item 外部 ID，可空	两表 `item_id`：优先使用 `page_id`，为空时使用 `channel_sku_id`
`channel_sku_id`	当前渠道 SKU 外部 ID	`channel_sku.sku_id`
`internal_code`	外部商品编码	`outer_id`；同时作为中台匹配输入
`channel_sku_name`	SKU 名称	`channel_sku.sku_name`；暂用于 item 名称
`price`	SKU 价格字符串	转成 `BigDecimal` 写 `product_price`
`changed_fields`	上游变更字段	指示增量更新字段，不单独落库

四种事件分别改变什么？

`event_type`	写库动作	后续动作
`UPSERT`	按 item / SKU 查询；存在则更新，不存在则新增。字段更新按 `changed_fields`	保存两表并清理缓存
`DELETED`	将 SKU 软删除；按业务规则同步父 item	保存两表并清理缓存
`ON_SHELF`	`is_onsale = 1`；同步父 item 状态	保存两表并清理缓存
`OFF_SHELF`	`is_onsale = 0`；不软删除	保存两表并清理缓存

所有事件均直接写数据库；不构造 before/after，也不发送内部 Kafka 消息。

price 更新为 product_price；internal_code 更新 outer_id，并触发一次商品匹配；同一匹配结果回填 item 与 SKU。channel_sku_name 更新 sku_name。

`channel_item`：42 列映射

Backend 直接写 odts.channel_item。下表逐列列出本事件链路写入方式；“不触碰”表示 JPA 不映射或本消息不提供该字段。

DDL 列	类型	Backend 映射 / 赋值	说明
`id`	bigint	新增 `RedisHelper.getEntityId()`；更新保留原 ID	主键 NOT NULL
`auth_code`	varchar	`auth_config.auth_code`	按 `store_code` 查询
`item_id`	varchar	解析后的 `itemId`	`page_id` 非空用其值；否则用 `channel_sku_id`
`item_name`	varchar	`msg.channel_sku_name`	PIM 未提供独立 item 名称
`outer_id`	varchar	`msg.internal_code`	外部商品编码
`product_code`	varchar	匹配商品后写 `ProductEntity.code`	中台匹配结果
`product_cname`	varchar	匹配商品后写 `ProductInfoEntity.chineseName`	中台匹配结果
`product_price`	decimal	不设置	PIM 无 item 级价格
`mp_id`	bigint	匹配商品后写 `ProductEntity.merchantProductId`	中台匹配结果
`store_mp_id`	bigint	匹配商品后写 `ProductEntity.id`	中台匹配结果
`modified`	timestamp	`msg.event_time`	毫秒转时间
`last_query_time`	timestamp	不设置	ODTS JD 同步字段
`ext_info`	longtext	不设置	写入 NULL / 保持默认
`stock_auto_push_status`	tinyint	不设置	ODTS 推送字段
`stock_push_error_remark`	varchar	不设置	ODTS 推送字段
`channel_code`	varchar	`auth_config.channel_code`	同一次授权查询
`merchant_id`	bigint	`commonProp.getWarehouseMerchantId()`	—
`store_id`	bigint	`auth_config.store_id`	同一次授权查询
`match_status`	tinyint	单 SKU 匹配后设 `3`；未匹配设 `1`	ALL_MATCH / UN_MATCH
`price_auto_push_status`	tinyint	不设置	ODTS 推送字段
`price_push_error_remark`	varchar	不设置	ODTS 推送字段
`is_available`	tinyint	DTO `of()` 显式设 `1`	避免 JPA NULL 覆盖 DDL 默认值
`is_deleted`	bigint	`DELETED` 时设 ID；其他事件设 `0`	软删除标记
`version_no`	int	新增设 `0`；更新用旧值加一，旧值 NULL 则 `0`	手动赋值，不用 `@Version`
`create_userid`	bigint	不设置	ODTS 审计字段
`create_username`	varchar	不设置	ODTS 审计字段
`create_userip`	varchar	不设置	ODTS 审计字段
`create_usermac`	varchar	不设置	ODTS 审计字段
`create_time`	timestamp	`@CreatedDate`	Backend 审计时间
`create_time_db`	timestamp	数据库默认值	不触碰
`server_ip`	varchar	不设置	ODTS 系统字段
`update_userid`	bigint	不设置	ODTS 审计字段
`update_username`	varchar	不设置	ODTS 审计字段
`update_userip`	varchar	不设置	ODTS 审计字段
`update_usermac`	varchar	不设置	ODTS 审计字段
`update_time`	timestamp	`@LastModifiedDate`	Backend 审计时间
`update_time_db`	timestamp	数据库自动更新	不触碰
`client_versionno`	varchar	不设置	ODTS 系统字段
`company_id`	bigint	`BizConstant.COMPANY_ID`	Backend 常量
`is_onsale`	int	按 `event_type` 映射	上架 `1`；下架 / 删除 `0`；UPSERT 不改变
`type_of_product`	tinyint	匹配商品后写 `ProductEntity.typeOfProduct`	中台匹配结果

`channel_sku`：38 列映射

Backend 直接写 odts.channel_sku。新增 SKU 的 channel_item_id 使用保存 item 后返回的 ID；其余字段逐列映射如下。

DDL 列	类型	Backend 映射 / 赋值	说明
`id`	bigint	新增 `RedisHelper.getEntityId()`；更新保留原 ID	主键 NOT NULL
`auth_code`	varchar	`auth_config.auth_code`	按 `store_code` 查询
`channel_item_id`	bigint	保存 `channel_item` 后返回的主键	父商品主键
`item_id`	varchar	解析后的 `itemId`	`page_id` 非空用其值；否则用 `channel_sku_id`
`sku_id`	varchar	`msg.channel_sku_id`	当前渠道 SKU 外部 ID
`sku_name`	varchar	`msg.channel_sku_name`	消息字段已补充
`outer_id`	varchar	`msg.internal_code`	外部商品编码
`product_code`	varchar	匹配商品后写 `ProductEntity.code`	中台匹配结果
`product_cname`	varchar	匹配商品后写 `ProductInfoEntity.chineseName`	中台匹配结果
`product_price`	decimal	`new BigDecimal(msg.price)`	SKU 价格
`mp_id`	bigint	匹配商品后写 `ProductEntity.merchantProductId`	中台匹配结果
`store_mp_id`	bigint	匹配商品后写 `ProductEntity.id`	中台匹配结果
`modified`	timestamp	`msg.event_time`	毫秒转时间
`ext_info`	text	不设置	PIM 消息不提供原始 SKU JSON
`properties_name`	varchar	不设置	PIM 消息无规格属性
`channel_code`	varchar	`auth_config.channel_code`	同一次授权查询
`merchant_id`	bigint	`commonProp.getWarehouseMerchantId()`	—
`store_id`	bigint	`auth_config.store_id`	同一次授权查询
`is_available`	tinyint	DTO `of()` 显式设 `1`	避免 JPA NULL 覆盖 DDL 默认值
`is_deleted`	bigint	`DELETED` 时设 ID；其他事件设 `0`	软删除标记
`version_no`	int	新增设 `0`；更新用旧值加一，旧值 NULL 则 `0`	手动赋值，不用 `@Version`
`create_userid`	bigint	不设置	ODTS 审计字段
`create_username`	varchar	不设置	ODTS 审计字段
`create_userip`	varchar	不设置	ODTS 审计字段
`create_usermac`	varchar	不设置	ODTS 审计字段
`create_time`	timestamp	`@CreatedDate`	Backend 审计时间
`create_time_db`	timestamp	数据库默认值	不触碰
`server_ip`	varchar	不设置	ODTS 系统字段
`update_userid`	bigint	不设置	ODTS 审计字段
`update_username`	varchar	不设置	ODTS 审计字段
`update_userip`	varchar	不设置	ODTS 审计字段
`update_usermac`	varchar	不设置	ODTS 审计字段
`update_time`	timestamp	`@LastModifiedDate`	Backend 审计时间
`update_time_db`	timestamp	数据库自动更新	不触碰
`client_versionno`	varchar	不设置	ODTS 系统字段
`company_id`	bigint	`BizConstant.COMPANY_ID`	Backend 常量
`is_onsale`	int	按 `event_type` 映射	上架 `1`；下架 / 删除 `0`；UPSERT 不改变
`type_of_product`	tinyint	匹配商品后写 `ProductEntity.typeOfProduct`	中台匹配结果

新增记录的字段如何保证完整？

Backend 写共享库时不能依赖数据库为主键生成值：两个 Entity 的 id 都是无 @GeneratedValue 的 @Id。设计要求在新增时调用 Redis ID helper；更新时沿用已查出的主键。

字段 / 依赖	设计处理	依据与边界
`id`	新增 item / SKU 调用 `RedisHelper.getEntityId()`；更新保留原 ID	Backend 已有 Redis 自增 helper；年维度计数，返回 14 位 Long
`auth_code`、`store_id`、`channel_code`	增加 `AuthConfigEntity` / Repository；按 `store_code` 查询 `odts.auth_config`	查询结果供 item、SKU 共用；字段属于唯一键组成部分
`is_available`	DTO `of()` 显式设为 `1`	Entity 映射该列时，JPA 的显式 NULL 会覆盖 DDL 默认值
`version_no`	新增 Entity 字段 `Integer versionNo`；新增赋 `0`，更新赋旧值加一，旧值为空则赋 `0`	Backend 没有此表字段的自动递增机制；参照 ODTS `BasePO.beforeInsert/beforeUpdate`，不使用 `@Version`
`company_id`	使用 `BizConstant.COMPANY_ID`	沿用当前 Backend 方案
中台匹配	用 `internal_code` 和 `storeId` 查询商品数据，回填商品编码、名称、类型及关联 ID	文档 §3.1.9 提出 Backend 查库替代 ODTS SOA

更新前先按 item / SKU 业务键查询，再按 ID 保存。Item 查询以 itemId + storeId 为条件；SKU 查询使用父 item ID。软删除记录查询策略需要保留，以支持既有记录更新。

Listener 与业务 Service 怎样分工？

组件	职责
`ChannelSkuEventMessage`	映射 JSON 字段，包括 snake_case 字段名
`ChannelSkuEventListener`	绑定新 topic、反序列化消息、调用 Service；沿用 `aliKafkaListener` 与现有错误处理器
`ChannelSkuEventServiceImpl`	校验必需字段、解析 `itemId`、按 `store_code` 查询授权、匹配一次商品、保存两表、清理缓存
`AuthConfigEntity` / Repository	映射 `odts.auth_config`；只按 `store_code` 查询一次，两表共用结果
`MiddlewareMatchService`	每条事件按 `internal_code + storeId` 匹配一次；item 与 SKU 共用同一结果

上线前必须处理哪些风险？

风险	影响	应对
旧写入任务与 Backend 并行	共享数据被覆盖或不一致	每个渠道启用前，确认并停止该渠道冲突写入任务；首批京东先关闭 ODTS JD Job
Kafka topic 尚未创建	Consumer 无法接收事件	上线前创建共用 topic
PIM 消息契约未锁定	渠道字段可能映射错误	PIM 确认 `internal_code`、价格精度、`changed_fields` 和 `store_code` 字典
消息乱序	SKU 状态可能被旧事件覆盖	同一店铺 SKU 使用同分区键，并按 `event_time` 处理
缓存未同步失效	查询可能读到旧数据	更新后清理 Backend `SkuIdCache`；逐渠道确认是否共享旧应用缓存
业务异常	部分事件不能落库	Listener 复用错误处理器重试；未知事件或店铺未匹配时记录错误并返回

首批京东：确认消息契约与 topic；关闭 ODTS JD Job；启用 Backend 共用 Consumer；先完成单店铺灰度。后续渠道按相同契约完成字段映射和旧任务切换后复用该路径。

多渠道接入边界

共用消费者只处理标准事件和授权配置，不按渠道名分支。PIM 将各渠道 SKU / item ID、商品编码、名称、价格和事件类型映射到统一契约。字段语义不同则先归一化或扩展契约。

每个渠道配置 store_code / channelCode，并协调旧写入任务。每条事件只匹配一次商品，item 与 SKU 共用结果。本方案不构造 ChannelSkuOnsaleDto，也不发内部 Kafka 消息。

交付范围是什么？

本次新增消息 DTO、事件枚举、Listener 和 Service；测试 Controller 可选。配置文件新增 Consumer group。方案不新增表、不发送内部 Kafka 消息，也不修改 ODTS 代码。

当前仍是设计稿，未落代码。上线前需完成：

创建共用 Kafka topic 并确认标准消息契约。
关闭京东旧 Job，完成 Backend 编码、测试和首批灰度。
后续渠道逐一确认字段映射、auth_config 配置及旧任务切换。