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.java
Consumer基类 用泛型<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() 确认 │ └────────────────────────┘| 步骤 | 行号 | 说明 |
|---|---|---|
| 初始化连接 | 49 | PulsarClient.builder().serviceUrl(serviceUrl).build() |
| 创建消费者 | 61 | client.newConsumer().topic(topic).subscribe() |
| 接收回调 | 64 | messageListener((consumer, msg) -> {...}) |
| 反序列化 | 72 | JSONObject.parseObject(data, tClass) |
| 业务分发 | 78 | handlerData(object)调子类实现 |
| 成功确认 | 80 | consumer.acknowledge(msg)告诉 Pulsar 可删 |
| 失败重投 | 75 | consumer.negativeAcknowledge(msg)稍后重发 |
确认机制（ack/nack）：ack 表示处理成功、消息可删；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 一致。
本地开发：offFlag=1跳过初始化即可启动。
同步验证：导入带街道的数据 → 看日志有无PULSAR---msg---和确认收到消息→ 查表residence_street有无值 → 调导出接口验证 Excel。