Pulsar 消息同步机制

Pulsar 就是一个消息中间件消息队列和 RabbitMQ、Kafka 是同一类东西。它解决的核心问题就一个让两个服务之间不用直接互相调用而是通过一个信箱传话。放在你的项目里场景特别具体谁发给谁雇员服务在导入员工数据后需要通知薪资服务这个员工的信息变了你同步一下。但雇员服务不想、也不能直接去调薪资服务的接口服务解耦、避免互相依赖、避免一方挂了拖垮另一方。于是雇员服务把变更的员工数据投进信箱薪资服务自己去取信处理。这个信箱就是 Pulsar三个关键概念对应你代码里的位置Topic主题/频道—— 信箱的名字。所有发给这个频道的消息订阅它的人都能收到。Producer生产者—— 寄信人。负责往 topic 里塞消息。核心就是send()方法把对象转 JSON 发出去。雇员服务那边也有对应的 producer。Consumer消费者—— 取信人。负责从 topic 里拿消息并处理。就是专门收员工消息、同步自然人数据的消费者。用你的街道数据走一遍完整流程雇员服务导入 Excel/import接口→ 写好EmployeeBasicInfo含residenceStreet/resideStreet→ 落自己的库 →通过 pulsar producer 把这个 Employee 对象发到employeetopic。薪资服务启动时 就连上 Pulsar订阅这个 topic。每当有新消息进来基类messageListener回调被触发把消息体反序列化成Employee对象校验通过就调handlerData(object)这个方法各消费者自己重写处理完consumer.acknowledge(msg)告诉 Pulsar这条我处理好了可以删了处理失败negativeAcknowledge会让 Pulsar 稍后重投。handlerData在IndividualIncomeTaxConsumer里就是那个把Employee/EmployeeBasicInfo的字段往IndividualDeclareInfo上搬的逻辑这也正好解释了街道为什么断在同步层handlerData搬地址时搬了省市区和详细地址却漏了residenceStreet/resideStreet两行——而薪资服务的 feign 副本EmployeeBasicInfo里也没声明这两个字段所以即便想搬也取不到值。这就是上次说的缺口。两个配置开关开发时很实用值得你知道offFlag设成1就完全跳过 pulsar 初始化本地没 pulsar 环境也能把服务跑起来不用连那台180.100.212.181。开发环境一般开着生产关掉。基类里还配了死信队列一条消息连续失败 3 次maxRedeliverCount(3)就不再重投扔到xxx-DLQ主题里避免一条坏消息把消费者卡死。这是消息队列的标配容错机制。一句话总结Pulsar 是雇员服务和薪资服务之间的传声筒雇员那边改了数据通过 topic 喊一声薪资服务听到后自己同步——双方互不直接调用。街道数据的断点不在 Pulsar 本身而在薪资服务收到消息后搬字段那步漏了两行。一、Pulsar 是什么Pulsar 是一个消息中间件消息队列和 RabbitMQ、Kafka 属同一类。核心作用就一个让两个服务之间不用直接互相调用而是通过一个信箱传话。本系统场景雇员服务导入或变更员工数据后要通知薪资服务同步自然人报送信息。雇员服务不直接调薪资服务的接口而是把变更数据投递到 Pulsar薪资服务自己订阅消费。好处是服务解耦——一方故障不拖垮另一方双方也无直接代码依赖。服务地址在pulsar: serviceUrl: pulsar://180.100.212.181:6650二、三个核心概念概念含义类比Topic消息频道信箱名字Producer发送消息寄信人Consumer接收处理消息取信人员工主题是persistent://public/salary/employee。Producer基类把对象转 JSON 发出。跨服务场景真正的发送入口是雇员服务它组装EmployeeSyncVO后调 SalaryProviderService.javaConsumer基类 用泛型T子类重写validate()和handlerData()。[IndividualIncomeTaxConsumer.java](D:\\EngmaProject\\salary-system\\src\\main\\java\\com\\engma\\salary\\service\\pulsar\\IndividualIncomeTaxConsumer.java) 就是同步自然人信息的消费者泛型是Employee。三、消息消费流程启动 → 连接 Pulsar → 订阅 topic → 注册 messageListener 回调 │ 每条消息到达时触发 │ ┌───────────┴────────────┐ │ 1. 反序列化为泛型对象 T │ │ 2. validate() 校验 │ │ 3. handlerData() 处理 │ │ 4. acknowledge() 确认 │ └────────────────────────┘步骤行号说明初始化连接49PulsarClient.builder().serviceUrl(serviceUrl).build()创建消费者61client.newConsumer().topic(topic).subscribe()接收回调64messageListener((consumer, msg) - {...})反序列化72JSONObject.parseObject(data, tClass)业务分发78handlerData(object)调子类实现成功确认80consumer.acknowledge(msg)告诉 Pulsar 可删失败重投75consumer.negativeAcknowledge(msg)稍后重发确认机制ack/nackack 表示处理成功、消息可删nack 表示失败、1 秒后重投死信队列第65-68行连续失败 3 次的消息不再重投扔进{topic}-DLQ避免坏消息卡死消费者。共享订阅第68行SubscriptionType.Shared同一订阅名下多消费者实例间负载均衡。四、跨服务同步链路以街道为例┌─────────────── 雇员服务 (employee) ───────────────┐ │ 1. POST /import 上传 Excel │ │ └─ EmployeePayTaxesPlusExcel 解析报税 sheet │ │ 含「户籍所在地街道」「居住地街道」两列 │ │ 2. EmployeePayTaxesPlusExcelListener.invoke() │ │ └─ buildBasicInfoForPayPlusTaxes() 搬字段 │ │ info.setResidenceStreet(...) 户籍街道 │ │ info.setResideStreet(...) 居住街道 │ │ 3. updateBatchById() 落库 employee_basic_info │ │ 4. NaturalReportHandle.sendDataToPulsar() │ │ └─ SalaryProviderService.send() 发到 employee topic │ └───────────────────────┬───────────────────────────┘ │ Pulsar 消息JSON ▼ ┌─────────────── 薪资服务 (salary-system) ──────────┐ │ 5. IndividualIncomeTaxConsumer 收到消息 │ │ └─ 反序列化为 Employee → handlerData() │ │ └─ 搬字段到 IndividualDeclareInfo │ │ 216-227行搬地址漏了街道 │ │ 6. 落库 individual_declare_info │ │ 7. /exportList → getDatas() 按 code 取值 │ │ └─ residenceStreet / streetOfResidence │ └───────────────────────────────────────────────────┘各环节代码位置特殊处理省/市/区在导入时经isAddress()[EmployeePayTaxesPlusExcelListener.java:441](D:\\EngmaProject\\employee\\src\\main\\java\\com\\engma\\employee\\excel\\EmployeePayTaxesPlusExcelListener.java)做级联校验、取标准码值街道不校验直接原值透传因为乡镇街道这级没有统一编码标准。五、街道数据断链问题现象导出 Excel 里两列街道始终为空。根因链路在薪资服务搬字段那步断了两个原因feign 副本 [EmployeeBasicInfo.java](D:\\EngmaProject\\salary-system\\src\\main\\java\\com\\engma\\salary\\entity\\feign\\EmployeeBasicInfo.java) 没声明residenceStreet/resideStreet反序列化取不到值消费者 [IndividualIncomeTaxConsumer.java:216-227](D:\\EngmaProject\\salary-system\\src\\main\\java\\com\\engma\\salary\\service\\pulsar\\IndividualIncomeTaxConsumer.java) 搬地址时漏了街道两行修复改薪资服务两处feign 副本补两个字段消费者第227行后补两行info.setResidenceStreet(basicInfo.getResidenceStreet())和info.setStreetOfResidence(basicInfo.getResideStreet())。注意字段名差异雇员侧resideStreet对应薪资侧streetOfResidence。六、配置开关offFlag[salary-dev.yaml:65](D:\\EngmaProject\\salary-system\\src\\main\\resources\\salary-dev.yaml)1关闭 pulsar 初始化本地无 pulsar 也能启动0开启。开发一般设 1生产必须设 0。mockPaidSwitch第67行1注入 mock 数据0关闭。七、Topic 命名坑重要两端 topic 命名不一致薪资服务订阅persistent://public/salary/employee无后缀雇员服务发送persistent://public/salary/employee-{active}active默认dev默认配置下雇员发到...-dev薪资服务订阅的是无后缀版本收不到。排查同步问题优先确认两端 topic 是否匹配。八、组件一览消费者IndividualIncomeTaxConsumer自然人同步、LaborContractHandleConsumer劳动合同、IndividualContractRenewConsumer合同续签、SalaryBatchPayInfoConsumer薪资发放、AttendanceHandleConsumer考勤。生产者PulsarProducerService基类、AttendanceUsedStatusProducer、SalaryBatchCalculatedProducer、RecruitSalaryChannelProducer。九、排查指引消费失败日志搜[PULSAR]message error连续失败 3 次进死信队列{topic}-DLQ确认两端 topic 一致。本地开发offFlag1跳过初始化即可启动。同步验证导入带街道的数据 → 看日志有无PULSAR---msg---和确认收到消息→ 查表residence_street有无值 → 调导出接口验证 Excel。