实现监利iot数据采集

custom-gateway-app/src/main/resources/application-jldma.yml

@@ -0,0 +1,29 @@
+#spring
+spring:
+  #数据源
+  datasource:
+    #多数据源
+    multi:
+      #主要
+      primary:
+        url: jdbc:postgresql://119.36.149.8:5433/watersmart?useSSL=false&useAffectedRows=false&allowMultiQueries=true&rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai
+        username: postgres
+        password: kpr.23417.postgres
+        driver-class-name: org.postgresql.Driver
+  #influx
+  influx:
+    read-timeout: 30000
+    clients:
+      #监利
+      - url: http://192.168.10.103:8060/
+        user: kpr
+        password: kpr.2025@117.influxdb
+        database: iot
+#对接
+calling:
+  #对接点
+  endpoints:
+    jianli-iot:
+      url: https://iot.cscec3b-iti.com
+      access-key: sKUJoJbisVJg7ei8NSKtPQ
+      secret-key: HtSU0Ul2NdJqLhK9uTa5BLetgzst0N4j9sIJ7dys

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/constants/AreaCode.java

@@ -14,6 +14,8 @@ import lombok.Getter;
 public enum AreaCode {
     //枣阳
     ZAO_YANG(710),
+    //监利
+    JIAN_LI(716),
     ;
     private final Integer code;
 }

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/constants/LogFlagBusiType.java

@@ -42,6 +42,7 @@ public enum LogFlagBusiType {
     BUSI_PIPE_BURST(38,"Pipe Burst Biz"),
     ZAO_YANG_DMA(39,"Zao Yang Dma"),
     ZHONG_HUAN_SACADA(40,"Zhong Huan Sacada"),
+    JIAN_LI_DMA(41,"Jian Li Dma"),
 
     BUSI_INTERNAL(99,"Internal Busi"),
 

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/domain/IntegrationKeyLoader.java

@@ -0,0 +1,24 @@
+package com.shkpr.service.customgateway.core.domain;
+
+/**
+ * 密钥加载器
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+public interface IntegrationKeyLoader {
+    /**
+     * 获取平台名称
+     *
+     * @return 平台名称
+     */
+    String getPlatformName();
+
+    /**
+     * 加载密钥
+     *
+     * @param endpoint 对接信息
+     * @return 密钥
+     */
+    IntegrationKey loadKey(CallingEndpoint endpoint);
+}

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/manager/IntegrationKeyManager.java

@@ -0,0 +1,125 @@
+package com.shkpr.service.customgateway.core.manager;
+
+import com.global.base.log.LogLevelFlag;
+import com.global.base.log.LogPrintMgr;
+import com.shkpr.service.customgateway.core.constants.LogFlagBusiType;
+import com.shkpr.service.customgateway.core.domain.CallingEndpoint;
+import com.shkpr.service.customgateway.core.domain.IntegrationKey;
+import com.shkpr.service.customgateway.core.domain.IntegrationKeyLoader;
+import com.shkpr.service.customgateway.core.properties.CallingProperties;
+import org.apache.commons.collections4.MapUtils;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 密钥管理器
+ *
+ * @author 欧阳劲驰
+ * @serial 1.0.0
+ */
+@Component
+public class IntegrationKeyManager {
+    /**
+     * log
+     */
+    private static final String CLASS_NAME = "IntegrationKeyManager";
+    private static final String BIZ_TYPE = LogFlagBusiType.BUSI_ALL.toStrValue();
+
+    final
+    ThreadPoolTaskExecutor taskScheduler;
+    final
+    CallingProperties callingProperties;
+
+    //密钥加载器
+    private final Map<String, IntegrationKeyLoader> keyLoaders = new ConcurrentHashMap<>();
+    //密钥缓存
+    private final Map<String, IntegrationKey> keyCache = new ConcurrentHashMap<>();
+
+    public IntegrationKeyManager(ThreadPoolTaskExecutor taskScheduler, CallingProperties callingProperties, List<IntegrationKeyLoader> loaders) {
+        this.taskScheduler = taskScheduler;
+        this.callingProperties = callingProperties;
+        // 注册所有密钥加载器
+        loaders.forEach(loader -> keyLoaders.put(loader.getPlatformName(), loader));
+    }
+
+    /**
+     * 初始化
+     */
+    @PostConstruct
+    public void init() {
+        taskScheduler.execute(this::refreshAllKeys);
+    }
+
+    /**
+     * 日任务
+     */
+    @Scheduled(cron = "0 0 0 * * ?")
+    public void minuteTask() {
+        taskScheduler.execute(this::refreshAllKeys);
+    }
+
+    /**
+     * 获取密钥
+     *
+     * @param platformName 平台名称
+     * @return 密钥
+     */
+    public IntegrationKey getKey(String platformName) {
+        IntegrationKey key = keyCache.get(platformName);
+        if (key == null) refreshKey(platformName);
+        return keyCache.get(platformName);
+    }
+
+    /**
+     * 刷新密钥
+     *
+     * @param platformName 平台名称
+     */
+    public void refreshKey(String platformName) {
+        //密钥加载器
+        IntegrationKeyLoader loader = keyLoaders.getOrDefault(platformName, null);
+        if (loader == null) {
+            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME,
+                    String.format("刷新密钥失败,加载器不存在,未找到平台[%s]对应的密钥加载器", platformName));
+            return;
+        }
+        //对接信息
+        CallingEndpoint endpoint = callingProperties.getEndpoint(platformName);
+        if (endpoint == null) {
+            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME,
+                    String.format("刷新密钥失败,对接信息不存在,未找到平台[%s]的对接信息", platformName));
+            return;
+        }
+        //加载密钥
+        IntegrationKey key = loader.loadKey(endpoint);
+        if (key == null) {
+            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME,
+                    String.format("刷新密钥失败,平台[%s]密钥加载返回空值", platformName));
+            return;
+        }
+        keyCache.put(platformName, key);
+    }
+
+    /**
+     * 刷新所有平台密钥
+     */
+    public void refreshAllKeys() {
+        if (MapUtils.isEmpty(keyLoaders)) return;
+
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME,
+                String.format("开始刷新所有平台密钥，共%d个平台", keyLoaders.size()));
+        long begin = System.currentTimeMillis();
+
+        for (String platformName : keyLoaders.keySet()) taskScheduler.execute(() -> refreshKey(platformName));
+
+        long end = System.currentTimeMillis();
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , String.format("结束刷新所有平台密钥,用时(毫秒):%d", (end - begin)));
+    }
+}

