|
|
@@ -1,5 +1,7 @@
|
|
|
package org.dromara.server.mq.consumer;
|
|
|
|
|
|
+import cn.hutool.core.util.ObjectUtil;
|
|
|
+import cn.hutool.json.JSONObject;
|
|
|
import cn.hutool.json.JSONUtil;
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
@@ -40,7 +42,11 @@ public class KafkaLocalConsumer {
|
|
|
public void cloudOperationSync(ConsumerRecord<String, String> record){
|
|
|
KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
|
|
|
try{
|
|
|
- if(receiveMsg.getHeader().getTenantId().equals(this.tenantId)){
|
|
|
+ KafkaHeader header = receiveMsg.getHeader();
|
|
|
+ JSONObject bodyObj = JSONUtil.parseObj(receiveMsg.getBody());
|
|
|
+ String tenantId = ObjectUtil.isNotEmpty(header.getTenantId()) ? header.getTenantId():
|
|
|
+ bodyObj.getOrDefault("tenantId", "0").toString();
|
|
|
+ if(tenantId.equals(this.tenantId)){
|
|
|
String eventType = receiveMsg.getHeader().getEventType();
|
|
|
String sender = receiveMsg.getHeader().getSender();
|
|
|
// yktEventStrategyContext.doMsgHandle(sender, eventType, receiveMsg.getBody());
|