feat(综合搜索)：增量更新

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-api/src/main/java/com/yeejoin/amos/boot/module/common/api/entity/ESEquipmentInfo.java

@@ -342,6 +342,12 @@ public class ESEquipmentInfo {
     private LocalDate INSPECT_DATE;
 
     /**
+     * 单位类型-多个逗号分隔开-实时的单位类型
+     */
+    @Field(type = FieldType.Keyword)
+    private String unitType;
+
+    /**
      * 最新一条检验信息
      */
     @Field(type = FieldType.Nested)

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-api/src/main/java/com/yeejoin/amos/boot/module/common/api/service/IDataRefreshDispatch.java

@@ -2,7 +2,9 @@ package com.yeejoin.amos.boot.module.common.api.service;
 
 import com.yeejoin.amos.boot.module.common.api.entity.TzsDataRefreshMessage;
 
+import java.util.List;
+
 public interface IDataRefreshDispatch {
 
-    void doDispatch(TzsDataRefreshMessage message);
+    void doDispatch(String dataType, List<TzsDataRefreshMessage> messages);
 }

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-biz/src/main/java/com/yeejoin/amos/boot/module/common/biz/refresh/DataRefreshEvent.java

@@ -3,13 +3,15 @@ package com.yeejoin.amos.boot.module.common.biz.refresh;
 import lombok.Getter;
 import org.springframework.context.ApplicationEvent;
 
+import java.util.List;
+
 @Getter
 public class DataRefreshEvent extends ApplicationEvent {
 
     /**
      * 业务数据ID：设备record、企业的id、人员的id
      */
-    private final String dataId;
+    private final List<String> dataIds;
 
     /**
      * 数据类型
@@ -27,9 +29,9 @@ public class DataRefreshEvent extends ApplicationEvent {
      * @param source the object on which the event initially occurred or with
      *               which the event is associated (never {@code null})
      */
-    public DataRefreshEvent(Object source, String dataId, String dataType, Operation operation) {
+    public DataRefreshEvent(Object source, List<String> dataIds, String dataType, Operation operation) {
         super(source);
-        this.dataId = dataId;
+        this.dataIds = dataIds;
         this.dataType = dataType;
         this.operation = operation;
     }

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-biz/src/main/java/com/yeejoin/amos/boot/module/common/biz/refresh/dispatch/DataRefreshDispatcher.java

@@ -13,6 +13,7 @@ import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
 import java.util.Date;
+import java.util.List;
 
 @Component
 @RequiredArgsConstructor
@@ -24,15 +25,17 @@ public class DataRefreshDispatcher implements IDataRefreshDispatch {
 
     @Override
     @Async
-    public void doDispatch(TzsDataRefreshMessage refreshMessage) {
-        try {
-            IDataRefreshHandler dataRefreshHandler = refreshHandlerFactory.getRefreshHandler(refreshMessage.getDataType());
-            dataRefreshHandler.doRefresh(refreshMessage);
-            markRefreshSuccess(refreshMessage);
-        } catch (Exception e) {
-            log.error("三库数据刷新执行失败，消息内容：{}", JSONObject.toJSONString(refreshMessage), e);
-            markRefreshFailure(refreshMessage);
-        }
+    public void doDispatch(String dataType, List<TzsDataRefreshMessage> messages) {
+        IDataRefreshHandler dataRefreshHandler = refreshHandlerFactory.getRefreshHandler(dataType);
+        messages.forEach(message -> {
+            try {
+                dataRefreshHandler.doRefresh(message);
+                markRefreshSuccess(message);
+            } catch (Exception e) {
+                log.error("三库数据刷新执行失败，消息内容：{}", JSONObject.toJSONString(messages), e);
+                markRefreshFailure(message);
+            }
+        });
     }
 
     private void markRefreshSuccess(TzsDataRefreshMessage message) {

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-biz/src/main/java/com/yeejoin/amos/boot/module/common/biz/refresh/listener/DataRefreshListener.java

@@ -15,11 +15,13 @@ import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionalEventListener;
 
 import javax.annotation.PostConstruct;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 @Component
@@ -65,24 +67,22 @@ public class DataRefreshListener {
 
 
     private void processEvent(DataRefreshEvent event) {
-        TzsDataRefreshMessage message = new TzsDataRefreshMessage();
-        try {
-            // 1.记录 message
-            createMsg(event, message);
-            // 2.调用更新处理
-            dataRefreshService.ifPresent(service -> service.doDispatch(message));
-        } catch (Exception e) {
-            message.setStatus(Constants.REFRESH_STATUS_FAILURE); // 标记为失败
-            message.setErrorMsg(e.getMessage());
-            tzsDataRefreshMessageService.saveOrUpdate(message);
-        }
+        // 1.记录 message
+        List<TzsDataRefreshMessage> messages = createMsg(event);
+        // 2.调用更新处理
+        dataRefreshService.ifPresent(service -> service.doDispatch(event.getDataType(), messages));
     }
 
-    private void createMsg(DataRefreshEvent event, TzsDataRefreshMessage message) {
-        message.setDataId(event.getDataId());
-        message.setDataType(event.getDataType());
-        message.setOperation(event.getOperation().name());
-        message.setStatus(Constants.REFRESH_STATUS_DEALING); // 流程中
-        tzsDataRefreshMessageService.save(message);
+    private List<TzsDataRefreshMessage> createMsg(DataRefreshEvent event) {
+        List<TzsDataRefreshMessage> messages = event.getDataIds().stream().map(dataId -> {
+            TzsDataRefreshMessage message = new TzsDataRefreshMessage();
+            message.setDataId(dataId);
+            message.setDataType(event.getDataType());
+            message.setOperation(event.getOperation().name());
+            message.setStatus(Constants.REFRESH_STATUS_DEALING); // 流程中
+            return message;
+        }).collect(Collectors.toList());
+        tzsDataRefreshMessageService.saveBatch(messages);
+        return messages;
     }
 }