新增 XXL-Job 分布式调度支持，完善订单自动派单、超时取消和 Webhook 重试任务逻辑

shaofan
1 parent 815273bb
Showing 8 changed files with 147 additions and 37 deletions
pom.xml
src/main/java/com/diligrp/rider/config/XxlJobConfig.java
src/main/java/com/diligrp/rider/service/impl/DispatchServiceImpl.java
src/main/java/com/diligrp/rider/service/impl/WebhookServiceImpl.java
src/main/java/com/diligrp/rider/task/DispatchScheduleTask.java
src/main/java/com/diligrp/rider/task/OrderScheduleTask.java
src/main/java/com/diligrp/rider/task/WebhookRetryTask.java
src/main/resources/application.yml
@@ -92,6 +92,13 @@
             <version>3.6.6</version>
         </dependency>
+        <!-- XXL-Job 分布式调度 -->
+        <dependency>
+            <groupId>com.xuxueli</groupId>
+            <artifactId>xxl-job-core</artifactId>
+            <version>2.4.1</version>
+        </dependency>
+
         <!-- Test -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
+package com.diligrp.rider.config;
+
+import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Slf4j
+@Configuration
+@ConditionalOnProperty(prefix = "xxl.job", name = "enabled", havingValue = "true", matchIfMissing = true)
+public class XxlJobConfig {
+
+    @Value("${xxl.job.admin.addresses:}")
+    private String adminAddresses;
+
+    @Value("${xxl.job.accessToken:default_token}")
+    private String accessToken;
+
+    @Value("${xxl.job.executor.appname:rider-service-executor}")
+    private String appname;
+
+    @Value("${xxl.job.executor.address:}")
+    private String address;
+
+    @Value("${xxl.job.executor.ip:}")
+    private String ip;
+
+    @Value("${xxl.job.executor.port:9999}")
+    private int port;
+
+    @Value("${xxl.job.executor.logpath:./logs/xxl-job/jobhandler}")
+    private String logPath;
+
+    @Value("${xxl.job.executor.logretentiondays:30}")
+    private int logRetentionDays;
+
+    @Bean(initMethod = "start", destroyMethod = "destroy")
+    public XxlJobSpringExecutor xxlJobExecutor() {
+        log.info("XXL-Job executor 启动 adminAddresses={} appname={} port={}", adminAddresses, appname, port);
+        XxlJobSpringExecutor executor = new XxlJobSpringExecutor();
+        executor.setAdminAddresses(adminAddresses);
+        executor.setAccessToken(accessToken);
+        executor.setAppname(appname);
+        executor.setAddress(address);
+        executor.setIp(ip);
+        executor.setPort(port);
+        executor.setLogPath(logPath);
+        executor.setLogRetentionDays(logRetentionDays);
+        return executor;
+    }
+}
@@ -14,10 +14,12 @@ import com.diligrp.rider.vo.DispatchRuleTemplateVO;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import java.math.BigDecimal;
+import java.time.Duration;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
@@ -28,6 +30,9 @@ import java.util.*;
 @RequiredArgsConstructor
 public class DispatchServiceImpl implements DispatchService {
+    private static final String DISPATCH_LOCK_PREFIX = "dispatch:lock:order:";
+    private static final Duration DISPATCH_LOCK_TTL = Duration.ofSeconds(10);
+
     private final DispatchRuleService dispatchRuleService;
     private final RiderMapper riderMapper;
     private final RiderLocationMapper locationMapper;
@@ -37,10 +42,26 @@ public class DispatchServiceImpl implements DispatchService {
     private final RiderHoldLimitService riderHoldLimitService;
     private final WebhookService webhookService;
     private final ObjectMapper objectMapper;
+    private final StringRedisTemplate redisTemplate;
     @Override
     @Transactional
     public Long dispatch(Orders order) {
+        String lockKey = DISPATCH_LOCK_PREFIX + order.getId();
+        Boolean locked = redisTemplate.opsForValue()
+                .setIfAbsent(lockKey, "1", DISPATCH_LOCK_TTL);
+        if (!Boolean.TRUE.equals(locked)) {
+            log.debug("订单 {} 派单锁被占用，跳过", order.getId());
+            return null;
+        }
+        try {
+            return doDispatch(order);
+        } finally {
+            redisTemplate.delete(lockKey);
+        }
+    }
+
+    private Long doDispatch(Orders order) {
         DispatchRuleTemplateVO rule = dispatchRuleService.getActiveRule(order.getCityId());
         if (rule == null) {
             log.debug("城市 {} 无生效调度规则，跳过派单", order.getCityId());
 package com.diligrp.rider.service.impl;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.diligrp.rider.entity.OpenApp;
 import com.diligrp.rider.entity.Orders;
@@ -70,9 +71,21 @@ public class WebhookServiceImpl implements WebhookService {
     @Override
     public void retry(Long logId) {
-        WebhookLog log = webhookLogMapper.selectById(logId);
-        if (log == null || log.getStatus() == 1) return;
-        retry(log);
+        WebhookLog webhookLog = webhookLogMapper.selectById(logId);
+        if (webhookLog == null || webhookLog.getStatus() == 1) return;
+
+        Integer prevRetryCount = webhookLog.getRetryCount() == null ? 0 : webhookLog.getRetryCount();
+        int updated = webhookLogMapper.update(null, new LambdaUpdateWrapper<WebhookLog>()
+                .eq(WebhookLog::getId, logId)
+                .eq(WebhookLog::getStatus, 0)
+                .eq(WebhookLog::getRetryCount, prevRetryCount)
+                .setSql("retry_count = retry_count + 1"));
+        if (updated == 0) {
+            log.debug("Webhook 重试抢占失败 logId={}（可能被其他实例处理或状态已变）", logId);
+            return;
+        }
+        webhookLog.setRetryCount(prevRetryCount + 1);
+        retry(webhookLog);
     }
     private void retry(WebhookLog webhookLog) {
@@ -84,12 +97,13 @@ public class WebhookServiceImpl implements WebhookService {
         }
         if (url == null || url.isBlank()) return;
         SendResult result = sendHttp(app, url, webhookLog.getEvent(), webhookLog.getPayload());
-        webhookLog.setUrl(url);
-        webhookLog.setResponseCode(result.responseCode());
-        webhookLog.setResponseBody(trimResponseBody(result.responseBody()));
-        webhookLog.setStatus(result.status());
-        webhookLog.setRetryCount((webhookLog.getRetryCount() == null ? 0 : webhookLog.getRetryCount()) + 1);
-        webhookLogMapper.updateById(webhookLog);
+        // retry_count 已在 retry(Long) 入口的 CAS 中递增，这里只更新发送结果字段
+        webhookLogMapper.update(null, new LambdaUpdateWrapper<WebhookLog>()
+                .eq(WebhookLog::getId, webhookLog.getId())
+                .set(WebhookLog::getUrl, url)
+                .set(WebhookLog::getResponseCode, result.responseCode())
+                .set(WebhookLog::getResponseBody, trimResponseBody(result.responseBody()))
+                .set(WebhookLog::getStatus, result.status()));
     }
     private void doSend(OpenApp app, String url, String event, Long bizId, String payload, int retryCount) {
@@ -8,17 +8,15 @@ import com.diligrp.rider.mapper.OrdersMapper;
 import com.diligrp.rider.service.DispatchRuleService;
 import com.diligrp.rider.service.DispatchService;
 import com.diligrp.rider.vo.DispatchRuleTemplateVO;
+import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import java.util.List;
 @Slf4j
 @Component
-@EnableScheduling
 @RequiredArgsConstructor
 public class DispatchScheduleTask {
@@ -29,9 +27,8 @@ public class DispatchScheduleTask {
     /**
      * 抢单超时后自动派单
-     * 每3秒执行一次
      */
-    @Scheduled(fixedDelay = 3_000)
+    @XxlJob("autoDispatchTimeoutOrderHandler")
     public void autoDispatchTimeoutOrders() {
         try {
             List<City> cities = cityMapper.selectList(new LambdaQueryWrapper<City>()
@@ -69,6 +66,7 @@ public class DispatchScheduleTask {
             }
         } catch (Exception e) {
             log.error("自动派单任务异常", e);
+            throw new RuntimeException(e);
         }
     }
 }
@@ -7,10 +7,9 @@ import com.diligrp.rider.entity.Orders;
 import com.diligrp.rider.mapper.OrdersMapper;
 import com.diligrp.rider.service.WebhookService;
 import com.diligrp.rider.service.AdminMessageService;
+import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import java.util.HashMap;
@@ -18,12 +17,10 @@ import java.util.List;
 import java.util.Map;
 /**
- * 订单定时任务
- * OrderhandleCron（每3秒执行）
+ * 订单定时任务（XXL-Job handler）
  */
 @Slf4j
 @Component
-@EnableScheduling
 @RequiredArgsConstructor
 public class OrderScheduleTask {
@@ -34,10 +31,8 @@ public class OrderScheduleTask {
     /**
      * 超时未接单订单自动取消（30分钟）
-     * Orders::cancel()
-     * 每分钟执行一次
      */
-    @Scheduled(fixedDelay = 60_000)
+    @XxlJob("autoCancelTimeoutOrderHandler")
     public void autoCancelTimeout() {
         try {
             long expireTime = System.currentTimeMillis() / 1000 - 30 * 60;
@@ -70,14 +65,14 @@ public class OrderScheduleTask {
             }
         } catch (Exception e) {
             log.error("超时取消任务异常", e);
+            throw new RuntimeException(e);
         }
     }
     /**
      * 订单超时提醒（配送中的订单，距离预计送达时间还有5分钟）
-     * 每分钟执行一次
      */
-    @Scheduled(fixedDelay = 60_000)
+    @XxlJob("orderTimeoutReminderHandler")
     public void timeoutReminder() {
         try {
             long now = System.currentTimeMillis() / 1000;
@@ -107,7 +102,8 @@ public class OrderScheduleTask {
                 }
             }
         } catch (Exception e) {
-            log.error("超时取消任务异常", e);
+            log.error("订单超时提醒任务异常", e);
+            throw new RuntimeException(e);
         }
     }
@@ -4,9 +4,9 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.diligrp.rider.entity.WebhookLog;
 import com.diligrp.rider.mapper.WebhookLogMapper;
 import com.diligrp.rider.service.WebhookService;
+import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import java.util.List;
@@ -19,19 +19,24 @@ public class WebhookRetryTask {
     private final WebhookLogMapper webhookLogMapper;
     private final WebhookService webhookService;
-    @Scheduled(fixedDelay = 300_000)
+    @XxlJob("webhookRetryHandler")
     public void retryFailedWebhooks() {
-        List<WebhookLog> logs = webhookLogMapper.selectList(new LambdaQueryWrapper<WebhookLog>()
-                .eq(WebhookLog::getStatus, 0)
-                .lt(WebhookLog::getRetryCount, 5)
-                .orderByAsc(WebhookLog::getCreateTime)
-                .last("LIMIT 20"));
-        for (WebhookLog webhookLog : logs) {
-            try {
-                webhookService.retry(webhookLog.getId());
-            } catch (Exception e) {
-                log.warn("Webhook 自动重试失败 logId={}", webhookLog.getId(), e);
+        try {
+            List<WebhookLog> logs = webhookLogMapper.selectList(new LambdaQueryWrapper<WebhookLog>()
+                    .eq(WebhookLog::getStatus, 0)
+                    .lt(WebhookLog::getRetryCount, 5)
+                    .orderByAsc(WebhookLog::getCreateTime)
+                    .last("LIMIT 20"));
+            for (WebhookLog webhookLog : logs) {
+                try {
+                    webhookService.retry(webhookLog.getId());
+                } catch (Exception e) {
+                    log.warn("Webhook 自动重试失败 logId={}", webhookLog.getId(), e);
+                }
             }
+        } catch (Exception e) {
+            log.error("Webhook 重试任务异常", e);
+            throw new RuntimeException(e);
         }
     }
 }
@@ -2,6 +2,8 @@ server:
   port: 8080
 spring:
+  application:
+    name: rider-service
   datasource:
     driver-class-name: com.mysql.cj.jdbc.Driver
     url: jdbc:mysql://mysql.diligrp.com:3306/dili_rider?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
@@ -45,6 +47,20 @@ amap:
   connect-timeout-millis: 3000
   read-timeout-millis: 3000
+xxl:
+  job:
+    enabled: false                            # 本地开发可用 -Dxxl.job.enabled=false 关闭 executor 自启
+    accessToken: default_token               # 调度中心与执行器通信 token，生产环境务必修改
+    admin:
+      addresses: http://localhost:8081/xxl-job-admin   # 调度中心地址，多个用逗号分隔
+    executor:
+      appname: rider-service-executor        # 执行器 AppName，调度中心需要匹配
+      address:                               # 留空时由 ip+port 自动拼接
+      ip:                                    # 留空时自动获取本机 IP（多网卡需显式指定）
+      port: 9999                             # 执行器内嵌服务端口
+      logpath: ./logs/xxl-job/jobhandler     # 任务日志本地落盘路径
+      logretentiondays: 30                   # 日志保留天数
+
 logging:
   level:
     com.diligrp.rider: debug