预约充电

common/src/main/java/com/kym/common/constant/ResponseEnum.java

@@ -47,6 +47,7 @@ public enum ResponseEnum implements BusinessExceptionAssert {
     EN_PLUS_EQUIP_OFFLINE(20011, "设备离线"),
     EN_PLUS_EQUIP_STOP_FAIL(20012, "设备停止充电失败"),
     EN_PLUS_EQUIP_EXIST_ORDER_UNFINISHED(20013, "设备存在未完成的订单"),
+    ORDER_IN_BOOKING(20014, "用户有预约中的订单"),
 
 
     // EN+

entity/src/main/java/com/kym/entity/miniapp/ChargeOrder.java

@@ -5,6 +5,7 @@ import com.fasterxml.jackson.annotation.JsonFormat;
 import com.kym.entity.BaseEntity;
 import lombok.Getter;
 import lombok.Setter;
+import lombok.experimental.Accessors;
 
 import java.io.Serializable;
 import java.time.LocalDateTime;
@@ -20,6 +21,7 @@ import java.time.LocalDateTime;
 @Getter
 @Setter
 @TableName("t_charge_order")
+@Accessors(chain = true)
 public class ChargeOrder extends BaseEntity implements Serializable {
 
     private static final long serialVersionUID = 1L;

mapper/pom.xml

@@ -47,7 +47,7 @@
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>druid-spring-boot-starter</artifactId>
-            <version>1.2.18</version>
+            <version>1.2.20</version>
         </dependency>
 
         <dependency>

miniapp/src/main/java/com/kym/miniapp/controller/ChargerController.java

@@ -1,5 +1,6 @@
 package com.kym.miniapp.controller;
 
+import cn.dev33.satoken.stp.StpUtil;
 import com.alibaba.fastjson2.JSONObject;
 import com.kym.common.R;
 import com.kym.common.annotation.ApiLog;
@@ -12,6 +13,7 @@ import com.kym.service.enplus.EnPlusService;
 import com.kym.service.miniapp.ChargeOrderService;
 import com.kym.service.miniapp.ChargeService;
 import lombok.SneakyThrows;
+import org.springframework.format.annotation.DateTimeFormat;
 import org.springframework.web.bind.annotation.*;
 
 import java.time.LocalDateTime;
@@ -82,9 +84,22 @@ public class ChargerController {
     @ApiLog("启动充电")
     @GetMapping({"/startCharge/{connectorId}", "/startCharge/{connectorId}"})
     R startCharge(@PathVariable("connectorId") String connectorId,
-                  @RequestParam(value = "isBooking", required = false) Boolean isBooking,
-                  @RequestParam(value = "startTime", required = false) LocalDateTime startTime) {
-        return R.success(chargeService.queryStartCharge(connectorId, isBooking,startTime));
+                  @RequestParam(value = "isBooking", defaultValue = "false") Boolean isBooking,
+                  @RequestParam(value = "startTime", required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") LocalDateTime startTime) {
+        var userId = StpUtil.getLoginIdAsLong();
+        return R.success(chargeService.queryStartCharge(userId, connectorId, isBooking, startTime));
+    }
+
+    @ApiLog("修改预约充电时间")
+    @GetMapping("/modifyBookingTime/{startChargeSeq}")
+    R<?> modifyBookingTime(@PathVariable("startChargeSeq") String startChargeSeq, @RequestParam LocalDateTime startTime) {
+        return R.success(chargeService.modifyBookingTime(startChargeSeq, startTime));
+    }
+
+    @ApiLog("预约改立即充电")
+    @GetMapping("immediatelyCharge/{connectorId}")
+    R<?> immediatelyCharge(@PathVariable("connectorId") String connectorId) {
+        return R.success(chargeService.immediatelyCharge(connectorId));
     }
 
     @ApiLog("停止充电")

service/src/main/java/com/kym/service/miniapp/ChargeService.java

@@ -1,5 +1,6 @@
 package com.kym.service.miniapp;
 
+import com.baomidou.dynamic.datasource.annotation.DS;
 import com.kym.entity.enplus.response.EnBusinessPolicy;
 import com.kym.entity.miniapp.ChargeOrder;
 
@@ -13,11 +14,18 @@ import java.util.Map;
  */
 public interface ChargeService {
 
-    Map queryStartCharge(String connectorId, Boolean isBooking, LocalDateTime startTime);
+    Map<String, String> queryStartCharge(Long userId, String connectorId, Boolean isBooking, LocalDateTime startTime);
+
+    @DS("db-admin")
+    void updateEquipmentStatus(String connectorId);
 
     ChargeOrder queryEquipChargeStatus();
 
     void queryStopCharge(String connectorId);
 
     EnBusinessPolicy queryEquipBusinessPolicy(String connectorId);
+
+    Map<String, String> immediatelyCharge(String connectorId);
+
+    Object modifyBookingTime(String startChargeSeq, LocalDateTime startTime);
 }

service/src/main/java/com/kym/service/miniapp/impl/ChargeServiceImpl.java

@@ -2,6 +2,8 @@ package com.kym.service.miniapp.impl;
 
 import cn.dev33.satoken.stp.StpUtil;
 import com.baomidou.dynamic.datasource.annotation.DS;
+import com.baomidou.dynamic.datasource.annotation.DSTransactional;
+import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
 import com.kym.common.config.EnPlusConfig;
 import com.kym.common.constant.ResponseEnum;
 import com.kym.common.exception.BusinessException;
@@ -15,16 +17,14 @@ import com.kym.service.enplus.EnPlusService;
 import com.kym.service.miniapp.AccountService;
 import com.kym.service.miniapp.ChargeOrderService;
 import com.kym.service.miniapp.ChargeService;
-import com.kym.service.queue.DelayedItem;
+import com.kym.service.miniapp.DelayService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
 import java.util.Map;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 /**
  * @author skyline
@@ -34,6 +34,7 @@ import java.util.concurrent.Executors;
 @Service
 @DS("db-miniapp")
 public class ChargeServiceImpl implements ChargeService {
+
     private static final Logger LOGGER = LoggerFactory.getLogger(ChargeServiceImpl.class);
 
     private final EquipmentRelationService equipmentRelationService;
@@ -48,22 +49,55 @@ public class ChargeServiceImpl implements ChargeService {
 
     private final EnPlusConfig enPlusConfig;
 
-    /**
-     * 预约充电队列
-     */
-    DelayQueue<DelayedItem> delayQueue = new DelayQueue<>();
-
-    private ExecutorService executorService = Executors.newFixedThreadPool(2);
+    private final DelayService<ChargeOrder> delayService;
 
-    public ChargeServiceImpl(EquipmentRelationService equipmentRelationService, EquipmentInfoService equipmentInfoService, ChargeOrderService chargeOrderService, AccountService accountService, EnPlusService enPlusService, EnPlusConfig enPlusConfig) {
+    public ChargeServiceImpl(EquipmentRelationService equipmentRelationService, EquipmentInfoService equipmentInfoService,
+                             ChargeOrderService chargeOrderService, AccountService accountService,
+                             EnPlusService enPlusService, EnPlusConfig enPlusConfig, @Lazy DelayService<ChargeOrder> delayService) {
         this.equipmentRelationService = equipmentRelationService;
         this.equipmentInfoService = equipmentInfoService;
         this.chargeOrderService = chargeOrderService;
         this.accountService = accountService;
         this.enPlusService = enPlusService;
         this.enPlusConfig = enPlusConfig;
+        this.delayService = delayService;
     }
 
+
+    /**
+     * 预约充电改立即充电
+     *
+     * @param connectorId
+     * @return
+     */
+    @Override
+    public Map<String, String> immediatelyCharge(String connectorId) {
+        var userId = StpUtil.getLoginIdAsLong();
+        // 修改订单状态，取消预约
+        chargeOrderService.lambdaUpdate()
+                .set(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_未知)
+                .eq(ChargeOrder::getConnectorId, connectorId)
+                .eq(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_预约中)
+                .update();
+        return queryStartCharge(userId, connectorId, false, null);
+    }
+
+    /**
+     * 修改预约充电时间
+     *
+     * @param startChargeSeq
+     * @param startTime
+     * @return
+     */
+    @Override
+    public Object modifyBookingTime(String startChargeSeq, LocalDateTime startTime) {
+        // 预约充电队列更新
+        var chargeOrder = chargeOrderService.getChargingOrderByStartChargeSeq(startChargeSeq);
+        delayService.removeToOrderDelayQueue(chargeOrder);
+        return null;
+    }
+
+
     /**
      * 启动充电
      *
@@ -73,25 +107,29 @@ public class ChargeServiceImpl implements ChargeService {
      * @return
      */
     @Override
-    public Map<String, String> queryStartCharge(String connectorId, Boolean isBooking, LocalDateTime startTime) {
-        var userId = StpUtil.getLoginIdAsLong();
-//        if (isBooking) {
-//            // 预约充电通过connectorId查询预约中的订单
-//            var bookingOrder = chargeOrderService.lambdaQuery()
-//                    .eq(ChargeOrder::getConnectorId, connectorId).eq(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_预约中).one();
-//            userId =bookingOrder.getUserId();
-//        }
+    @DSTransactional(rollbackFor = Exception.class)
+    public Map<String, String> queryStartCharge(Long userId, String connectorId, Boolean isBooking, LocalDateTime startTime) {
+        if (isBooking) {
+            // 预约充电通过connectorId查询预约中的订单
+            var bookingOrder = chargeOrderService.lambdaQuery()
+                    .eq(ChargeOrder::getConnectorId, connectorId).eq(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_预约中).one();
+            if (bookingOrder != null) {
+                LOGGER.error("用户：{}存在进行中的订单：{}", userId, bookingOrder.getStartChargeSeq());
+                throw new BusinessException(ResponseEnum.ORDER_IN_BOOKING);
+            }
+        }
         var map = getConnectorIdAndStationId(connectorId);
         connectorId = map.get("connectorId");
         var stationId = map.get("stationId");
         LOGGER.info("用户：{},设备：{}请求启动充电", userId, connectorId);
+
         // 二维码文本
         var qrCode = "";
 
         // 当前设备是是否有正在进行中的订单
         var chargeOrder = chargeOrderService.getChargingOrderByUserId(userId);
         if (chargeOrder != null) {
-            LOGGER.error("用户：{}存在进行中的订单：{}", userId, chargeOrder.getId());
+            LOGGER.error("用户：{}存在进行中的订单：{}", userId, chargeOrder.getStartChargeSeq());
             throw new BusinessException(ResponseEnum.ORDER_IN_PROGRESS);
         }
         // 查询用户余额
@@ -103,47 +141,61 @@ public class ChargeServiceImpl implements ChargeService {
         // 传递给EN+的余额要小于实际余额，防止订单超扣的情况，这里少传0.5元
         var amount = account.getBalance() - 50;
 
-        // 充电订单号/设备认证号
-        String startChargeSeq = OrderUtils.getOrderNo(enPlusConfig.getOperatorId());
+        // 是否有之前预约充电创建的订单记录，有则直接用，没有则创建
+        ChargeOrder order = chargeOrderService.lambdaQuery()
+                .eq(ChargeOrder::getConnectorId, connectorId)
+                .in(ChargeOrder::getChargeStatus, ChargeOrder.ORDER_STATUS_未知, ChargeOrder.CHARGE_STATUS_预约中)
+                .one();
 
-        // 请求设备认证
-        var equipAuth = enPlusService.queryEquipAuth(connectorId, startChargeSeq);
+        if (order == null) {
+            // 充电订单号/设备认证号
+            String startChargeSeq = OrderUtils.getOrderNo(enPlusConfig.getOperatorId());
+
+            // 组装订单数据
+            order = new ChargeOrder();
+            order.setUserId(userId);
+            order.setStationId(stationId);
+            order.setStartChargeSeq(startChargeSeq);
+            order.setConnectorId(connectorId);
+            order.setOrderStatus(ChargeOrder.ORDER_STATUS_未知);
+            chargeOrderService.save(order);
+        }
 
-        // 组装订单数据
-        var order = new ChargeOrder();
-        order.setUserId(userId);
-        order.setStationId(stationId);
-        order.setStartChargeSeq(startChargeSeq);
-        order.setConnectorId(connectorId);
-        order.setOrderStatus(ChargeOrder.ORDER_STATUS_未知);
+        // 请求设备认证
+        var equipAuth = enPlusService.queryEquipAuth(connectorId, order.getStartChargeSeq());
 
         // 如果是预约订单，则将订单放入预约充电延迟队列
         if (isBooking) {
-            order.setChargeStatus(ChargeOrder.CHARGE_STATUS_预约中);
-            chargeOrderService.save(order);
-            var flag = delayQueue.offer(new DelayedItem<ChargeOrder>(order, startTime));
+            order = order.setStartTime(startTime).setChargeStatus(ChargeOrder.CHARGE_STATUS_预约中);
+            chargeOrderService.lambdaUpdate()
+                    .set(ChargeOrder::getChargeStatus, ChargeOrder.CHARGE_STATUS_预约中)
+                    .set(ChargeOrder::getStartTime, startTime)
+                    .eq(ChargeOrder::getStartChargeSeq, order.getStartChargeSeq())
+                    .update();
+            var flag = delayService.addToDelayQueue(order.setStartTime(startTime));
             if (flag) {
                 // 修改设备状态为预约中
-                equipmentInfoService.lambdaUpdate()
-                        .set(EquipmentInfo::getServiceStatus, EquipmentInfo.SERVICE_STATUS_预约中)
-                        .eq(EquipmentInfo::getEquipmentId, connectorId.substring(0, 16))
-                        .update();
-                return Map.of("startChargeSeq", startChargeSeq);
+                // TODO: 2023-10-09 切换数据源 
+                updateEquipmentStatus(connectorId);
+                LOGGER.info("预约充电成功，用户：{}，订单号：{}", userId, order.getStartChargeSeq());
+                return Map.of("startChargeSeq", order.getStartChargeSeq());
             } else {
+                LOGGER.error("预约充电失败，用户：{}，订单号：{}", userId, order.getStartChargeSeq());
                 throw new BusinessException("充电预约失败");
             }
         }
 
-        chargeOrderService.save(order);
 
         if (equipAuth.containsKey("SuccStat") && equipAuth.getIntValue("SuccStat") == 0) {
             // 启动充电
-            var startCharge = enPlusService.queryStartCharge(startChargeSeq, connectorId, qrCode, amount);
+            var startCharge = enPlusService.queryStartCharge(order.getStartChargeSeq(), connectorId, qrCode, amount);
             if (startCharge.containsKey("SuccStat") && startCharge.getIntValue("SuccStat") == 0) {
-                // 启动成功，生成充电订单
-                order.setChargeStatus(startCharge.getIntValue("StartChargeSeqStat"));
-                chargeOrderService.save(order);
-                return Map.of("startChargeSeq", startChargeSeq);
+                // 启动成功，更新充电订单状态
+                chargeOrderService.lambdaUpdate()
+                        .set(ChargeOrder::getChargeStatus, startCharge.getIntValue("StartChargeSeqStat"))
+                        .eq(ChargeOrder::getStartChargeSeq, order.getStartChargeSeq())
+                        .update();
+                return Map.of("startChargeSeq", order.getStartChargeSeq());
             } else {
                 // 启动充电失败
                 LOGGER.error("设备启动充电失败：{}", startCharge);
@@ -179,6 +231,24 @@ public class ChargeServiceImpl implements ChargeService {
     }
 
 
+    /**
+     * 修改充电桩状态（这里@DS使用AOP实现，切换数据源需要单独抽出方法，这里@DS不生效，改为手动切换）
+     *
+     * @param connectorId
+     */
+    @Override
+    public void updateEquipmentStatus(String connectorId) {
+        // 手动切换数据源
+        DynamicDataSourceContextHolder.push("db-admin");
+        equipmentInfoService.lambdaUpdate()
+                .set(EquipmentInfo::getServiceStatus, EquipmentInfo.SERVICE_STATUS_预约中)
+                .eq(EquipmentInfo::getEquipmentId, connectorId.substring(0, 16))
+                .update();
+        DynamicDataSourceContextHolder.poll();
+
+    }
+
+
     /**
      * 请求EN+设备充电状态
      *
@@ -210,6 +280,13 @@ public class ChargeServiceImpl implements ChargeService {
         }
     }
 
+    /**
+     * 获取设备接口编号和站点编号
+     *
+     * @param connectorId 充电桩id，短编号等
+     * @return
+     */
+    @DS("db-admin")
     public Map<String, String> getConnectorIdAndStationId(String connectorId) {
         var stationId = "";
         if (connectorId.length() == 17) {
@@ -232,6 +309,7 @@ public class ChargeServiceImpl implements ChargeService {
         return Map.of("connectorId", connectorId, "stationId", stationId);
     }
 
+    @DS("db-admin")
     public String getConnectorId(String connectorId) {
         if (connectorId.length() == 17) {
             return connectorId;
@@ -251,7 +329,6 @@ public class ChargeServiceImpl implements ChargeService {
         return connectorId;
     }
 
-
     /**
      * 请求停止充电
      *
@@ -283,7 +360,6 @@ public class ChargeServiceImpl implements ChargeService {
 
     }
 
-
     /**
      * 请求设备计费策略
      *

service/src/main/java/com/kym/service/miniapp/impl/DelayServiceImpl.java

@@ -1,12 +1,15 @@
 package com.kym.service.miniapp.impl;
 
 import com.baomidou.dynamic.datasource.annotation.DS;
+import com.kym.common.exception.BusinessException;
 import com.kym.entity.miniapp.ChargeOrder;
 import com.kym.service.miniapp.ChargeOrderService;
 import com.kym.service.miniapp.ChargeService;
 import com.kym.service.miniapp.DelayService;
 import com.kym.service.queue.DelayedItem;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Service;
@@ -23,13 +26,14 @@ import java.util.concurrent.Executors;
 @Service
 @Slf4j
 @DS("db-miniapp")
+@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON) // 设置成单例
 public class DelayServiceImpl implements DelayService<ChargeOrder> {
 
+
     /**
      * 预约订单队列
      */
     private final static DelayQueue<DelayedItem<ChargeOrder>> DELAY_QUEUE = new DelayQueue<>();
-
     private final ChargeOrderService chargeOrderService;
     private final ChargeService chargeService;
     /**
@@ -55,16 +59,27 @@ public class DelayServiceImpl implements DelayService<ChargeOrder> {
 
         // 开启线程处理队列消息
         executor.execute(() -> {
+            ThreadLocal<String> threadLocal = new ThreadLocal<>();
             log.info("启动预约充电订单处理线程：{}", Thread.currentThread().getName());
             DelayedItem<ChargeOrder> delayedItem;
             while (true) {
                 try {
                     delayedItem = DELAY_QUEUE.take();
                     // 启动充电
-//                    chargeService.queryStartCharge(delayedItem.data.getConnectorId())
-                } catch (InterruptedException e) {
-                    log.error("预约充电队列take异常", e);
-                    throw new RuntimeException(e);
+                    var order = delayedItem.data;
+                    threadLocal.set(order.getStartChargeSeq());
+                    chargeService.queryStartCharge(order.getUserId(), order.getConnectorId(), false, null);
+                    log.info("预约充电启动成:用户:{}，订单号：{}，预约启动时间：{}", order.getUserId(), order.getStartChargeSeq(), order.getStartTime());
+                    // 线程休眠100ms
+                    Thread.sleep(100);
+                } catch (Exception e) {
+                    if (e instanceof InterruptedException) {
+                        log.error("预约充电队列take异常", e);
+                    } else {
+                        log.info("预约充电启动失败，订单号：{}", threadLocal.get(), e);
+                    }
+                } finally {
+                    threadLocal.remove();
                 }
             }
         });

service/src/main/java/com/kym/service/queue/DelayedItem.java

@@ -21,7 +21,7 @@ public class DelayedItem<T> implements Delayed {
     /**
      * 预约启动时间
      */
-    public final LocalDateTime startTime;
+    public LocalDateTime startTime = LocalDateTime.MIN;
 
     public DelayedItem(T data, LocalDateTime startTime) {
         this.data = data;
@@ -31,7 +31,7 @@ public class DelayedItem<T> implements Delayed {
 
     @Override
     public long getDelay(@NotNull TimeUnit unit) {
-        return unit.convert(Duration.between(startTime, LocalDateTime.now()));
+        return unit.convert(Duration.between(LocalDateTime.now(), startTime));
     }
 
     @Override