custom-gateway-core/src/main/java/com/shkpr/service/customgateway/core/utils/CallingUtil.java

@@ -104,9 +104,6 @@ public class CallingUtil {
             if (!result.isOk()) throw new SelfException(ResponseCode.RESULT_BAD.getCode() + "", result.getMessage());
 
             //返回数据
-            LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, mBizType, mStrClassName
-                    , String.format("请求成功:%s", result)
-            );
             return result.getData();
         } catch (IOException e) {
             LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, mBizType, mStrClassName
@@ -172,11 +169,8 @@ public class CallingUtil {
                         .execute();
 
                 //解析结果
-                R result = objectMapper.readValue(response.returnContent().toString(), resultType);
+                R result = objectMapper.readValue(response.returnContent().asString(StandardCharsets.UTF_8), resultType);
                 if (!result.isOk()) throw new SelfException(ResponseCode.RESULT_BAD.getCode() + "", result.getMessage());
-                LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, mBizType, mStrClassName
-                        , String.format("请求成功:%s", result)
-                );
 
                 //解析当前页，并存入数据
                 P page = result.getData();

custom-gateway-jldma/pom.xml

@@ -0,0 +1,49 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <!--父工程信息-->
+    <parent>
+        <groupId>com.shkpr.service</groupId>
+        <artifactId>kpr-custom-gateway</artifactId>
+        <version>1.0.0</version>
+    </parent>
+
+    <!--工件名-->
+    <artifactId>custom-gateway-jldma</artifactId>
+    <!--版本号-->
+    <version>${project.version}</version>
+    <!--打包方式-->
+    <packaging>jar</packaging>
+    <!--项目名-->
+    <name>CustomGatewayJldma</name>
+    <!--项目描述-->
+    <description>中环数采</description>
+
+    <!--依赖项-->
+    <dependencies>
+        <!--核心模块-->
+        <dependency>
+            <groupId>com.shkpr.service</groupId>
+            <artifactId>custom-gateway-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+    <!--构建脚本-->
+    <build>
+        <!--插件项-->
+        <plugins>
+            <!--maven-install-->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-install-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>default-install</id>
+                        <phase>none</phase>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/components/DataCollector.java

@@ -0,0 +1,131 @@
+package com.shkpr.service.customgateway.jldma.components;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.global.base.log.LogLevelFlag;
+import com.global.base.log.LogPrintMgr;
+import com.shkpr.service.customgateway.core.components.DeviceRegistry;
+import com.shkpr.service.customgateway.core.constants.LogFlagBusiType;
+import com.shkpr.service.customgateway.core.domain.CallingEndpoint;
+import com.shkpr.service.customgateway.core.domain.Device;
+import com.shkpr.service.customgateway.core.domain.IntegrationKey;
+import com.shkpr.service.customgateway.core.exception.SelfException;
+import com.shkpr.service.customgateway.core.manager.IntegrationKeyManager;
+import com.shkpr.service.customgateway.core.properties.CallingProperties;
+import com.shkpr.service.customgateway.core.utils.CallingUtil;
+import com.shkpr.service.customgateway.core.utils.InfluxDbUtil;
+import com.shkpr.service.customgateway.jldma.constants.IotPlatformMetadata;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformData;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformResult;
+import org.apache.http.Header;
+import org.apache.http.HttpHeaders;
+import org.apache.http.message.BasicHeader;
+import org.influxdb.dto.Point;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * 数据采集器
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Component
+public class DataCollector {
+    /**
+     * log
+     */
+    private static final String CLASS_NAME = "DataCollector";
+    private static final String BIZ_TYPE = LogFlagBusiType.BUSI_ALL.toStrValue();
+
+    final
+    CallingProperties callingProperties;
+    final
+    IntegrationKeyManager integrationKeyManager;
+    final
+    DeviceRegistry deviceRegistry;
+    final
+    InfluxDbUtil influxDbUtil;
+    final
+    CallingUtil callingUtil;
+
+
+    public DataCollector(CallingProperties callingProperties, IntegrationKeyManager integrationKeyManager, DeviceRegistry deviceRegistry, InfluxDbUtil influxDbUtil, CallingUtil callingUtil) {
+        this.callingProperties = callingProperties;
+        this.integrationKeyManager = integrationKeyManager;
+        this.deviceRegistry = deviceRegistry;
+        this.influxDbUtil = influxDbUtil;
+        this.callingUtil = callingUtil;
+    }
+
+
+    /**
+     * 采集物联网平台
+     *
+     * @param previousHours 往前回溯小时数
+     */
+    public void collectIotPlatform(int previousHours) {
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , "开始采集物联网平台数据,开始拉取数据");
+        long begin = System.currentTimeMillis();
+
+        //平台对接点
+        CallingEndpoint endpoint = callingProperties.getEndpoint(IotPlatformMetadata.NAME);
+        //密钥
+        IntegrationKey key = integrationKeyManager.getKey(IotPlatformMetadata.NAME);
+        if (key == null) return;
+
+        //请求地址
+        String url = endpoint.getUrl() + IotPlatformMetadata.Uri.QUERY_EVENT;
+        //请求头
+        List<Header> headers = Arrays.asList(
+                new BasicHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE),
+                new BasicHeader(key.getAccessKey(), key.getSecretKey())
+        );
+        //查询时间
+        LocalDateTime beginTime = LocalDateTime.now().truncatedTo(ChronoUnit.HOURS).minusHours(previousHours);
+        LocalDateTime endTime = LocalDateTime.now().truncatedTo(ChronoUnit.HOURS);
+
+        //结果
+        List<Point> results = new ArrayList<>();
+        //按设备遍历
+        for (Device device : deviceRegistry.findAll(IotPlatformMetadata.NAME)) {
+            //参数
+            Map<String, Object> params = IotPlatformMetadata.getHistoryDataParams(device.getDeviceSn(), beginTime, endTime);
+            //请求结果项
+            List<IotPlatformData> items;
+            try {
+                items = callingUtil.requestList(url, HttpMethod.POST, params, headers,
+                        new TypeReference<IotPlatformResult<List<IotPlatformData>>>() {
+                        });
+                LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                        , String.format("拉取数据成功,数据量:%d", items.size()));
+            } catch (SelfException e) {
+                LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME
+                        , String.format("拉取数据失败 error:%s", e)
+                );
+                continue;
+            }
+            //构建influxdb
+            results.addAll(items.stream()
+                    .flatMap(it -> it.toPoint(device).stream())
+                    .collect(Collectors.toList()));
+        }
+
+        //写入influxdb
+        influxDbUtil.write(results);
+
+        long end = System.currentTimeMillis();
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , String.format("结束采集物联网平台数据 用时(毫秒):%d", (end - begin))
+        );
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/components/InfoSynchronizer.java

