1
This commit is contained in:
@@ -202,6 +202,8 @@ jarvis:
|
||||
base-url: http://192.168.8.88:5001
|
||||
fetch-path: /fetch_logistics
|
||||
health-path: /health
|
||||
# 每次定时任务最多处理多少条企微分享链待队列(RPUSH 入队、LPOP 出队)
|
||||
adhoc-pending-batch-size: 50
|
||||
# 获取评论接口服务地址(后端转发,避免前端跨域)
|
||||
fetch-comments:
|
||||
base-url: http://192.168.8.60:5008
|
||||
|
||||
@@ -202,6 +202,7 @@ jarvis:
|
||||
base-url: http://127.0.0.1:5001
|
||||
fetch-path: /fetch_logistics
|
||||
health-path: /health
|
||||
adhoc-pending-batch-size: 50
|
||||
# 获取评论接口服务地址(后端转发)
|
||||
fetch-comments:
|
||||
base-url: http://192.168.8.60:5008
|
||||
|
||||
@@ -17,6 +17,16 @@ public interface ILogisticsService {
|
||||
* 分享类京东物流短链:查运单并通过 wxts 推送到指定 touser(不依赖订单表)
|
||||
*/
|
||||
boolean fetchLogisticsByShareLinkAndPush(String trackingUrl, String remark, String touser);
|
||||
|
||||
/**
|
||||
* 企微备注提交后仅入队,由 {@link com.ruoyi.jarvis.task.LogisticsScanTask} 与订单扫描一并拉取物流,避免阻塞 HTTP 回调。
|
||||
*/
|
||||
void enqueueShareLinkForScan(String trackingUrl, String remark, String touser);
|
||||
|
||||
/**
|
||||
* 定时任务内:依次弹出队列并调用 {@link #fetchLogisticsByShareLinkAndPush}。
|
||||
*/
|
||||
void drainPendingShareLinkQueue();
|
||||
|
||||
/**
|
||||
* 检查订单是否已处理过(Redis中是否有运单号)
|
||||
|
||||
@@ -35,6 +35,8 @@ public class LogisticsServiceImpl implements ILogisticsService {
|
||||
private static final String REDIS_LOCK_KEY_PREFIX = "logistics:lock:order:";
|
||||
private static final String REDIS_HEALTH_CHECK_ALERT_KEY = "logistics:health:alert:";
|
||||
private static final String REDIS_ADHOC_WAYBILL_PREFIX = "logistics:adhoc:waybill:";
|
||||
/** 企微分享链+备注入队,FIFO:RPUSH 入队、LPOP 出队 */
|
||||
private static final String REDIS_ADHOC_PENDING_QUEUE = "logistics:adhoc:pending:queue";
|
||||
private static final String PUSH_URL = "https://wxts.van333.cn/wx/send/pdd";
|
||||
private static final String PUSH_TOKEN = "super_token_b62190c26";
|
||||
private static final String CONFIG_KEY_PREFIX = "logistics.push.touser.";
|
||||
@@ -49,6 +51,9 @@ public class LogisticsServiceImpl implements ILogisticsService {
|
||||
|
||||
@Value("${jarvis.server.logistics.health-path:/health}")
|
||||
private String logisticsHealthPath;
|
||||
|
||||
@Value("${jarvis.server.logistics.adhoc-pending-batch-size:50}")
|
||||
private int adhocPendingBatchSize;
|
||||
|
||||
private String externalApiUrlTemplate;
|
||||
private String healthCheckUrl;
|
||||
@@ -423,6 +428,43 @@ public class LogisticsServiceImpl implements ILogisticsService {
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void enqueueShareLinkForScan(String trackingUrl, String remark, String touser) {
|
||||
if (!StringUtils.hasText(trackingUrl)) {
|
||||
logger.warn("adhoc 入队跳过:分享链接为空");
|
||||
return;
|
||||
}
|
||||
JSONObject o = new JSONObject();
|
||||
o.put("trackingUrl", trackingUrl.trim());
|
||||
o.put("remark", remark != null ? remark : "");
|
||||
o.put("touser", touser != null ? touser : "");
|
||||
stringRedisTemplate.opsForList().rightPush(REDIS_ADHOC_PENDING_QUEUE, o.toJSONString());
|
||||
logger.info("adhoc 分享链接已入队待定时扫描(与订单物流任务一致) trackingUrl={} queueKey={}", trackingUrl.trim(), REDIS_ADHOC_PENDING_QUEUE);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void drainPendingShareLinkQueue() {
|
||||
int n = Math.max(1, adhocPendingBatchSize);
|
||||
for (int i = 0; i < n; i++) {
|
||||
String json = stringRedisTemplate.opsForList().leftPop(REDIS_ADHOC_PENDING_QUEUE);
|
||||
if (json == null || !StringUtils.hasText(json)) {
|
||||
break;
|
||||
}
|
||||
try {
|
||||
JSONObject o = JSON.parseObject(json);
|
||||
if (o == null) {
|
||||
continue;
|
||||
}
|
||||
String url = o.getString("trackingUrl");
|
||||
String remark = o.getString("remark");
|
||||
String touser = o.getString("touser");
|
||||
fetchLogisticsByShareLinkAndPush(url, remark, touser);
|
||||
} catch (Exception e) {
|
||||
logger.error("adhoc 队列项处理失败 raw={}", json.length() > 200 ? json.substring(0, 200) + "..." : json, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean fetchLogisticsByShareLinkAndPush(String trackingUrl, String remark, String touser) {
|
||||
if (!StringUtils.hasText(trackingUrl)) {
|
||||
|
||||
@@ -82,8 +82,8 @@ public class WeComInboundServiceImpl implements IWeComInboundService {
|
||||
weComChatSessionService.delete(from);
|
||||
String touser = resolveTouser(row, isSuper);
|
||||
log.info("企微物流会话提交备注 user={} url={} remarkLen={}", from, url, t.length());
|
||||
boolean ok = logisticsService.fetchLogisticsByShareLinkAndPush(url, content.trim(), touser);
|
||||
return ok ? "已查询并推送运单(接收人见超级管理员 touser)" : "物流查询或推送失败,请稍后重试";
|
||||
logisticsService.enqueueShareLinkForScan(url, content.trim(), touser);
|
||||
return "已登记备注,物流将随定时任务查询并推送(与订单物流扫描一致,约每10分钟)";
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ import java.util.List;
|
||||
|
||||
/**
|
||||
* 物流信息扫描定时任务
|
||||
* 每20分钟扫描一次分销标记为F或PDD的订单(最近30天),获取物流信息并推送
|
||||
* 每10分钟扫描一次分销标记为F或PDD的订单(最近30天),获取物流信息并推送;结束后处理企微分享链 adhoc 队列
|
||||
*/
|
||||
@Component
|
||||
public class LogisticsScanTask {
|
||||
@@ -26,8 +26,7 @@ public class LogisticsScanTask {
|
||||
private ILogisticsService logisticsService;
|
||||
|
||||
/**
|
||||
* 定时任务:每20分钟执行一次
|
||||
* Cron表达式格式:0 每N分钟 * * * ? 表示每N分钟执行一次
|
||||
* 定时任务:每10分钟执行一次(与 @Scheduled 中 cron 一致)
|
||||
* 只扫描最近30天的订单
|
||||
*/
|
||||
@Scheduled(cron = "0 */10 * * * ?")
|
||||
@@ -40,62 +39,67 @@ public class LogisticsScanTask {
|
||||
|
||||
if (orders == null || orders.isEmpty()) {
|
||||
logger.info("未找到需要处理的订单");
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
logger.info("找到 {} 个需要处理的订单", orders.size());
|
||||
|
||||
logger.info("找到 {} 个需要处理的订单", orders.size());
|
||||
int processedCount = 0;
|
||||
int skippedCount = 0;
|
||||
int successCount = 0;
|
||||
int failedCount = 0;
|
||||
|
||||
int processedCount = 0;
|
||||
int skippedCount = 0;
|
||||
int successCount = 0;
|
||||
int failedCount = 0;
|
||||
// 串行处理订单(避免并发调用接口)
|
||||
for (JDOrder order : orders) {
|
||||
try {
|
||||
// 检查Redis中是否已处理过(避免重复处理)
|
||||
if (logisticsService.isOrderProcessed(order.getId())) {
|
||||
//logger.debug("订单已处理过,跳过 - 订单ID: {}", order.getId());
|
||||
skippedCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// 串行处理订单(避免并发调用接口)
|
||||
for (JDOrder order : orders) {
|
||||
try {
|
||||
// 检查Redis中是否已处理过(避免重复处理)
|
||||
if (logisticsService.isOrderProcessed(order.getId())) {
|
||||
//logger.debug("订单已处理过,跳过 - 订单ID: {}", order.getId());
|
||||
skippedCount++;
|
||||
continue;
|
||||
}
|
||||
logger.info("开始处理订单 - 订单ID: {}, 订单号: {}, 分销标识: {}",
|
||||
order.getId(), order.getOrderId(), order.getDistributionMark());
|
||||
|
||||
logger.info("开始处理订单 - 订单ID: {}, 订单号: {}, 分销标识: {}",
|
||||
order.getId(), order.getOrderId(), order.getDistributionMark());
|
||||
// 获取物流信息并推送(串行执行,不并发)
|
||||
boolean success = logisticsService.fetchLogisticsAndPush(order);
|
||||
|
||||
// 获取物流信息并推送(串行执行,不并发)
|
||||
boolean success = logisticsService.fetchLogisticsAndPush(order);
|
||||
if (success) {
|
||||
successCount++;
|
||||
logger.info("订单处理成功 - 订单ID: {}, 订单号: {}", order.getId(), order.getOrderId());
|
||||
} else {
|
||||
failedCount++;
|
||||
logger.warn("订单处理失败 - 订单ID: {}, 订单号: {}", order.getId(), order.getOrderId());
|
||||
}
|
||||
|
||||
if (success) {
|
||||
successCount++;
|
||||
logger.info("订单处理成功 - 订单ID: {}, 订单号: {}", order.getId(), order.getOrderId());
|
||||
} else {
|
||||
processedCount++;
|
||||
|
||||
// 添加短暂延迟,避免请求过于频繁
|
||||
Thread.sleep(500); // 每次请求间隔500毫秒
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
logger.error("定时任务被中断", e);
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
} catch (Exception e) {
|
||||
failedCount++;
|
||||
logger.warn("订单处理失败 - 订单ID: {}, 订单号: {}", order.getId(), order.getOrderId());
|
||||
logger.error("处理订单时发生异常 - 订单ID: {}, 错误: {}", order.getId(), e.getMessage(), e);
|
||||
// 继续处理下一个订单
|
||||
}
|
||||
|
||||
processedCount++;
|
||||
|
||||
// 添加短暂延迟,避免请求过于频繁
|
||||
Thread.sleep(500); // 每次请求间隔500毫秒
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
logger.error("定时任务被中断", e);
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
} catch (Exception e) {
|
||||
failedCount++;
|
||||
logger.error("处理订单时发生异常 - 订单ID: {}, 错误: {}", order.getId(), e.getMessage(), e);
|
||||
// 继续处理下一个订单
|
||||
}
|
||||
|
||||
logger.info("========== 物流信息扫描定时任务执行完成 ==========");
|
||||
logger.info("总订单数: {}, 已处理: {}, 跳过: {}, 成功: {}, 失败: {}",
|
||||
orders.size(), processedCount, skippedCount, successCount, failedCount);
|
||||
}
|
||||
|
||||
logger.info("========== 物流信息扫描定时任务执行完成 ==========");
|
||||
logger.info("总订单数: {}, 已处理: {}, 跳过: {}, 成功: {}, 失败: {}",
|
||||
orders.size(), processedCount, skippedCount, successCount, failedCount);
|
||||
|
||||
} catch (Exception e) {
|
||||
logger.error("执行物流信息扫描定时任务时发生异常", e);
|
||||
} finally {
|
||||
try {
|
||||
logger.info("---------- 处理企微分享链物流待队列(adhoc) ----------");
|
||||
logisticsService.drainPendingShareLinkQueue();
|
||||
} catch (Exception e) {
|
||||
logger.error("企微分享链 adhoc 队列处理异常", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user