离线通知

admin/src/main/java/com/kym/admin/controller/MonitorLogController.java

@@ -0,0 +1,18 @@
+package com.kym.admin.controller;
+
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * <p>
+ * 设备监控日志表 前端控制器
+ * </p>
+ *
+ * @author skyline
+ * @since 2023-08-29
+ */
+@RestController
+@RequestMapping("/monitor-log")
+public class MonitorLogController {
+
+}

entity/src/main/java/com/kym/entity/admin/EquipmentRelation.java

@@ -1,25 +1,21 @@
 package com.kym.entity.admin;
 
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
-import java.io.Serializable;
-import java.time.LocalDateTime;
-
 import com.kym.entity.BaseEntity;
 import lombok.Getter;
-import lombok.Setter;
+
+import java.io.Serializable;
 
 /**
  * <p>
- * 
+ *
  * </p>
  *
  * @author skyline
  * @since 2023-08-07
  */
 @Getter
-@Setter
 @TableName("t_equipment_relation")
 public class EquipmentRelation extends BaseEntity implements Serializable {
 
@@ -39,10 +35,31 @@ public class EquipmentRelation extends BaseEntity implements Serializable {
      * EN+充电的sn号
      */
     private String equipmentId;
+    /**
+     * 充电枪口编号
+     */
+    @TableField(exist = false)
+    private String connectorId;
 
     /**
      * 状态 0：启用，1：未启用
      */
     private Integer status;
 
+    public void setStationId(String stationId) {
+        this.stationId = stationId;
+    }
+
+    public void setShortId(String shortId) {
+        this.shortId = shortId;
+    }
+
+    public void setEquipmentId(String equipmentId) {
+        this.equipmentId = equipmentId;
+        this.connectorId = equipmentId.concat("1");
+    }
+
+    public void setStatus(Integer status) {
+        this.status = status;
+    }
 }

entity/src/main/java/com/kym/entity/admin/MonitorLog.java

@@ -0,0 +1,66 @@
+package com.kym.entity.admin;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.kym.entity.BaseEntity;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+
+import java.time.LocalDateTime;
+
+/**
+ * <p>
+ * 设备监控日志表
+ * </p>
+ *
+ * @author skyline
+ * @since 2023-08-29
+ */
+@Getter
+@Setter
+@TableName("t_monitor_log")
+@Accessors(chain = true)
+public class MonitorLog extends BaseEntity {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 站点id
+     */
+    private String stationId;
+
+    /**
+     * 类型：0未知 1网关 2充电桩
+     */
+    private Integer type;
+
+    /**
+     * 设备SN
+     */
+    private String sn;
+
+    /**
+     * 离线时间
+     */
+    private LocalDateTime offlineTime;
+
+    /**
+     * 离线原因状态 充电桩0-离网
+     */
+    private Integer offlineStatus;
+
+    /**
+     * 是否已通知:0未通知 1已通知
+     */
+    private Integer isNotice;
+
+    /**
+     * 是否已恢复 0:未恢复 1：已恢复
+     */
+    private Integer isRecover;
+
+    /**
+     * 恢复时间
+     */
+    private LocalDateTime recoverTime;
+}

entity/src/main/java/com/kym/entity/common/RedisKeys.java

@@ -7,4 +7,5 @@ package com.kym.entity.common;
  */
 public interface RedisKeys {
     String EN_PLUS_TOKEN = "EN_PLUS_TOKEN";
+    String OFFLINE = "OFFLINE";
 }

mapper/src/main/java/com/kym/mapper/admin/MonitorLogMapper.java

@@ -0,0 +1,16 @@
+package com.kym.mapper.admin;
+
+import com.kym.entity.admin.MonitorLog;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 设备监控日志表 Mapper 接口
+ * </p>
+ *
+ * @author skyline
+ * @since 2023-08-29
+ */
+public interface MonitorLogMapper extends BaseMapper<MonitorLog> {
+
+}

mapper/src/main/resources/mappers/admin/MonitorLogMapper.xml

@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.kym.mapper.admin.MonitorLogMapper">
+
+    <!-- 通用查询映射结果 -->
+    <resultMap id="BaseResultMap" type="com.kym.entity.admin.MonitorLog">
+        <result column="station_id" property="stationId" />
+        <result column="type" property="type" />
+        <result column="sn" property="sn" />
+        <result column="offline_time" property="offlineTime" />
+        <result column="offline_status" property="offlineStatus" />
+        <result column="is_notice" property="isNotice" />
+        <result column="is_recover" property="isRecover" />
+        <result column="recover_time" property="recoverTime" />
+    </resultMap>
+
+    <!-- 通用查询结果列 -->
+    <sql id="Base_Column_List">
+        station_id, type, sn, offline_time, offline_status, is_notice, is_recover, recover_time
+    </sql>
+
+</mapper>

miniapp/pom.xml

@@ -56,6 +56,13 @@
             <version>4.5.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.sun.mail</groupId>
+            <artifactId>javax.mail</artifactId>
+            <version>1.6.2</version>
+        </dependency>
+
+
 
     </dependencies>
 

miniapp/src/main/java/com/kym/miniapp/config/ThreadPoolTaskExecutorConfig.java

@@ -0,0 +1,25 @@
+package com.kym.miniapp.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+/**
+ * 异步线程池ThreadPoolExecutor 配置类
+ *
+ * @author skyline
+ * @date 2023-08-29 14:11
+ */
+@Configuration
+public class ThreadPoolTaskExecutorConfig {
+    @Bean
+    public ThreadPoolTaskScheduler syncScheduler() {
+        ThreadPoolTaskScheduler syncScheduler = new ThreadPoolTaskScheduler();
+        syncScheduler.setPoolSize(10);
+        // 这里给线程设置名字，主要是为了在项目能够更快速的定位错误。
+        syncScheduler.setThreadGroupName("syncTg");
+        syncScheduler.setThreadNamePrefix("syncThread-");
+        syncScheduler.initialize();
+        return syncScheduler;
+    }
+}

miniapp/src/main/java/com/kym/miniapp/jobs/DynamicTaskService.java

@@ -0,0 +1,108 @@
+package com.kym.miniapp.jobs;
+
+import cn.hutool.core.convert.ConverterRegistry;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledFuture;
+
+/**
+ * 动态定时任务
+ */
+@Component
+@Slf4j
+public class DynamicTaskService {
+
+    private final ThreadPoolTaskScheduler syncScheduler;
+    /**
+     * 以下两个都是线程安全的集合类。
+     */
+    public Map<String, ScheduledFuture<?>> taskMap = new ConcurrentHashMap<>();
+    public List<String> taskList = new CopyOnWriteArrayList<>();
+
+    public DynamicTaskService(ThreadPoolTaskScheduler syncScheduler) {
+        this.syncScheduler = syncScheduler;
+    }
+
+    /**
+     * 查看已开启但还未执行的动态任务
+     *
+     * @return
+     */
+    public List<String> getTaskList() {
+        return taskList;
+    }
+
+
+    /**
+     * 添加一个动态任务
+     *
+     * @param task
+     * @return
+     */
+    public boolean add(Task task) {
+        // 此处的逻辑是 ，如果当前已经有这个名字的任务存在，先删除之前的，再添加现在的。（即重复就覆盖）
+        if (null != taskMap.get(task.getName())) {
+            stop(task.getName());
+        }
+
+        // hutool 工具包下的一个转换类型工具类 好用的很
+        ConverterRegistry converterRegistry = ConverterRegistry.getInstance();
+        Date startTime = converterRegistry.convert(Date.class, task.getStartTime());
+
+        // schedule :调度给定的Runnable ，在指定的执行时间调用它。
+        //一旦调度程序关闭或返回的ScheduledFuture被取消，执行将结束。
+        //参数：
+        //任务 – 触发器触发时执行的 Runnable
+        //startTime – 任务所需的执行时间（如果这是过去，则任务将立即执行，即尽快执行）
+        ScheduledFuture<?> schedule = syncScheduler.schedule(task, startTime);
+        taskMap.put(task.getName(), schedule);
+        taskList.add(task.getName());
+        return true;
+    }
+
+
+    /**
+     * 运行任务
+     *
+     * @param task
+     * @return
+     */
+/*    public Runnable getRunnable(Task task) {
+        return () -> {
+            log.info("---动态定时任务运行---");
+            try {
+                log.debug("此时时间==>" + LocalDateTime.now());
+                log.debug("task中设定的时间==>" + task);
+                Thread.sleep(10);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            log.info("---end--------");
+        };
+    }*/
+
+    /**
+     * 停止任务
+     *
+     * @param name
+     * @return
+     */
+    public boolean stop(String name) {
+        if (null == taskMap.get(name)) {
+            return false;
+        }
+        ScheduledFuture<?> scheduledFuture = taskMap.get(name);
+        scheduledFuture.cancel(true);
+        taskMap.remove(name);
+        taskList.remove(name);
+        return true;
+    }
+}

miniapp/src/main/java/com/kym/miniapp/jobs/OfflineNotice.java

@@ -0,0 +1,28 @@
+package com.kym.miniapp.jobs;
+
+import com.kym.service.admin.MonitorLogService;
+import org.springframework.stereotype.Component;
+
+/**
+ * 离线通知
+ *
+ * @author skyline
+ * @date 2023-08-29 15:45
+ */
+@Component
+public class OfflineNotice extends Task {
+
+    private final MonitorLogService monitorLogService;
+
+
+    public OfflineNotice(MonitorLogService monitorLogService) {
+        this.monitorLogService = monitorLogService;
+    }
+
+    @Override
+    public void run() {
+        // TODO 查询该设备在线情况，恢复在线则修改记录状态为已恢复，依旧离线则告警通知，修改状态为以告警
+
+
+    }
+}

miniapp/src/main/java/com/kym/miniapp/jobs/Task.java

@@ -0,0 +1,26 @@
+package com.kym.miniapp.jobs;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+import java.time.LocalDateTime;
+
+/**
+ * @author skyline
+ * @description
+ * @date 2023-08-29 14:15
+ */
+@Data
+@Accessors(chain = true) // 方便链式编写
+public abstract class Task implements Runnable {
+    /**
+     * 动态任务名称
+     */
+    private String name;
+
+    /**
+     * 设定动态任务开始时间
+     */
+    private LocalDateTime startTime;
+
+}

miniapp/src/main/resources/mail.setting

@@ -0,0 +1,8 @@
+# 发件人
+from = system@kuaiyuman.cn
+# 用户名，默认为发件人邮箱前缀
+user = system
+# 密码（注意，某些邮箱需要为SMTP服务单独设置授权码，详情查看相关帮助）
+pass = Uocj6qhGXeaqUAUC
+# 使用SSL安全连接
+sslEnable = true

service/src/main/java/com/kym/service/admin/MonitorLogService.java

@@ -0,0 +1,16 @@
+package com.kym.service.admin;
+
+import com.kym.entity.admin.MonitorLog;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 设备监控日志表 服务类
+ * </p>
+ *
+ * @author skyline
+ * @since 2023-08-29
+ */
+public interface MonitorLogService extends IService<MonitorLog> {
+
+}

service/src/main/java/com/kym/service/admin/impl/MonitorLogServiceImpl.java

@@ -0,0 +1,20 @@
+package com.kym.service.admin.impl;
+
+import com.kym.entity.admin.MonitorLog;
+import com.kym.mapper.admin.MonitorLogMapper;
+import com.kym.service.admin.MonitorLogService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 设备监控日志表 服务实现类
+ * </p>
+ *
+ * @author skyline
+ * @since 2023-08-29
+ */
+@Service
+public class MonitorLogServiceImpl extends ServiceImpl<MonitorLogMapper, MonitorLog> implements MonitorLogService {
+
+}

service/src/main/java/com/kym/service/enplus/impl/EnNotifyServiceImpl.java

@@ -1,22 +1,29 @@
 package com.kym.service.enplus.impl;
 
+import cn.hutool.extra.mail.MailUtil;
 import com.alibaba.fastjson2.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
+import com.kym.entity.admin.MonitorLog;
+import com.kym.entity.common.RedisKeys;
+import com.kym.entity.enplus.EnConnectorStatusInfo;
 import com.kym.entity.miniapp.ChargeOrder;
 import com.kym.entity.miniapp.WalletDetail;
+import com.kym.service.admin.MonitorLogService;
 import com.kym.service.enplus.EnNotifyService;
 import com.kym.service.enplus.EnPlusService;
 import com.kym.service.miniapp.AccountService;
 import com.kym.service.miniapp.ChargeOrderService;
 import com.kym.service.miniapp.WalletDetailService;
+import com.kym.service.utils.KymCache;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author skyline
@@ -28,17 +35,29 @@ import java.time.format.DateTimeFormatter;
 public class EnNotifyServiceImpl implements EnNotifyService {
     private static final Logger LOGGER = LoggerFactory.getLogger(EnNotifyServiceImpl.class);
 
-    @Autowired
-    private EnPlusService enPlusService;
+    private final EnPlusService enPlusService;
 
-    @Autowired
-    private ChargeOrderService chargeOrderService;
+    private final ChargeOrderService chargeOrderService;
 
-    @Autowired
-    private AccountService accountService;
+    private final AccountService accountService;
 
-    @Autowired
-    private WalletDetailService walletDetailService;
+    private final WalletDetailService walletDetailService;
+
+    private final MonitorLogService monitorLogService;
+
+    private final KymCache kymCache;
+
+    private final RedisTemplate<String, String> redisTemplate;
+
+    public EnNotifyServiceImpl(EnPlusService enPlusService, ChargeOrderService chargeOrderService, AccountService accountService, WalletDetailService walletDetailService, MonitorLogService monitorLogService, KymCache kymCache, RedisTemplate<String, String> redisTemplate) {
+        this.enPlusService = enPlusService;
+        this.chargeOrderService = chargeOrderService;
+        this.accountService = accountService;
+        this.walletDetailService = walletDetailService;
+        this.monitorLogService = monitorLogService;
+        this.kymCache = kymCache;
+        this.redisTemplate = redisTemplate;
+    }
 
     /**
      * EN+ 充电站设备状态变化推送
@@ -50,7 +69,32 @@ public class EnNotifyServiceImpl implements EnNotifyService {
     public String handleNotificationStationStatus(JSONObject json) {
         var data = enPlusService.signValidation(json);
         LOGGER.info("【EN+推送】收到充电桩设备状态变化推送：\n{}，解密数据：\\n{}\"", json, data);
-        // TODO: 2023-08-21 更新数据库中设备的状态，如有设备离线，发送通知
+        // TODO: 2023-08-21 更新数据库中设备的状态
+        var connectorStatusInfo = JSONObject.parseObject(data).getJSONObject("ConnectorStatusInfo").toJavaObject(EnConnectorStatusInfo.class);
+        if (connectorStatusInfo.getStatus() == 0) {
+            // 如果设备离线，则存入redis，有效期24h，5分钟之内如果收到该设备上线的推送，则不发送通知并删除redis记录，否则发送通知
+            var monitorLog = new MonitorLog()
+                    .setStationId(kymCache.getStationId(connectorStatusInfo.getConnectorId()))
+                    .setSn(connectorStatusInfo.getConnectorId())
+                    .setOfflineTime(LocalDateTime.now()).setType(2)
+                    .setOfflineStatus(connectorStatusInfo.getStatus());
+            monitorLogService.save(monitorLog);
+            redisTemplate.opsForValue().set(RedisKeys.OFFLINE.concat(connectorStatusInfo.getConnectorId()), "", 1, TimeUnit.DAYS);
+            MailUtil.send("skyline@kuaiyuman.cn", "通知", "站点:%s,设备%s离线".formatted(monitorLog.getStationId(), monitorLog.getSn()), false);
+        } else {
+            // 查询redis是否有记录，有就删除并更新数据库恢复时间
+            var exist = redisTemplate.hasKey(RedisKeys.OFFLINE.concat(connectorStatusInfo.getConnectorId()));
+            if (Boolean.TRUE.equals(exist)) {
+                monitorLogService.lambdaUpdate()
+                        .eq(MonitorLog::getSn, connectorStatusInfo.getConnectorId())
+                        .eq(MonitorLog::getIsRecover, 0) // 未恢复的记录
+                        .set(MonitorLog::getRecoverTime, LocalDateTime.now())
+                        .set(MonitorLog::getIsRecover, 1) // 设置为已恢复
+                        .update();
+                redisTemplate.delete(RedisKeys.OFFLINE.concat(connectorStatusInfo.getConnectorId()));
+            }
+        }
+
         return """
                 {
                     "Status":%d

service/src/main/java/com/kym/service/enplus/impl/EnPlusServiceImpl.java

@@ -10,6 +10,7 @@ import com.kym.common.enums.EnPlusApi;
 import com.kym.common.exception.BusinessException;
 import com.kym.common.exception.EnPushException;
 import com.kym.common.utils.AESUtil;
+import com.kym.entity.common.RedisKeys;
 import com.kym.entity.enplus.EnRespQueryToken;
 import com.kym.entity.enplus.response.EnResponse;
 import com.kym.service.enplus.EnPlusService;
@@ -120,7 +121,7 @@ public class EnPlusServiceImpl implements EnPlusService {
             var enRespQueryToken = JSONObject.parseObject(AESUtil.decrypt(enResponse.getData()), EnRespQueryToken.class);
             LOGGER.debug("EN+接口AccessToken:{}", enRespQueryToken.toString());
             // 缓存token,有效期7天
-            redisTemplate.opsForValue().set("EN_PLUS_TOKEN", enRespQueryToken.getAccessToken(), enRespQueryToken.getTokenAvailableTime(), TimeUnit.SECONDS);
+            redisTemplate.opsForValue().set(RedisKeys.EN_PLUS_TOKEN, enRespQueryToken.getAccessToken(), enRespQueryToken.getTokenAvailableTime(), TimeUnit.SECONDS);
             return enRespQueryToken.getAccessToken();
         } else {
             // 记录错误码，返回错误信息

service/src/main/java/com/kym/service/utils/KymCache.java

@@ -19,11 +19,20 @@ public class KymCache {
 
     private static Map<String, String> SHORT_ID_MAPPING;
 
+    private static Map<String, String> CONNECTOR_STATION_MAPPING;
+
     public KymCache(EquipmentRelationService relationService) {
         SHORT_ID_MAPPING = relationService.list().stream().collect(Collectors.toMap(EquipmentRelation::getEquipmentId, EquipmentRelation::getShortId));
+        CONNECTOR_STATION_MAPPING = relationService.list().stream().collect(Collectors.toMap(EquipmentRelation::getConnectorId, EquipmentRelation::getStationId));
+    }
+
+    public String getShortId(String equipmentId) {
+        return SHORT_ID_MAPPING.get(equipmentId);
     }
 
-    public String getShortId(String key) {
-        return SHORT_ID_MAPPING.get(key);
+    public String getStationId(String connectorId) {
+        return CONNECTOR_STATION_MAPPING.get(connectorId);
     }
+
+
 }