@@ -0,0 +1,219 @@
+package com.shkpr.service.customgateway.jldma.components;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.global.base.log.LogLevelFlag;
+import com.global.base.log.LogPrintMgr;
+import com.shkpr.service.customgateway.core.components.DeviceIdGenerator;
+import com.shkpr.service.customgateway.core.components.DeviceRegistry;
+import com.shkpr.service.customgateway.core.constants.LogFlagBusiType;
+import com.shkpr.service.customgateway.core.domain.CallingEndpoint;
+import com.shkpr.service.customgateway.core.domain.Device;
+import com.shkpr.service.customgateway.core.domain.DeviceTag;
+import com.shkpr.service.customgateway.core.domain.IntegrationKey;
+import com.shkpr.service.customgateway.core.exception.SelfException;
+import com.shkpr.service.customgateway.core.manager.IntegrationKeyManager;
+import com.shkpr.service.customgateway.core.properties.CallingProperties;
+import com.shkpr.service.customgateway.core.service.FunctionInfoService;
+import com.shkpr.service.customgateway.core.service.PersonnelInfoService;
+import com.shkpr.service.customgateway.core.utils.CallingUtil;
+import com.shkpr.service.customgateway.jldma.constants.IotPlatformMetadata;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformDevice;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformPage;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformResult;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.message.BasicHeader;
+import org.springframework.http.HttpMethod;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * 信息同步器
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Component
+@Slf4j
+public class InfoSynchronizer {
+    /**
+     * log
+     */
+    private static final String CLASS_NAME = "InfoSynchronizer";
+    private static final String BIZ_TYPE = LogFlagBusiType.JIAN_LI_DMA.toStrValue();
+
+    final
+    CallingProperties callingProperties;
+    final
+    IntegrationKeyManager integrationKeyManager;
+    final
+    DeviceRegistry deviceRegistry;
+    final
+    DeviceIdGenerator deviceIdGenerator;
+    final
+    PersonnelInfoService personnelInfoService;
+    final
+    FunctionInfoService functionInfoService;
+    final
+    CallingUtil callingUtil;
+
+    public InfoSynchronizer(CallingProperties callingProperties, IntegrationKeyManager integrationKeyManager, DeviceRegistry deviceRegistry, DeviceIdGenerator deviceIdGenerator, PersonnelInfoService personnelInfoService, FunctionInfoService functionInfoService, CallingUtil callingUtil) {
+        this.callingProperties = callingProperties;
+        this.integrationKeyManager = integrationKeyManager;
+        this.deviceRegistry = deviceRegistry;
+        this.deviceIdGenerator = deviceIdGenerator;
+        this.personnelInfoService = personnelInfoService;
+        this.functionInfoService = functionInfoService;
+        this.callingUtil = callingUtil;
+    }
+
+    /**
+     * 同步设备信息
+     */
+    public void syncDevices() {
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , "开始同步设备信息,开始滚动拉取数据");
+        long begin = System.currentTimeMillis();
+
+        //对接点
+        CallingEndpoint endpoint = callingProperties.getEndpoint(IotPlatformMetadata.NAME);
+
+        //按设备类型遍历
+        for (IotPlatformMetadata.DeviceMapping deviceMapping : IotPlatformMetadata.DeviceMapping.values()) {
+            //请求地址
+            String url = endpoint.getUrl() + IotPlatformMetadata.Uri.DEVICE;
+            //请求获取设备
+            List<IotPlatformDevice> items = Collections.emptyList();
+            try {
+                items = callingUtil.requestScroll(url, HttpMethod.GET,
+                        new TypeReference<IotPlatformResult<IotPlatformPage<IotPlatformDevice>>>() {
+                        }, pageable -> new HashMap<String, String>() {{
+                            put(IotPlatformMetadata.Params.PAGE, pageable.getPageNumber() + "");
+                            put(IotPlatformMetadata.Params.SIZE, pageable.getPageSize() + "");
+                            put(IotPlatformMetadata.Params.PRODUCT_CODE, IotPlatformMetadata.DefaultValues.PRODUCT_CODE);
+                            put(IotPlatformMetadata.Params.DEVICE_TYPE, deviceMapping.getMapping());
+                        }}, params -> {
+                            //获取密钥
+                            IntegrationKey key = integrationKeyManager.getKey(IotPlatformMetadata.NAME);
+                            //存入请求头
+                            return Collections.singletonList(new BasicHeader(key.getAccessKey(), key.getSecretKey()));
+                        });
+            } catch (SelfException e) {
+                LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME
+                        , String.format("获取设备失败 error:%s", e)
+                );
+            }
+
+            //sn映射
+            Map<String, IotPlatformDevice> snMap = items.stream().collect(Collectors.toMap(IotPlatformDevice::getSn, Function.identity(), (it1, it2) -> it2));
+            //注册设备
+            registryDevices(deviceMapping, snMap);
+            //更新设备
+            updateDevices(deviceMapping, snMap);
+        }
+
+        long end = System.currentTimeMillis();
+        LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                , String.format("结束执行同步设备信息,用时(毫秒):%d", (end - begin))
+        );
+    }
+
+    /**
+     * 注册设备
+     *
+     * @param deviceMapping 设备类型映射
+     * @param snMap         数据集合
+     */
+    private void registryDevices(IotPlatformMetadata.DeviceMapping deviceMapping, Map<String, IotPlatformDevice> snMap) {
+        //需要添加的设备
+        Set<String> addedDevices = deviceRegistry.findAddedDevices(snMap.keySet());
+        //处理需要添加的设备:生成设备,并注册
+        addedDevices.stream()
+                //过滤无效设备
+                .filter(sn -> snMap.get(sn) != null && StringUtils.isNoneBlank(snMap.get(sn).getName()))
+                //排序
+                .sorted()
+                //生成设备
+                .map(sn -> deviceIdGenerator.generateDevice(
+                        IotPlatformMetadata.Devices.AREA_CODE,
+                        deviceMapping.getKind(),
+                        IotPlatformMetadata.NAME,
+                        buildTag(deviceMapping, snMap.get(sn)),
+                        sn, snMap.get(sn).getName()
+                ))
+                //设置厂家
+                .peek(device -> device.setMfrs(deviceMapping.getName()))
+
+                //注册
+                .forEach(deviceRegistry::registerDevice);
+    }
+
+    /**
+     * 更新设备
+     *
+     * @param deviceMapping 设备类型映射
+     * @param snMap         数据集合
+     */
+    private void updateDevices(IotPlatformMetadata.DeviceMapping deviceMapping, Map<String, IotPlatformDevice> snMap) {
+        //构建设备信息,k:远传id,v:标签集合
+        Map<String, Set<String>> deviceInfo = snMap.entrySet().stream().collect(Collectors.toMap(
+                Map.Entry::getKey, entry -> {
+                    //获取设备属性
+                    IotPlatformDevice.SchemaAttr schemaAttr = entry.getValue().getSchemaAttrs().stream()
+                            .filter(it -> Objects.equals(it.getDeviceType() + "", deviceMapping.getMapping()))
+                            .findFirst().orElse(null);
+                    if (schemaAttr == null) return Collections.emptySet();
+                    //标签
+                    Set<String> tags = schemaAttr.getData().getProperties().keySet();
+                    //过滤存在的标签
+                    return deviceMapping.getFieldMappings().stream()
+                            .filter(fieldMapping -> tags.contains(fieldMapping.getMapping()))
+                            .map(fieldMapping -> fieldMapping.getField().getKey())
+                            .collect(Collectors.toSet());
+                }
+        ));
+        //需要更新的设备
+        Set<Device> updatedDevices = deviceRegistry.findUpdatedDevices(deviceInfo);
+        updatedDevices.forEach(device -> {
+            IotPlatformDevice item = snMap.get(device.getDeviceSn());
+
+            device.setTags(buildTag(deviceMapping, item));
+
+            deviceRegistry.updateDevice(device);
+        });
+    }
+
+    /**
+     * 构建采集标签
+     *
+     * @param deviceMapping 设备映射
+     * @param item          结果项
+     * @return 采集标签集合
+     */
+    private List<DeviceTag> buildTag(IotPlatformMetadata.DeviceMapping deviceMapping, IotPlatformDevice item) {
+        //获取设备属性
+        IotPlatformDevice.SchemaAttr schemaAttr = item.getSchemaAttrs().stream().filter(it -> Objects.equals(it.getDeviceType() + "", deviceMapping.getMapping()))
+                .findFirst().orElse(null);
+        if (schemaAttr == null) return Collections.emptyList();
+
+        //标签
+        Set<String> tags = schemaAttr.getData().getProperties().keySet();
+
+        //构建标签
+        return deviceMapping.getFieldMappings().stream()
+                //过滤存在的标签
+                .filter(fieldMapping -> tags.contains(fieldMapping.getMapping()))
+                //转换为采集标签
+                .map(fieldMapping -> new DeviceTag(
+                        fieldMapping.getMapping(),
+                        deviceMapping.getProtocol().getKey(),
+                        deviceMapping.getKind().getMeasurement(),
+                        fieldMapping.getField().getKey(),
+                        fieldMapping.getValueType().getKey()
+                ))
+                .collect(Collectors.toList());
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/components/IotPlatformKeyLoader.java

@@ -0,0 +1,76 @@
+package com.shkpr.service.customgateway.jldma.components;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.shkpr.service.customgateway.core.domain.CallingEndpoint;
+import com.shkpr.service.customgateway.core.domain.IntegrationKey;
+import com.shkpr.service.customgateway.core.domain.IntegrationKeyLoader;
+import com.shkpr.service.customgateway.jldma.constants.IotPlatformMetadata;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformResult;
+import com.shkpr.service.customgateway.jldma.domain.IotPlatformToken;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.fluent.Request;
+import org.apache.http.client.fluent.Response;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.util.HashMap;
+
+/**
+ * 物联网平台
+ */
+@Component
+public class IotPlatformKeyLoader implements IntegrationKeyLoader {
+    final
+    ObjectMapper objectMapper;
+
+    public IotPlatformKeyLoader(ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper;
+    }
+
+    /**
+     * 获取平台名称
+     *
+     * @return 平台名称
+     */
+    @Override
+    public String getPlatformName() {
+        return IotPlatformMetadata.NAME;
+    }
+
+    /**
+     * 加载密钥
+     *
+     * @param endpoint 对接信息
+     * @return 密钥
+     */
+    @Override
+    public IntegrationKey loadKey(CallingEndpoint endpoint) {
+        try {
+            //请求token
+            Response response = Request.Post(endpoint.getUrl() + IotPlatformMetadata.Uri.TOKEN)
+                    .addHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
+                    .bodyByteArray(objectMapper.writeValueAsBytes(
+                            new HashMap<String, String>() {{
+                                put(IotPlatformMetadata.Params.ACCESS_KEY, endpoint.getAccessKey());
+                                put(IotPlatformMetadata.Params.SECURITY_KEY, endpoint.getSecretKey());
+                            }}
+                    ))
+                    .execute();
+
+
+            //解析token
+            IotPlatformResult<IotPlatformToken> tokenResult = objectMapper.readValue(response.returnContent().asString(), new TypeReference<IotPlatformResult<IotPlatformToken>>() {
+            });
+            IotPlatformToken token = tokenResult.getData();
+            if (!tokenResult.isOk() || !ObjectUtils.allNotNull(token, token.getTokenKey(), token.getTokenValue()))
+                return null;
+
+            return new IntegrationKey(null, token.getTokenKey(), token.getTokenValue(), null);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/constants/IotPlatformMetadata.java

@@ -0,0 +1,148 @@
+package com.shkpr.service.customgateway.jldma.constants;
+
+import com.shkpr.service.customgateway.core.constants.*;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.*;
+
+/**
+ * 物联网平台集成元数据
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+public abstract class IotPlatformMetadata {
+    //系统名称
+    public static String NAME = "jianli-iot";
+
+
+    /**
+     * 获取历史数据参数
+     *
+     * @param beginTime     开始时间
+     * @param endTime       结束时间
+     * @return 参数
+     */
+    public static Map<String, Object> getHistoryDataParams(String sn, LocalDateTime beginTime, LocalDateTime endTime) {
+        return new HashMap<String, Object>() {{
+            put(Params.START_TIME, beginTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+            put(Params.END_TIME, endTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+            put(Params.EVENT_ITEMS, Collections.singletonList(
+                    new HashMap<String, String>() {{
+                        put(Params.PRODUCT_CODE, DefaultValues.PRODUCT_CODE);
+                        put(Params.SN, sn);
+                        put(Params.EVENT_CODE, DefaultValues.EVENT_CODE);
+                    }}
+            ));
+        }};
+    }
+
+    /**
+     * 设备映射
+     */
+    @Getter
+    @AllArgsConstructor
+    public enum DeviceMapping {
+        //流量压力计
+        FLOW_PRESS(DeviceKind.FLOW_PRESS, "413", "", ProtocolType.JSON, Arrays.asList(
+                new FieldMapping(DeviceField.FLOW_CUR, "instantFlow", ValueType.DOUBLE_FLOAT),
+                new FieldMapping(DeviceField.FLOW_TOTAL_POS, "forwardFlow", ValueType.DOUBLE_FLOAT),
+                new FieldMapping(DeviceField.FLOW_TOTAL_REV, "reverseFlow", ValueType.DOUBLE_FLOAT),
+                new FieldMapping(DeviceField.PRESS_CUR, "pressure", ValueType.DOUBLE_FLOAT)
+        ));
+
+        //类型
+        private final DeviceKind kind;
+        //映射
+        private final String mapping;
+        //名称
+        private final String name;
+        //协议类型
+        private final ProtocolType protocol;
+        //字段映射
+        private final List<FieldMapping> fieldMappings;
+
+        @Getter
+        @Setter
+        @AllArgsConstructor
+        public static class FieldMapping {
+            //字段
+            private DeviceField field;
+            //映射
+            private String mapping;
+            //值类型
+            private ValueType valueType;
+        }
+
+    }
+
+    /**
+     * 接口地址
+     */
+    public interface Uri {
+        //获取token
+        String TOKEN = "/api/webgateway/v1/access/token";
+        //获取设备
+        String DEVICE = "/api/hermes/v1/device";
+        //查询事件
+        String QUERY_EVENT = "/api/data/v1/tsdb/queryEvent";
+    }
+
+    /**
+     * 参数
+     */
+    public interface Params {
+        //访问密钥
+        String ACCESS_KEY = "ak";
+        //认证密钥
+        String SECURITY_KEY = "sk";
+        //页码
+        String PAGE = "page";
+        //分页大小
+        String SIZE = "size";
+        //产品编码
+        String PRODUCT_CODE = "productCode";
+        //设备类型
+        String DEVICE_TYPE = "deviceType";
+        //设备标识
+        String SN = "sn";
+        //开始时间
+        String START_TIME = "startTime";
+        //结束时间
+        String END_TIME = "endTime";
+        //查询条件
+        String EVENT_ITEMS = "eventItems";
+        //条件code
+        String EVENT_CODE = "eventCode";
+    }
+
+    /**
+     * 结果
+     */
+    public interface Results {
+        //采集时间
+        String COLLECT_TIME = "collectTime";
+    }
+
+    /**
+     * 默认值
+     */
+    public interface DefaultValues {
+        //产品编码
+        String PRODUCT_CODE = "38df4e7d0566499caf88ae28a39a2b23";
+        //条件code
+        String EVENT_CODE = "ATTR";
+    }
+
+    /**
+     * 设备
+     */
+    public interface Devices {
+        //区号
+        AreaCode AREA_CODE = AreaCode.JIAN_LI;
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/domain/IotPlatformData.java

@@ -0,0 +1,107 @@
+package com.shkpr.service.customgateway.jldma.domain;
+
+
+import com.shkpr.service.customgateway.core.constants.InfluxdbMetadata;
+import com.shkpr.service.customgateway.core.domain.Device;
+import com.shkpr.service.customgateway.core.domain.DeviceTag;
+import com.shkpr.service.customgateway.core.utils.ExpressionUtil;
+import com.shkpr.service.customgateway.jldma.constants.IotPlatformMetadata;
+import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.influxdb.dto.Point;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * 物联网平台数据
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Data
+public class IotPlatformData {
+    /**
+     * 时间
+     */
+    private Long timestamp;
+    /**
+     * 索引
+     */
+    private String metricIndex;
+    /**
+     * uuid
+     */
+    private Object uuid;
+    /**
+     * 结果
+     */
+    private Map<String, Object> resultMap;
+
+    /**
+     * 转influxdb对象
+     *
+     * @param device 设备
+     * @return influxdb对象
+     */
+    public List<Point> toPoint(Device device) {
+        //日期格式化
+        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+        //解析采集时间
+        String collectTimeStr = (String) resultMap.getOrDefault(IotPlatformMetadata.Results.COLLECT_TIME, null);
+        if (StringUtils.isBlank(collectTimeStr)) return Collections.emptyList();
+        LocalDateTime collectTime = LocalDateTime.parse(collectTimeStr, formatter);
+
+        //解析结果
+        return resultMap.entrySet().stream().map(it -> {
+            //结果标签
+            String k = it.getKey();
+            if (StringUtils.isBlank(k)) return null;
+            //结果值
+            String v = String.valueOf(it.getValue());
+            if (!NumberUtils.isParsable(v)) return null;
+
+            //获取标签
+            DeviceTag tag = device.getTags().stream()
+                    .filter(t -> Objects.equals(k, t.getTag()))
+                    .findFirst().orElse(null);
+            if (tag == null) return null;
+
+            //值
+            Double value = Double.parseDouble(v);
+            //应用公式计算
+            if (StringUtils.isNotBlank(tag.getCalcFormula()))
+                value = ExpressionUtil.evaluateFormula(value, tag.getCalcFormula());
+            //数值范围校验
+            if (tag.getMinValue() != null && value < tag.getMinValue()) value = tag.getMinValue();
+            if (tag.getMaxValue() != null && value > tag.getMaxValue()) value = tag.getMaxValue();
+
+            //获取时间戳
+            long timestamp = collectTime
+                    .truncatedTo(ChronoUnit.MINUTES)
+                    .atZone(ZoneId.systemDefault())
+                    .toInstant()
+                    .toEpochMilli();
+            //构建influxdb
+            return Point
+                    //表名
+                    .measurement(tag.getMeasurement())
+                    //设备id
+                    .tag(InfluxdbMetadata.Tags.DEVICE_ID, device.getDeviceId())
+                    //时间
+                    .time(timestamp, TimeUnit.MILLISECONDS)
+                    //值
+                    .addField(tag.getField(), value)
+                    .build();
+        }).filter(Objects::nonNull).collect(Collectors.toList());
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/domain/IotPlatformDevice.java

@@ -0,0 +1,332 @@
+package com.shkpr.service.customgateway.jldma.domain;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import lombok.Data;
+import org.springframework.format.annotation.DateTimeFormat;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 物联网平台设备
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Data
+public class IotPlatformDevice {
+    /**
+     * 设备ID
+     */
+    private Long id;
+
+    /**
+     * 产品编码
+     */
+    private String productCode;
+
+    /**
+     * 产品名称
+     */
+    private String productName;
+
+    /**
+     * 设备名称
+     */
+    private String name;
+
+    /**
+     * 设备类型编码
+     */
+    private Integer type;
+
+    /**
+     * 设备类型中文名称
+     */
+    private String typeCn;
+
+    /**
+     * 设备序列号
+     */
+    private String sn;
+
+    /**
+     * 节点类型编码
+     */
+    private Integer nodeType;
+
+    /**
+     * 节点类型中文名称
+     */
+    private String nodeTypeCn;
+
+    /**
+     * 访问令牌
+     */
+    private String accessToken;
+
+    /**
+     * 协议参数
+     */
+    private Map<String, String> protocolParams;
+
+    /**
+     * 网关设备key
+     */
+    private String gatewayKey;
+
+    /**
+     * 扩展字段
+     */
+    private Map<String, Object> ext;
+
+    /**
+     * 创建时间
+     */
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", locale = "zh_CN", timezone = "Asia/Shanghai")
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private LocalDateTime createTime;
+
+    /**
+     * 更新时间
+     */
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", locale = "zh_CN", timezone = "Asia/Shanghai")
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private LocalDateTime updateTime;
+
+    /**
+     * 子设备数量
+     */
+    private Integer childCount;
+
+    /**
+     * 设备唯一key
+     */
+    private String deviceKey;
+
+    /**
+     * 设备在线状态
+     */
+    private Boolean online;
+
+    /**
+     * 最后在线时间
+     */
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", locale = "zh_CN", timezone = "Asia/Shanghai")
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private LocalDateTime lastOnlineTime;
+
+    /**
+     * 协议编码
+     */
+    private String protocolCode;
+
+    /**
+     * 公司ID
+     */
+    private Long companyId;
+
+    /**
+     * 公司名称
+     */
+    private String companyName;
+
+    /**
+     * 品牌ID
+     */
+    private Long brandId;
+
+    /**
+     * 品牌名称
+     */
+    private String brandName;
+
+    /**
+     * 型号名称
+     */
+    private String modelName;
+
+    /**
+     * 可进行的操作列表
+     */
+    private List<String> operations;
+
+    /**
+     * 是否禁用
+     */
+    private Boolean disabled;
+
+    /**
+     * 所属应用
+     */
+    private Integer ownApp;
+
+    /**
+     * 用户ID
+     */
+    private Integer userId;
+
+    /**
+     * 布局信息
+     */
+    private Object layOut;
+
+    /**
+     * 属性模式定义
+     */
+    private List<SchemaAttr> schemaAttrs;
+
+    /**
+     * 包版本
+     */
+    private String packageVersion;
+
+    /**
+     * 是否有品牌
+     */
+    private Boolean branded;
+
+    /**
+     * 在线状态编码
+     */
+    private Integer onlineStatus;
+
+    /**
+     * 开关事件时间戳
+     */
+    private Long onOffEventTs;
+
+    /**
+     * 删除标识
+     */
+    private Integer deleted;
+
+    /**
+     * 网关名称
+     */
+    private String gatewayName;
+
+    /**
+     * 生产日期
+     */
+    private LocalDateTime productionDate;
+
+    /**
+     * 保修信息
+     */
+    private Object warranty;
+
+    /**
+     * 保修类型
+     */
+    private Integer warrantyType;
+
+    /**
+     * 属性
+     */
+    @Data
+    public static class SchemaAttr {
+        /**
+         * 属性ID
+         */
+        private Long id;
+
+        /**
+         * 属性编码
+         */
+        private String code;
+
+        /**
+         * 属性数据
+         */
+        private SchemaData data;
+
+        /**
+         * 模式
+         */
+        private Integer mode;
+
+        /**
+         * 属性名称
+         */
+        private String name;
+
+        /**
+         * 属性类型
+         */
+        private Integer type;
+
+        /**
+         * 回复信息
+         */
+        private Object reply;
+
+        /**
+         * 删除标识
+         */
+        private Integer deleted;
+
+        /**
+         * 是否必填
+         */
+        private Boolean required;
+
+        /**
+         * 设备类型编码
+         */
+        private Integer deviceType;
+
+        /**
+         * 设备类型中文名称
+         */
+        private String deviceTypeCn;
+    }
+
+    /**
+     * 属性数据
+     */
+    @Data
+    public static class SchemaData {
+        /**
+         * 数据类型
+         */
+        private String type;
+
+        /**
+         * 属性定义Map
+         */
+        private Map<String, Property> properties;
+    }
+
+    /**
+     * 配置
+     */
+    @Data
+    public static class Property {
+        /**
+         * 模式
+         */
+        private Integer mode;
+
+        /**
+         * 属性类型
+         */
+        private String type;
+
+        /**
+         * 标题
+         */
+        private String title;
+
+        /**
+         * 单位符号
+         */
+        private String symbol;
+
+        /**
+         * 描述信息
+         */
+        private String description;
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/domain/IotPlatformPage.java

@@ -0,0 +1,61 @@
+package com.shkpr.service.customgateway.jldma.domain;
+
+import com.shkpr.service.customgateway.core.domain.PageResponse;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NonNull;
+
+import java.util.List;
+
+/**
+ * 物联网平台分页
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+public class IotPlatformPage<T> extends PageResponse<T> {
+    /**
+     * 页码数
+     */
+    private Integer page;
+    /**
+     * 分页数
+     */
+    private int size;
+    /**
+     * 数量
+     */
+    private Long total;
+    /**
+     * 内容
+     */
+    private List<T> list;
+
+    @Override
+    public Integer getPageNumber() {
+        return this.page;
+    }
+
+    @Override
+    public Integer getPageSize() {
+        return this.size;
+    }
+
+    @Override
+    public long getTotalElements() {
+        return this.total;
+    }
+
+    @Override
+    public int getTotalPages() {
+        return (int) (this.total % size == 0 ? this.total / size : this.total / size + 1);
+    }
+
+    @Override
+    @NonNull
+    public List<T> getContent() {
+        return this.list;
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/domain/IotPlatformResult.java

@@ -0,0 +1,65 @@
+package com.shkpr.service.customgateway.jldma.domain;
+
+import com.shkpr.service.customgateway.core.domain.Result;
+import lombok.Data;
+
+/**
+ * 物联网平台结果
+ *
+ * @param <T> 数据类型
+ */
+@Data
+public class IotPlatformResult<T> implements Result<T> {
+    /**
+     * 状态码
+     */
+    private Integer status;
+    /**
+     * 消息
+     */
+    private String msg;
+    /**
+     * 数据
+     */
+    private T data;
+
+    /**
+     * 获取状态码
+     *
+     * @return 状态码
+     */
+    @Override
+    public Integer getCode() {
+        return this.status;
+    }
+
+    /**
+     * 获取信息
+     *
+     * @return 信息
+     */
+    @Override
+    public String getMessage() {
+        return this.msg;
+    }
+
+    /**
+     * 获取数据
+     *
+     * @return 数据
+     */
+    @Override
+    public T getData() {
+        return this.data;
+    }
+
+    /**
+     * 获取ok状态码
+     *
+     * @return ok状态码
+     */
+    @Override
+    public Boolean isOk() {
+        return this.status == 0;
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/domain/IotPlatformToken.java

@@ -0,0 +1,22 @@
+package com.shkpr.service.customgateway.jldma.domain;
+
+import lombok.Data;
+
+/**
+ * 物联网平台token
+ */
+@Data
+public class IotPlatformToken {
+    /**
+     * token密钥
+     */
+    private String tokenKey;
+    /**
+     * token值
+     */
+    private String tokenValue;
+    /**
+     * 过期时间
+     */
+    private Long expireTs;
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/manager/DataManager.java

@@ -0,0 +1,45 @@
+package com.shkpr.service.customgateway.jldma.manager;
+
+import com.shkpr.service.customgateway.jldma.components.DataCollector;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * 数据管理
+ *
+ * @author 欧阳劲驰
+ * @serial 1.0.0
+ */
+@Component
+public class DataManager {
+    final
+    ThreadPoolTaskExecutor taskScheduler;
+    final
+    DataCollector dataCollector;
+
+    public DataManager(ThreadPoolTaskExecutor taskScheduler, DataCollector dataCollector) {
+        this.taskScheduler = taskScheduler;
+        this.dataCollector = dataCollector;
+    }
+
+    /**
+     * 初始化
+     */
+    @PostConstruct
+    public void init() {
+        //采集物联网平台
+        taskScheduler.execute(() -> dataCollector.collectIotPlatform(30 * 24));
+    }
+
+    /**
+     * 小时任务
+     */
+    @Scheduled(cron = "0 * * * * ?")
+    public void minuteTask() {
+        //采集物联网平台
+        taskScheduler.execute(() -> dataCollector.collectIotPlatform(1));
+    }
+}

custom-gateway-jldma/src/main/java/com/shkpr/service/customgateway/jldma/manager/InfoSyncManager.java

@@ -0,0 +1,45 @@
+package com.shkpr.service.customgateway.jldma.manager;
+
+import com.shkpr.service.customgateway.jldma.components.InfoSynchronizer;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * 信息同步管理
+ *
+ * @author 欧阳劲驰
+ * @since 1.0.0
+ */
+@Component
+public class InfoSyncManager {
+    final
+    ThreadPoolTaskExecutor taskScheduler;
+    final
+    InfoSynchronizer infoSynchronizer;
+
+    public InfoSyncManager(ThreadPoolTaskExecutor taskScheduler, InfoSynchronizer infoSynchronizer) {
+        this.taskScheduler = taskScheduler;
+        this.infoSynchronizer = infoSynchronizer;
+    }
+
+    /**
+     * 初始化
+     */
+    @PostConstruct
+    public void init() {
+        //同步设备信息
+        taskScheduler.execute(infoSynchronizer::syncDevices);
+    }
+
+    /**
+     * 分钟任务
+     */
+    @Scheduled(cron = "0 */10 * * * *")
+    public void minuteTask() {
+        //同步设备信息
+        taskScheduler.execute(infoSynchronizer::syncDevices);
+    }
+}

custom-gateway-zydma/src/main/java/com/shkpr/service/customgateway/zydma/components/DataCollector.java

@@ -45,7 +45,7 @@ public class DataCollector {
      * log
      */
     private static final String CLASS_NAME = "DataCollector";
-    private static final String BIZ_TYPE = LogFlagBusiType.BUSI_ALL.toStrValue();
+    private static final String BIZ_TYPE = LogFlagBusiType.ZAO_YANG_DMA.toStrValue();
 
     final
     CallingProperties callingProperties;
@@ -217,7 +217,7 @@ public class DataCollector {
     }
 
     /**
-     * 注册设备
+     * 更新设备
      *
      * @param deviceMapping 设备类型映射
      * @param snGroup       数据集合
@@ -240,7 +240,7 @@ public class DataCollector {
                                 .map(optional -> optional.get().getField().getKey())
                                 .collect(Collectors.toSet())
         ));
-        //需要添加的设备
+        //需要更新的设备
         Set<Device> updatedDevices = deviceRegistry.findUpdatedDevices(deviceInfo);
         updatedDevices.forEach(device -> {
             List<IotPlatformData> items = snGroup.get(device.getDeviceSn());

custom-gateway-zydma/src/main/java/com/shkpr/service/customgateway/zydma/components/InfoSynchronizer.java

@@ -41,7 +41,7 @@ public class InfoSynchronizer {
      * log
      */
     private static final String CLASS_NAME = "InfoSynchronizer";
-    private static final String BIZ_TYPE = LogFlagBusiType.BUSI_ALL.toStrValue();
+    private static final String BIZ_TYPE = LogFlagBusiType.ZAO_YANG_DMA.toStrValue();
 
     final
     CallingProperties callingProperties;

pom.xml

@@ -28,6 +28,8 @@
 
         <!--枣阳漏控-->
         <module>custom-gateway-zydma</module>
+        <!--监利漏控-->
+        <module>custom-gateway-jldma</module>
         <!--中环数采-->
         <module>custom-gateway-zhscada</module>
     </modules>