救援新功能

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/constant/UserConstants.java

@@ -80,4 +80,7 @@ public class UserConstants
 
     /** 救援司机 */
     public static final String RESCUE_DRIVER = "jysj";
+
+    /** 救援定时任务时间 */
+    public static final Integer RESCUE_TIME = 3 * 60;
 }

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/rescue/domain/RescueInfo.java

@@ -184,4 +184,7 @@ public class RescueInfo extends TenantBaseDO
     @TableField(exist = false)
     private String rescueStartMonth;
 
+    @TableField(exist = false)
+    private Long[] roadIds;
+
 }

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/rescue/dto/TaskDto.java

@@ -0,0 +1,18 @@
+package cn.iocoder.yudao.module.rescue.dto;
+
+import lombok.Data;
+
+/**
+ * 用于创建定时任务
+ * @author 小李
+ * @date 18:41 2024/8/26
+**/
+@Data
+public class TaskDto {
+
+    // 谁发起的订单
+    private Long manageId;
+
+    // 发起的订单
+    private Long rescueInfoId;
+}

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/rescue/service/impl/DriverInfoServiceImpl.java

@@ -127,33 +127,46 @@ public class DriverInfoServiceImpl extends ServiceImpl<DriverInfoMapper, DriverI
      **/
     @Override
     public void syncData(){
-        List<DriverInfo> driverInfos = baseMapper.selectList(new QueryWrapper<>());
+//        List<DriverInfo> driverInfos = baseMapper.selectList(new QueryWrapper<>());
-        List<CompanyStaff> companyStaffs = new ArrayList<>();
+//        List<CompanyStaff> companyStaffs = new ArrayList<>();
-        driverInfos.stream().forEach(item -> {
+//        driverInfos.stream().forEach(item -> {
-            AdminUserRespDTO user = adminUserApi.getUser(item.getUserId());
+//            AdminUserRespDTO user = adminUserApi.getUser(item.getUserId());
-            if (ObjectUtil.isNotEmpty(user)){
+//            if (ObjectUtil.isNotEmpty(user)){
-                long count = staffService.count(new LambdaQueryWrapper<CompanyStaff>().eq(CompanyStaff::getUserId, user.getId()));
+//                long count = staffService.count(new LambdaQueryWrapper<CompanyStaff>().eq(CompanyStaff::getUserId, user.getId()));
-                if (count == 0){
+//                if (count == 0){
-                    CompanyStaff staff = new CompanyStaff();
+//                    CompanyStaff staff = new CompanyStaff();
-                    staff.setUserId(user.getId());
+//                    staff.setUserId(user.getId());
-                    staff.setDeptId(user.getDeptId());
+//                    staff.setDeptId(user.getDeptId());
-                    DeptRespDTO dept = deptApi.getDept(user.getDeptId());
+//                    DeptRespDTO dept = deptApi.getDept(user.getDeptId());
-                    staff.setCorpId(dept.getCorpId());
+//                    staff.setCorpId(dept.getCorpId());
-                    String uniqueCode = uniqueCodeService.createUniqueCode();
+//                    String uniqueCode = uniqueCodeService.createUniqueCode();
-                    if (!ObjectUtil.isNotEmpty(uniqueCode)) {
+//                    if (!ObjectUtil.isNotEmpty(uniqueCode)) {
-                        throw exception(CommonErrorCodeConstants.UNIQUE_CODE_CREATE_REPEAT);
+//                        throw exception(CommonErrorCodeConstants.UNIQUE_CODE_CREATE_REPEAT);
-                    }
+//                    }
-                    staff.setUniqueCode(uniqueCode);
+//                    staff.setUniqueCode(uniqueCode);
-
+//
-                    staff.setName(user.getNickname());
+//                    staff.setName(user.getNickname());
-                    staff.setTel(item.getPhonenumber());
+//                    staff.setTel(item.getPhonenumber());
-                    staff.setSex(String.valueOf(user.getSex()));
+//                    staff.setSex(String.valueOf(user.getSex()));
-                    companyStaffs.add(staff);
+//                    companyStaffs.add(staff);
-                }
+//                }
-            }
+//            }
-        });
+//        });
-        if (ObjectUtil.isNotEmpty(companyStaffs)){
+//        if (ObjectUtil.isNotEmpty(companyStaffs)){
-            staffService.saveBatch(companyStaffs);
+//            staffService.saveBatch(companyStaffs);
+//        }
+        AdminUserRespDTO user = adminUserApi.getUser(708L);
+        CompanyStaff staff = new CompanyStaff();
+        staff.setUserId(user.getId());
+        staff.setDeptId(100L);
+        staff.setWorkNo("la0000");
+        staff.setName(user.getNickname());
+        staff.setTel(user.getMobile());
+        staff.setSex(String.valueOf(user.getSex()));
+        String uniqueCode = uniqueCodeService.createUniqueCode();
+        if (ObjectUtil.isNotEmpty(uniqueCode)){
+            staff.setUniqueCode(uniqueCode);
         }
-    };
+        staffService.save(staff);
+    }
 }

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/rescue/service/impl/RescueDriverInfoServiceImpl.java

