From 7a394e5bf750464ef5cdfe4034847fe354fcfe20 Mon Sep 17 00:00:00 2001
From: “mitao” <“2763622819@qq.com”>
Date: 星期一, 02 三月 2026 19:09:50 +0800
Subject: [PATCH] feat:常规任务通知定时任务
---
xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java | 117 +++++++++++++++++++++++++++++++++++++---------------------
1 files changed, 74 insertions(+), 43 deletions(-)
diff --git a/xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java b/xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java
index 6789a28..36d0b6e 100644
--- a/xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java
+++ b/xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java
@@ -41,6 +41,7 @@
import com.xiaopeng.maintenance.service.IXpTaskRuleService;
import com.xiaopeng.maintenance.service.XpTaskService;
import com.xiaopeng.maintenance.utils.UUIDUtil;
+import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
@@ -60,6 +61,8 @@
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -71,6 +74,7 @@
* @author WuGuanFengYue
* @since 2026-02-03
*/
+@Slf4j
@Service
public class XpTaskServiceImpl extends ServiceImpl<XpTaskMapper, XpTask> implements XpTaskService {
@Autowired
@@ -223,33 +227,84 @@
xpTask.setEndTime(endTime);
this.save(xpTask);
- // 发送消息通知
- List<Long> userIds = Arrays.stream(xpTask.getHandleUserId().split(",")).map(Long::valueOf).collect(Collectors.toList());
- if (CollUtil.isNotEmpty(userIds)) {
- R<List<XpOperationsUser>> xpOperationUserResult = xpOperationUserClient.listByIds(userIds);
- List<XpOperationsUser> data = xpOperationUserResult.getData();
- if (CollUtil.isNotEmpty(data)) {
+ // 异步发送消息通知
+ String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
+ sendAccidentNotificationAsync(xpTask, plateNumber, taskRule.getProcessingTimeLimit());
+ }
- String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
- // 车辆${车牌号}在${车辆位置信息}发生了紧急事故，请在${到达现场限时}内到达现场！${APP链接}
- String content = String.format("车辆%s在%s发生了紧急事故，请在%d分钟内到达现场！", plateNumber, xpTask.getVehicleLocation(), taskRule.getProcessingTimeLimit());
+ /**
+ * 异步发送事故通知
+ * @param xpTask 任务信息
+ * @param plateNumber 车牌号
+ * @param processingTimeLimit 处理时限（分钟）
+ */
+ private void sendAccidentNotificationAsync(XpTask xpTask, String plateNumber, Integer processingTimeLimit) {
+ ThreadPoolExecutor executor = new ThreadPoolExecutor(
+ 1, 1, 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>()
+ );
+
+ executor.execute(() -> {
+ try {
+ List<Long> userIds = Arrays.stream(xpTask.getHandleUserId().split(",")).map(Long::valueOf).collect(Collectors.toList());
+ if (CollUtil.isEmpty(userIds)) {
+ return;
+ }
+
+ R<List<XpOperationsUser>> xpOperationUserResult = xpOperationUserClient.listByIds(userIds);
+ List<XpOperationsUser> data = xpOperationUserResult.getData();
+ if (CollUtil.isEmpty(data)) {
+ return;
+ }
+
+ // 构建通知内容
+ String content = String.format("车辆%s在%s发生了紧急事故，请在%d分钟内到达现场！", plateNumber, xpTask.getVehicleLocation(), processingTimeLimit);
+ String bodyJson = buildAccidentPushBodyJson(content, xpTask);
+
+ // 构建通知记录（包含 notificationExt）
List<XpOperationsUserNotification> notifications = data.stream().map(item -> {
XpOperationsUserNotification notification = new XpOperationsUserNotification();
notification.setOperationsUserId(item.getId());
+ notification.setNotificationType(NotificationTypeEnum.ACCIDENT_NEW.getCode());
notification.setNotificationContent(content);
+ notification.setDataId(xpTask.getId());
notification.setReadStatus(0);
- notification.setCreateTime(LocalDateTime.now());
+ notification.setNotificationExt(bodyJson);
return notification;
}).collect(Collectors.toList());
+
+ // 批量保存通知记录
xpOperationUserNotificationClient.addNotificationBatch(notifications);
- // 发送阿里云推送
- Map<Integer, List<XpOperationsUser>> operationUserMap = data.stream().collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
- for (Map.Entry<Integer, List<XpOperationsUser>> integerListEntry : operationUserMap.entrySet()) {
- List<XpOperationsUser> value = integerListEntry.getValue();
- pushAliMessage(value, content, xpTask);
+ log.info("事故通知记录保存成功，共 {} 条", notifications.size());
+
+ // 按设备类型分组发送阿里云推送
+ Map<Integer, List<XpOperationsUser>> operationUserMap = data.stream()
+ .filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
+ .collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
+ for (Map.Entry<Integer, List<XpOperationsUser>> entry : operationUserMap.entrySet()) {
+ pushAliMessage(entry.getValue(), content, xpTask);
}
+ } catch (Exception e) {
+ log.error("发送事故通知失败，任务ID={}", xpTask.getId(), e);
}
- }
+ });
+
+ executor.shutdown();
+ }
+
+ /**
+ * 构建事故推送 body JSON
+ * @param content 消息内容
+ * @param xpTask 任务信息
+ * @return JSON 字符串
+ */
+ private String buildAccidentPushBodyJson(String content, XpTask xpTask) {
+ Map<String, Object> param = new HashMap<>();
+ param.put("content", content);
+ param.put("strategy", NotificationTypeEnum.ACCIDENT_NEW.getCode());
+ param.put("type", TaskTypeNotificationEnum.ACCIDENT_TASK.getCode());
+ param.put("id", xpTask.getId().toString());
+ return JSONObject.toJSONString(param);
}
/**
@@ -1070,33 +1125,9 @@
xpTask.setEndTime(endTime);
this.save(xpTask);
- // 发送消息通知
- List<Long> userIds = Arrays.stream(xpTask.getHandleUserId().split(",")).map(Long::valueOf).collect(Collectors.toList());
- if (CollUtil.isNotEmpty(userIds)) {
- R<List<XpOperationsUser>> xpOperationUserResult = xpOperationUserClient.listByIds(userIds);
- List<XpOperationsUser> data = xpOperationUserResult.getData();
- if (CollUtil.isNotEmpty(data)) {
-
- String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
- // 车辆${车牌号}在${车辆位置信息}发生了紧急事故，请在${到达现场限时}内到达现场！${APP链接}
- String content = String.format("车辆%s在%s发生了紧急事故，请在%d分钟内到达现场！", plateNumber, xpTask.getVehicleLocation(), taskRule.getProcessingTimeLimit());
- List<XpOperationsUserNotification> notifications = data.stream().map(item -> {
- XpOperationsUserNotification notification = new XpOperationsUserNotification();
- notification.setOperationsUserId(item.getId());
- notification.setNotificationContent(content);
- notification.setReadStatus(0);
- notification.setCreateTime(LocalDateTime.now());
- return notification;
- }).collect(Collectors.toList());
- xpOperationUserNotificationClient.addNotificationBatch(notifications);
- // 发送阿里云推送
- Map<Integer, List<XpOperationsUser>> operationUserMap = data.stream().collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
- for (Map.Entry<Integer, List<XpOperationsUser>> integerListEntry : operationUserMap.entrySet()) {
- List<XpOperationsUser> value = integerListEntry.getValue();
- pushAliMessage(value, content, xpTask);
- }
- }
- }
+ // 异步发送消息通知
+ String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
+ sendAccidentNotificationAsync(xpTask, plateNumber, taskRule.getProcessingTimeLimit());
}
@Override
--
Gitblit v1.7.1