任务提交 分发记录入库和ES

common/fire-dto/src/main/java/com/fire/dto/FlowOrderInfo.java

@@ -2,12 +2,14 @@ package com.fire.dto;
 
 import java.util.Date;
 
+import com.baomidou.mybatisplus.annotation.TableName;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
 @ApiModel(value = "订单记录表实体")
 @Data
+@TableName("flow_order_info")
 public class FlowOrderInfo {
 
     @ApiModelProperty(value = "接入ID号")

common/fire-dto/src/main/java/com/fire/dto/MobileFlowDispatchRec.java

@@ -2,14 +2,20 @@ package com.fire.dto;
 
 import java.util.Date;
 
+import com.baomidou.mybatisplus.annotation.TableName;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 @ApiModel(value = "分发记录实体")
 @Data
+@TableName("mobile_flow_dispatch_rec")
 @Builder
+@NoArgsConstructor
+@AllArgsConstructor
 public class MobileFlowDispatchRec {
 
     @ApiModelProperty(value = "分发ID号")
@@ -87,5 +93,4 @@ public class MobileFlowDispatchRec {
     @ApiModelProperty(value = "运营商：1、移动 2、电信 3、联通")
     private Integer phoneOperator;
 
-
 }

modules/save-data/src/main/java/com/fire/savedata/consumer/DispatchTidbConsumer.java

@@ -0,0 +1,50 @@
+package com.fire.savedata.consumer;
+
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fire.common.constants.RocketTopic;
+import com.fire.dto.FlowOrderInfo;
+import com.fire.dto.MobileFlowDispatchRec;
+import com.fire.savedata.mapper.FlowOrderInfoMapper;
+import com.fire.savedata.mapper.MobileFlowDispatchRecMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+import static com.fire.common.constants.RocketTags.MAKE_TAG;
+
+@Slf4j
+@Service
+@RocketMQMessageListener(consumerGroup = "DispatchTidb",topic = RocketTopic.CHILD_ORDER_TOPIC,consumeMode = ConsumeMode.ORDERLY)
+public class DispatchTidbConsumer implements RocketMQListener<MessageExt> {
+
+    @Resource
+    private MobileFlowDispatchRecMapper mobileFlowDispatchRecMapper;
+
+    @Override
+    public void onMessage(MessageExt msg) {
+        String s = new String(msg.getBody());
+        log.info("DispatchTidb message: " + s);
+        String tag = msg.getTags();
+        ObjectMapper om = new ObjectMapper();
+        try {
+            MobileFlowDispatchRec mobileFlowDispatchRec = om.readValue(s, MobileFlowDispatchRec.class);
+            if (tag.equals(MAKE_TAG)){
+                mobileFlowDispatchRecMapper.insert(mobileFlowDispatchRec);
+            }else{
+                mobileFlowDispatchRecMapper.updateById(mobileFlowDispatchRec);
+            }
+        } catch (JsonProcessingException e) {
+            e.printStackTrace();
+            log.error(e.toString());
+            //throw new BaseException("Tidb消费出错");
+        }
+    }
+
+}

modules/save-data/src/main/java/com/fire/savedata/mapper/MobileFlowDispatchRecMapper.java

@@ -0,0 +1,11 @@
+package com.fire.savedata.mapper;
+
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.fire.dto.MobileFlowDispatchRec;
+
+
+public interface MobileFlowDispatchRecMapper extends BaseMapper<MobileFlowDispatchRec> {
+
+
+}