@@ -4,7 +4,9 @@ import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.CoordinateUtil;
 import cn.hutool.http.HttpUtil;
 import cn.iocoder.yudao.module.rescue.domain.*;
+import cn.iocoder.yudao.module.rescue.dto.TaskDto;
 import cn.iocoder.yudao.module.rescue.mapper.RescueDriverInfoMapper;
+import cn.iocoder.yudao.module.rescue.utils.RedissonDelayQueue;
 import cn.iocoder.yudao.module.system.api.dict.DictDataApi;
 import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
 import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
@@ -31,7 +33,8 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 import static cn.hutool.core.util.CoordinateUtil.wgs84ToGcj02;
-
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
+import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception0;
 
 
 /**
@@ -68,6 +71,8 @@ public class RescueDriverInfoServiceImpl extends ServiceImpl<RescueDriverInfoMap
     private WechatPayRequest wechatPayRequest;
     @Autowired
     private WechatPayConfig wechatPayConfig;
+    @Resource
+    private RedissonDelayQueue redissonDelayQueue;
 
     public static String Redis_Driver_Key = "Rescue:Driver:";
     public static String Redis_Driver_Position_Key = "DriverPosition:";
@@ -294,7 +299,7 @@ public class RescueDriverInfoServiceImpl extends ServiceImpl<RescueDriverInfoMap
             lock.lock();
             try {
                 if (!ObjectUtils.isEmpty(rescueInfo.getDriverId())) {
-                    throw new Exception("该订单已被抢走");
+                    throw exception0(500, "订单已被抢走");
                 }
                 DriverInfo driverInfo = driverInfoService.getById(rescueDriverInfo.getDriverId());
 
@@ -312,7 +317,7 @@ public class RescueDriverInfoServiceImpl extends ServiceImpl<RescueDriverInfoMap
                     rescueInfo.setDriverPhoneNum(driverInfo.getPhonenumber());
                     rescueInfo.setDriverCarNum(carInfo.getRescueCarNum());
                 } else {
-                    throw new Exception("请联系管理员维护车辆信息开始接单");
+                    throw exception0(500,"请联系管理员维护车辆信息开始接单");
                 }
                 //状态修改为救援中
                 rescueInfo.setRescueStatus("3");
@@ -361,6 +366,9 @@ public class RescueDriverInfoServiceImpl extends ServiceImpl<RescueDriverInfoMap
                     map.put("rescueIds", rescueInfo.getId());
                     redisCache.setCacheMap(redisKey, map);
                 }
+
+                // 删除定时任务
+                redissonDelayQueue.removeAllTasks(rescueInfo.getId());
             } catch (Exception e) {
                 log.error(e.getMessage());
                 throw new Exception("出现异常");

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/rescue/service/impl/RescueInfoServiceImpl.java

@@ -2,17 +2,21 @@ package cn.iocoder.yudao.module.rescue.service.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.ObjectUtil;
 import cn.iocoder.yudao.framework.security.core.LoginUser;
 import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils;
 import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO;
 import cn.iocoder.yudao.module.appBase.domain.SysAnnouncement;
 import cn.iocoder.yudao.module.appBase.service.ISysAnnouncementService;
+import cn.iocoder.yudao.module.constant.UserConstants;
 import cn.iocoder.yudao.module.rescue.domain.*;
 import cn.iocoder.yudao.module.rescue.dto.DriverInfo2Dto;
 import cn.iocoder.yudao.module.rescue.dto.DriverInfoDto;
+import cn.iocoder.yudao.module.rescue.dto.TaskDto;
 import cn.iocoder.yudao.module.rescue.mapper.RescueInfoMapper;
 import cn.iocoder.yudao.module.rescue.service.*;
 import cn.iocoder.yudao.module.rescue.utils.RedisUtil;
+import cn.iocoder.yudao.module.rescue.utils.RedissonDelayQueue;
 import cn.iocoder.yudao.module.rescue.utils.StringUtils;
 import cn.iocoder.yudao.module.rescue.vo.MoneyManagement;
 import cn.iocoder.yudao.module.staff.service.CompanyStaffService;
@@ -25,6 +29,7 @@ import cn.iocoder.yudao.module.system.api.permission.dto.RoleReqDTO;
 import cn.iocoder.yudao.module.system.api.user.AdminUserApi;
 import cn.iocoder.yudao.module.system.api.user.dto.AdminUserRespDTO;
 import cn.iocoder.yudao.module.system.api.user.dto.UserDTO;
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
@@ -92,6 +97,10 @@ public class RescueInfoServiceImpl extends ServiceImpl<RescueInfoMapper, RescueI
     private CompanyStaffService staffService;
     @Resource
     private RoleApi roleApi;
+    @Resource
+    private RedissonDelayQueue redissonDelayQueue;
+    @Resource
+    private IRescueDictStaffService rescueDictStaffService;
 
 
     /**
@@ -353,6 +362,8 @@ public class RescueInfoServiceImpl extends ServiceImpl<RescueInfoMapper, RescueI
         sysAnnouncement.setContent("有新的救援订单请立即处理");
         announcementService.insertSysAnnouncements(sysAnnouncement);
 
+        // 删除延时消息
+        redissonDelayQueue.removeAllTasks(rescueId);
     }
 
 
@@ -365,7 +376,7 @@ public class RescueInfoServiceImpl extends ServiceImpl<RescueInfoMapper, RescueI
     @Override
     public void insertRescueInfo(RescueInfo rescueInfo) {
         // 获取当前的用户
-        LoginUser loginUser = getLoginUser();
+        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
 
         rescueInfo.setUserId(loginUser.getId());
         rescueInfo.setRescueTime(new Date());
@@ -380,12 +391,29 @@ public class RescueInfoServiceImpl extends ServiceImpl<RescueInfoMapper, RescueI
             rescueInfo.setRescueTime(new Date());
             baseMapper.insert(rescueInfo);
             detailService.save(new RescueInfoDetail(rescueInfo.getId(), "0", "救援发起", "救援发起"));
-            // TODO: 2023/12/24  暂时取消自动指派司机
+
-//            Thread thread = new Thread(() -> {
+            // 自动通知对应路段司机
-//                //自动派单
+            List<RescueDictStaff> list = rescueDictStaffService.list(new LambdaQueryWrapper<RescueDictStaff>().in(RescueDictStaff::getDictId, rescueInfo.getRoadIds()));
-//                this.appointDriverByBusiness(deptId,rescueInfo.getId());
+            if (CollectionUtil.isNotEmpty(list)){
-//            });
+                Set<Long> driverIds = list.stream()
-//            thread.start();
+                        .map(RescueDictStaff::getDriverIds)
+                        .flatMap(item -> Arrays.stream(item.split(",")))
+                        .map(Long::parseLong)
+                        .collect(Collectors.toSet());
+                driverIds.forEach(item -> {
+                    SysAnnouncement sysAnnouncement = new SysAnnouncement();
+                    sysAnnouncement.setType("救援信息");
+                    sysAnnouncement.setToUserIds(Collections.singletonList(item));
+                    sysAnnouncement.setTitle("有新的救援订单请立即处理");
+                    sysAnnouncement.setContent("有新的救援订单请立即处理");
+                    announcementService.insertSysAnnouncements(sysAnnouncement);
+                });
+            }
+            // 新增延迟消息，用于三分钟后没人接单通知
+            TaskDto taskDto = new TaskDto();
+            taskDto.setManageId(loginUser.getId());
+            taskDto.setRescueInfoId(rescueInfo.getId());
+            redissonDelayQueue.offerTask(JSON.toJSONString(taskDto), UserConstants.RESCUE_TIME);
         }
 
     }
@@ -401,6 +429,10 @@ public class RescueInfoServiceImpl extends ServiceImpl<RescueInfoMapper, RescueI
         if (rescueInfo.getRescueStatus().compareTo("9") > 0) {
             rescueInfo.setRescueStatus(null);
         }
+        // 取消订单时删除定时任务
+        if (rescueInfo.getRescueStatus().equals("0")){
+            redissonDelayQueue.removeAllTasks(rescueInfo.getId());
+        }
         return baseMapper.updateById(rescueInfo);
     }
 

dl-module-rescue/src/main/java/cn/iocoder/yudao/module/rescue/utils/RedissonDelayQueue.java

@@ -0,0 +1,178 @@
+package cn.iocoder.yudao.module.rescue.utils;
+
+import cn.iocoder.yudao.module.appBase.domain.SysAnnouncement;
+import cn.iocoder.yudao.module.appBase.service.ISysAnnouncementService;
+import cn.iocoder.yudao.module.constant.UserConstants;
+import cn.iocoder.yudao.module.rescue.dto.TaskDto;
+import com.alibaba.fastjson.JSON;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RBlockingQueue;
+import org.redisson.api.RDelayedQueue;
+import org.redisson.api.RMap;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.util.Collections;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Redisson 延迟队列组件，用于处理带有延迟时间的任务。
+ *
+ * @author 小李
+ * @date 16:37 2024/8/26
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RedissonDelayQueue {
+
+    // 注入 Redisson 提供的延迟队列接口
+    @Resource
+    private RDelayedQueue<String> delayedQueue;
+
+    // 注入 Redisson 提供的阻塞队列接口
+    @Resource
+    private RBlockingQueue<String> blockingQueue;
+
+    @Resource
+    private ISysAnnouncementService announcementService;
+
+    // 任务索引，用于快速查找任务
+    @Resource
+    private RMap<Long, String> taskIndex;
+
+    /**
+     * 初始化方法，在 Spring 容器加载后自动执行。
+     * 创建一个固定大小的线程池来消费阻塞队列中的任务。
+     */
+    @PostConstruct
+    public void init() {
+        // 创建一个单线程的线程池
+        ExecutorService executorService = Executors.newFixedThreadPool(1);
+
+        // 提交一个无限循环的任务到线程池，用于持续消费队列中的任务
+        executorService.submit(() -> {
+            while (true) {
+                try {
+                    // 从阻塞队列中获取一个任务，如果队列为空，则此方法会一直阻塞直到有任务可用
+                    String task = blockingQueue.take();
+
+                    // 记录接收到的任务信息
+                    log.info("定时任务:{}", task);
+
+                    // 通知发起者
+                    TaskDto taskDto = JSON.parseObject(task, TaskDto.class);
+                    SysAnnouncement sysAnnouncement = new SysAnnouncement();
+                    sysAnnouncement.setType("救援超时信息");
+                    sysAnnouncement.setToUserIds(Collections.singletonList(taskDto.getManageId()));
+                    sysAnnouncement.setTitle("有救援订单超时请立即处理");
+                    sysAnnouncement.setContent("有救援订单超时请立即处理");
+                    announcementService.insertSysAnnouncements(sysAnnouncement);
+
+                    // TODO 还要通知总调度
+
+                    // 在新创建定时任务
+                    offerTask(task, UserConstants.RESCUE_TIME);
+                } catch (Exception e) {
+                    // 记录并打印异常信息
+                    log.error("发生错误", e);
+                }
+            }
+        });
+    }
+
+    /**
+     * 向延迟队列中添加一个任务，并设置延迟时间。
+     *
+     * @param task 要添加的任务
+     * @param seconds 任务的延迟时间，单位为秒
+     */
+    public void offerTask(String task, long seconds) {
+        // 记录添加任务的日志信息
+        log.info("添加定时任务:{},过期时间为:{}s", task, seconds);
+
+        // 从任务中提取 rescueInfoId
+        TaskDto taskDto = JSON.parseObject(task, TaskDto.class);
+        Long rescueInfoId = taskDto.getRescueInfoId();
+
+        // 存储任务索引
+        taskIndex.put(rescueInfoId, task);
+
+        // 向延迟队列添加任务，并设置延迟时间
+        delayedQueue.offer(task, seconds, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 从延迟队列中移除指定的任务。
+     *
+     * @param rescueInfoId 要移除的任务
+     * @return 移除是否成功
+     */
+    private boolean removeDelayedTask(Long rescueInfoId) {
+        try {
+            // 从任务索引中查找任务
+            String task = taskIndex.get(rescueInfoId);
+
+            // 如果找到了任务，则尝试从延迟队列中移除任务
+            if (task != null) {
+                boolean removed = delayedQueue.remove(task);
+                if (removed) {
+                    // 从任务索引中移除任务
+                    taskIndex.remove(rescueInfoId);
+                }
+                return removed;
+            }
+        } catch (Exception e) {
+            log.error("延迟队列删除失败", e);
+        }
+        return false;
+    }
+
+    /**
+     * 从阻塞队列中移除指定的任务。
+     *
+     * @param rescueInfoId 要移除的任务
+     * @return 移除是否成功
+     */
+    private boolean removeBlockingTask(Long rescueInfoId) {
+        try {
+            // 从任务索引中查找任务
+            String task = taskIndex.get(rescueInfoId);
+
+            // 如果找到了任务，则尝试从阻塞队列中移除任务
+            if (task != null) {
+                boolean removed = blockingQueue.remove(task);
+                if (removed) {
+                    // 从任务索引中移除任务
+                    taskIndex.remove(rescueInfoId);
+                }
+                return removed;
+            }
+        } catch (Exception e) {
+            log.error("阻塞队列删除失败", e);
+        }
+        return false;
+    }
+
+    /**
+     * 尝试同时从延迟队列和阻塞队列中移除指定的任务。
+     * 如果任务在任何一个队列中存在，则尝试移除。
+     *
+     * @param rescueInfoId 要移除的任务
+     * @return 是否至少在一个队列中移除成功
+     */
+    public boolean removeAllTasks(Long rescueInfoId) {
+        // 尝试从延迟队列中移除任务
+        boolean removedFromDelayed = removeDelayedTask(rescueInfoId);
+
+        // 尝试从阻塞队列中移除任务
+        boolean removedFromBlocking = removeBlockingTask(rescueInfoId);
+
+        // 如果任何一个队列中移除成功，则返回 true
+        return removedFromDelayed || removedFromBlocking;
+    }
+}