diff --git a/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenOnlineSpuServiceImpl.java b/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenOnlineSpuServiceImpl.java index 52a74e22..cb3229c9 100644 --- a/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenOnlineSpuServiceImpl.java +++ b/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenOnlineSpuServiceImpl.java @@ -643,8 +643,9 @@ public class CanteenOnlineSpuServiceImpl extends ServiceImpl w.isNull(CanteenOnlineSpuEntity::getSpuType).or().eq(CanteenOnlineSpuEntity::getSpuType, 1)) + .and(w -> w.eq(CanteenOnlineSpuEntity::getDelFlag, "0").or().isNull(CanteenOnlineSpuEntity::getDelFlag)) .eq(CanteenOnlineSpuEntity::getStatus, 0) .orderByAsc(CanteenOnlineSpuEntity::getSort).orderByDesc(CanteenOnlineSpuEntity::getId); Page pageData = baseMapper.selectPageByScope(page, wrapper, DataScope.of()); diff --git a/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenSpuServiceImpl.java b/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenSpuServiceImpl.java index 4da09360..200e4140 100644 --- a/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenSpuServiceImpl.java +++ b/pjcloud-canteen/pjcloud-canteen-biz/src/main/java/com/bsd/pjcloud/canteen/service/impl/CanteenSpuServiceImpl.java @@ -569,8 +569,10 @@ public class CanteenSpuServiceImpl extends ServiceImpl wrapper = new LambdaQueryWrapper<>(); wrapper.like(StrUtil.isNotBlank(canteenSpu.getName()), CanteenSpuEntity::getName, canteenSpu.getName()); wrapper.eq(Objects.nonNull(canteenSpu.getCategoryId()), CanteenSpuEntity::getCategoryId, canteenSpu.getCategoryId()); - // 套餐组成只可选单品(spu_type 为空或 1) + // 套餐组成只可选单品(spu_type 为空或 1);排除逻辑删除与回收 wrapper.and(w -> w.isNull(CanteenSpuEntity::getSpuType).or().eq(CanteenSpuEntity::getSpuType, 1)); + wrapper.and(w -> w.eq(CanteenSpuEntity::getDelFlag, "0").or().isNull(CanteenSpuEntity::getDelFlag)); + wrapper.ne(CanteenSpuEntity::getStatus, -1); merchantScopeHelper.applyMerchantIdScope(wrapper, canteenSpu.getMerchantId(), CanteenSpuEntity::getMerchantId); wrapper.orderByAsc(CanteenSpuEntity::getSort).orderByDesc(CanteenSpuEntity::getId); diff --git a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelCallbackTaskEntity.java b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelCallbackTaskEntity.java index 2c8814bf..9f7a2449 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelCallbackTaskEntity.java +++ b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelCallbackTaskEntity.java @@ -21,7 +21,7 @@ public class SmHotelCallbackTaskEntity extends Model private Long tenantId; - @Schema(description = "PRICE / INVENTORY") + @Schema(description = "PRICE / INVENTORY / ORDER_STATUS(ORDER_STATUS 高优先级)") private String notifyType; @Schema(description = "上游 hotelId") diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelCallbackTaskController.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelCallbackTaskController.java index a4c761a4..d87e01cc 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelCallbackTaskController.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelCallbackTaskController.java @@ -15,7 +15,7 @@ import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.*; /** - * 上游房价/房量回调任务(后台查询与重试)。 + * 上游房价/房量回调任务(后台查询、重试、清理)。 */ @RestController @RequiredArgsConstructor @@ -30,8 +30,10 @@ public class SmHotelCallbackTaskController { @GetMapping("/page") @PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_view')") public R> page(@ParameterObject Page page, - @ParameterObject SmHotelCallbackTaskEntity query) { - return R.ok(callbackTaskService.pageTasks(page, query)); + @ParameterObject SmHotelCallbackTaskEntity query, + @RequestParam(required = false) String createTimeBegin, + @RequestParam(required = false) String createTimeEnd) { + return R.ok(callbackTaskService.pageTasks(page, query, createTimeBegin, createTimeEnd)); } @Operation(summary = "回调任务详情") @@ -48,4 +50,29 @@ public class SmHotelCallbackTaskController { public R retry(@PathVariable Long id) { return callbackTaskService.retry(id); } + + @SysLog("删除酒店回调任务") + @Operation(summary = "删除单条回调任务") + @DeleteMapping("/{id}") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_del')") + public R remove(@PathVariable Long id) { + return callbackTaskService.removeTask(id); + } + + @SysLog("按时间区间删除酒店回调任务") + @Operation(summary = "按创建时间区间删除回调任务") + @DeleteMapping("/range") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_del')") + public R removeRange(@RequestParam(required = false) String createTimeBegin, + @RequestParam(required = false) String createTimeEnd) { + return callbackTaskService.removeByCreateTimeRange(createTimeBegin, createTimeEnd); + } + + @SysLog("清空酒店回调任务") + @Operation(summary = "清空全部回调任务并清理 Redis 队列") + @DeleteMapping("/clear") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_del')") + public R clearAll() { + return callbackTaskService.clearAllTasks(); + } } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java index d9d916cf..92e9abe9 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java @@ -96,6 +96,13 @@ public class SmHotelMidController { return midService.pullPriceFromUpstream(id); } + @Operation(summary = "仅拉取上游产品样例(getRoomPrice 今天 1 天落产品)") + @PostMapping("/mid/pullProducts/{id}") + @PreAuthorize("@pms.hasPermission('supermarket_smHotel_edit')") + public R> pullProducts(@PathVariable Long id) { + return midService.pullProductsFromUpstream(id); + } + @Operation(summary = "推送到下游销售方") @PostMapping("/mid/push/{id}") @PreAuthorize("@pms.hasPermission('supermarket_smHotel_push')") @@ -159,6 +166,22 @@ public class SmHotelMidController { return syncService.getSyncLog(id); } + @SysLog("终止酒店同步死任务") + @Operation(summary = "终止仍显示进行中的同步日志(死任务)") + @PostMapping("/mid/syncLog/{id}/abort") + @PreAuthorize("@pms.hasPermission('supermarket_smHotel_edit')") + public R abortSyncLog(@PathVariable Long id) { + return syncService.abortSyncLog(id); + } + + @SysLog("清理酒店同步卡死任务") + @Operation(summary = "清理长时间无进度的 RUNNING 同步日志") + @PostMapping("/mid/syncLog/abortStale") + @PreAuthorize("@pms.hasPermission('supermarket_smHotel_edit')") + public R abortStaleSyncLogs() { + return R.ok(syncService.abortStaleRunningSyncLogs()); + } + @Operation(summary = "日更时间配置(只读展示)") @GetMapping("/mid/syncSchedule") @PreAuthorize("@pms.hasPermission('supermarket_smHotel_view')") diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouCallbackQueueWorker.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouCallbackQueueWorker.java index 1aa00406..f10cef58 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouCallbackQueueWorker.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouCallbackQueueWorker.java @@ -10,11 +10,12 @@ import org.springframework.boot.ApplicationRunner; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; -import java.util.concurrent.TimeUnit; +import java.time.Duration; +import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; /** - * 消费 Redis 回调队列:BRPOP → 异步拉取。 + * 消费 Redis 回调队列:BRPOP 高优队列优先于普通队列。 *

* 另起看门狗线程:即使 Worker 卡在某次上游调用里,也能把超时 RUNNING 重置入队,避免整队假死。 */ @@ -55,16 +56,24 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner { watchdog.setDaemon(true); watchdog.start(); - log.info("|回调队列|Worker+Watchdog 已启动|key={}|", ShangyouHotelApi.REDIS_CALLBACK_QUEUE); + log.info("|回调队列|Worker+Watchdog 已启动|high={}|normal={}|", + ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, ShangyouHotelApi.REDIS_CALLBACK_QUEUE); } private void loop() { int consecutiveFail = 0; while (running.get()) { try { - String taskId = redisTemplate.opsForList() - .rightPop(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, 5, TimeUnit.SECONDS); - if (taskId == null || taskId.isBlank()) { + // BRPOP 多 key:先高优(订单状态)再普通;阻塞期间新到的高优也会被优先弹出 + List popped = redisTemplate.opsForList().rightPop( + List.of(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, + ShangyouHotelApi.REDIS_CALLBACK_QUEUE), + Duration.ofSeconds(5)); + if (popped == null || popped.size() < 2 || popped.get(1) == null) { + continue; + } + String taskId = popped.get(1).trim(); + if (taskId.isBlank() || "null".equalsIgnoreCase(taskId)) { continue; } Long id; diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelApi.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelApi.java index af39e08a..fdadeae8 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelApi.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelApi.java @@ -41,9 +41,12 @@ public final class ShangyouHotelApi { /** 酒店列表同步锁 */ public static final String REDIS_HOTEL_LIST_LOCK = "supermarket:shangyou:hotel:list:lock"; - /** 房价/房量回调异步队列(LIST,元素=taskId) */ + /** 回调异步队列(LIST,元素=taskId):房价/房量等普通优先级 */ public static final String REDIS_CALLBACK_QUEUE = "supermarket:shangyou:hotel:callback:queue"; + /** 回调高优先级队列:订单状态等,Worker 始终先消费本队列 */ + public static final String REDIS_CALLBACK_QUEUE_HIGH = "supermarket:shangyou:hotel:callback:queue:high"; + /** 上游授权 hotelId 集合(与 HASH/ZSET 同步重建,供 SDIFF) */ public static final String REDIS_UPSTREAM_ID_SET = "supermarket:shangyou:hotel:idset:v4"; diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelCallbackTaskService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelCallbackTaskService.java index bf86b523..86e1496d 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelCallbackTaskService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelCallbackTaskService.java @@ -17,15 +17,27 @@ public interface SmHotelCallbackTaskService { */ Map enqueueFromNotify(String notifyType, Map body, String time, String sign); - /** 后台分页 */ + /** 后台分页(可按回调时间 createTime 区间筛选) */ Page pageTasks(Page page, - SmHotelCallbackTaskEntity query); + SmHotelCallbackTaskEntity query, String createTimeBegin, String createTimeEnd); SmHotelCallbackTaskEntity getDetail(Long id); /** 失败/跳过任务重新入队 */ R retry(Long id); + /** 删除单条回调任务 */ + R removeTask(Long id); + + /** + * 按创建时间区间删除(含边界)。begin/end 都空则拒绝。 + * @return 删除条数 + */ + R removeByCreateTimeRange(String createTimeBegin, String createTimeEnd); + + /** 清空全部回调任务,并清空 Redis 消费队列 */ + R clearAllTasks(); + /** 消费一条队列任务(Worker 调用);返回 true=本次成功/跳过,false=失败或待再试 */ boolean processQueuedTask(Long taskId); diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java index 4854621f..44da4e81 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java @@ -26,6 +26,11 @@ public interface SmHotelMidService { */ R> pullPriceFromUpstream(Long hotelId); + /** + * 仅拉取产品样例(默认今天 1 天 getRoomPrice 落产品),用于单独验证产品解析落库。 + */ + R> pullProductsFromUpstream(Long hotelId); + /** * 仅拉取指定日期区间房价(含房型综合若本地无房型),用于「滚一天」;区间建议 1 天,最长按上游 31 天切片。 */ diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java index ba2ac3cc..64b9382c 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java @@ -44,6 +44,15 @@ public interface SmHotelSyncService { */ R startRetryFailedAsync(Long sourceLogId); + /** 人工终止一条仍显示进行中的同步日志(死任务) */ + R abortSyncLog(Long id); + + /** + * 将长时间无进度更新的 RUNNING 日志标为失败,避免挡后续任务。 + * @return 清理条数 + */ + int abortStaleRunningSyncLogs(); + /** 兼容旧同步入口(内部转异步) */ R> syncHotelListAndDiff(boolean autoImport); diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelCallbackTaskServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelCallbackTaskServiceImpl.java index f0867d61..c3c75d97 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelCallbackTaskServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelCallbackTaskServiceImpl.java @@ -79,11 +79,12 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic SmHotelCallbackTaskEntity task = buildTask(notifyType, body, time, tid); taskMapper.insert(task); if (properties.isCallbackQueueEnabled()) { - redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, - String.valueOf(task.getId())); + enqueueTaskId(task.getId(), task.getNotifyType()); } - log.info("|上游酒店通知|已入队|taskId={}|hotelCode={}|hotelId={}|status={}|", task.getId(), - task.getHotelCode(), task.getHotelId(), task.getStatus()); + log.info("|上游酒店通知|已入队|taskId={}|type={}|priority={}|hotelCode={}|hotelId={}|status={}|", + task.getId(), task.getNotifyType(), + isHighPriority(task.getNotifyType()) ? "HIGH" : "NORMAL", task.getHotelCode(), + task.getHotelId(), task.getStatus()); }); } catch (Exception e) { @@ -96,7 +97,9 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic @Override public Page pageTasks(Page page, - SmHotelCallbackTaskEntity query) { + SmHotelCallbackTaskEntity query, String createTimeBegin, String createTimeEnd) { + LocalDateTime begin = parseDateTime(createTimeBegin, true); + LocalDateTime end = parseDateTime(createTimeEnd, false); return taskMapper.selectPage(page, Wrappers.lambdaQuery() .eq(StrUtil.isNotBlank(query.getNotifyType()), SmHotelCallbackTaskEntity::getNotifyType, query.getNotifyType()) @@ -105,6 +108,8 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic query.getHotelCode()) .like(StrUtil.isNotBlank(query.getHotelName()), SmHotelCallbackTaskEntity::getHotelName, query.getHotelName()) + .ge(begin != null, SmHotelCallbackTaskEntity::getCreateTime, begin) + .le(end != null, SmHotelCallbackTaskEntity::getCreateTime, end) .orderByDesc(SmHotelCallbackTaskEntity::getCreateTime)); } @@ -137,8 +142,87 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic task.setCostMs(null); task.setUpdateTime(LocalDateTime.now()); taskMapper.updateById(task); - redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(task.getId())); - return R.ok(Boolean.TRUE, "已重新入队"); + enqueueTaskId(task.getId(), task.getNotifyType()); + return R.ok(Boolean.TRUE, isHighPriority(task.getNotifyType()) ? "已重新入队(高优先级)" : "已重新入队"); + } + + @Override + public R removeTask(Long id) { + if (id == null) { + return R.failed("任务ID不能为空"); + } + SmHotelCallbackTaskEntity task = taskMapper.selectById(id); + if (task == null) { + return R.failed("任务不存在"); + } + if (STATUS_RUNNING.equals(task.getStatus())) { + return R.failed("任务处理中,请稍后再删"); + } + taskMapper.deleteById(id); + return R.ok(Boolean.TRUE, "已删除"); + } + + @Override + public R removeByCreateTimeRange(String createTimeBegin, String createTimeEnd) { + LocalDateTime begin = parseDateTime(createTimeBegin, true); + LocalDateTime end = parseDateTime(createTimeEnd, false); + if (begin == null && end == null) { + return R.failed("请指定开始或结束时间"); + } + if (begin != null && end != null && end.isBefore(begin)) { + return R.failed("结束时间不能早于开始时间"); + } + // 不删 RUNNING,避免删掉正在消费的任务导致状态错乱 + var qw = Wrappers.lambdaQuery() + .ne(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING); + if (begin != null) { + qw.ge(SmHotelCallbackTaskEntity::getCreateTime, begin); + } + if (end != null) { + qw.le(SmHotelCallbackTaskEntity::getCreateTime, end); + } + Long cnt = taskMapper.selectCount(qw); + int n = taskMapper.delete(qw); + log.info("|回调任务|按创建时间删除|begin={}|end={}|deleted={}|matched={}|", begin, end, n, cnt); + return R.ok(n, "已删除 " + n + " 条"); + } + + @Override + public R clearAllTasks() { + Long cnt = taskMapper.selectCount(Wrappers.lambdaQuery() + .ne(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING)); + int n = taskMapper.delete(Wrappers.lambdaQuery() + .ne(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING)); + // 清空消费队列(含高优),避免旧 taskId 反复弹出 + Long qLen = redisTemplate.opsForList().size(ShangyouHotelApi.REDIS_CALLBACK_QUEUE); + Long qHigh = redisTemplate.opsForList().size(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH); + redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUE); + redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH); + log.warn("|回调任务|清空全部|deleted={}|queueCleared≈{}/high≈{}|", n, qLen, qHigh); + return R.ok(n, "已清空 " + n + " 条任务,并清理队列" + + (cnt != null && cnt > n ? "(处理中任务未删)" : "")); + } + + /** 支持 yyyy-MM-dd 或 yyyy-MM-dd HH:mm:ss;日期无时分秒时 begin=00:00:00、end=23:59:59 */ + private static LocalDateTime parseDateTime(String raw, boolean beginOfDay) { + if (StrUtil.isBlank(raw)) { + return null; + } + String s = raw.trim().replace('T', ' '); + try { + if (s.length() <= 10) { + LocalDate d = LocalDate.parse(s.substring(0, 10), DateTimeFormatter.ISO_LOCAL_DATE); + return beginOfDay ? d.atStartOfDay() : d.atTime(23, 59, 59); + } + if (s.length() == 16) { + return LocalDateTime.parse(s, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm")); + } + return LocalDateTime.parse(s.substring(0, Math.min(19, s.length())), + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); + } + catch (Exception e) { + return null; + } } @Override @@ -156,14 +240,13 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId(); return TenantBroker.applyAs(tenantId, tid -> { int n = 0; - // 1) PENDING + // 1) PENDING:按创建时间 FIFO;ORDER_STATUS 进高优队列 var pending = taskMapper.selectList(Wrappers.lambdaQuery() .eq(SmHotelCallbackTaskEntity::getStatus, STATUS_PENDING) .orderByAsc(SmHotelCallbackTaskEntity::getCreateTime) .last("limit 200")); - // 按创建时间从早到晚 leftPush:最终右侧是最早的,rightPop 先消费最早任务 for (SmHotelCallbackTaskEntity t : pending) { - redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId())); + enqueueTaskId(t.getId(), t.getNotifyType()); n++; } // 2) 卡住的 RUNNING @@ -209,7 +292,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic t.setStatus(STATUS_PENDING); t.setResultMsg("超时 RUNNING 已重置为 PENDING 并重新入队"); taskMapper.updateById(t); - redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId())); + enqueueTaskId(t.getId(), t.getNotifyType()); n++; } if (n > 0) { @@ -342,9 +425,9 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic task.setCostMs(System.currentTimeMillis() - t0); task.setUpdateTime(LocalDateTime.now()); taskMapper.updateById(task); - // 队列约定:入队 leftPush、消费 rightPop(FIFO)。 + // 队列约定:入队 leftPush、消费 rightPop(FIFO);ORDER_STATUS 进高优队列仍 FIFO。 // 失败重试必须 leftPush 到队尾;rightPush 会插到 rightPop 端,等于立刻再跑同一条,饿死后面的新任务。 - redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(task.getId())); + enqueueTaskId(task.getId(), task.getNotifyType()); return; } finish(task, STATUS_FAILED, msg, t0); @@ -387,7 +470,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic task.setReqJson(body == null ? null : JSONUtil.toJsonStr(body)); task.setStatus(STATUS_PENDING); task.setAttemptCount(0); - task.setResultMsg("ORDER_STATUS".equalsIgnoreCase(notifyType) ? "已入队,等待异步更新订单状态" + task.setResultMsg(isHighPriority(notifyType) ? "已入队(高优先级),等待异步更新订单状态" : "已入队,等待异步拉取"); task.setCreateBy("shangyou-callback"); task.setCreateTime(LocalDateTime.now()); @@ -396,6 +479,21 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic return task; } + /** 订单状态回调最高优先:进高优队列,Worker 始终先于房价/房量消费 */ + private void enqueueTaskId(Long taskId, String notifyType) { + String id = String.valueOf(taskId); + if (isHighPriority(notifyType)) { + redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, id); + } + else { + redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, id); + } + } + + private static boolean isHighPriority(String notifyType) { + return "ORDER_STATUS".equalsIgnoreCase(notifyType); + } + private static String str(Map body, String key) { if (body == null || body.get(key) == null) { return null; diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java index 37888961..395a7382 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java @@ -221,6 +221,23 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { return R.ok(data); } + @Override + @Transactional(rollbackFor = Exception.class) + public R> pullProductsFromUpstream(Long hotelId) { + // 产品来自 getRoomPrice 的 roomRateList;与「拉取房价」同接口,侧重看产品落库数 + R> r = pullPriceFromUpstream(hotelId); + if (r == null) { + return R.failed("拉取产品返回空"); + } + if (r.getCode() != 0) { + return r; + } + Map data = r.getData() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(r.getData()); + log.info("|酒店拉取产品|结束|hotelCode={}|product={}|daily={}|", data.get("hotelCode"), data.get("productCount"), + data.get("dailyCount")); + return R.ok(data); + } + @Override @Transactional(rollbackFor = Exception.class) public R> pullPriceRange(Long hotelId, LocalDate startInclusive, LocalDate endExclusive) { diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java index c209c45c..7d899140 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java @@ -54,6 +54,9 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { private static final String STATUS_FAILED = "FAILED"; + /** 同步日志超过该时间无 progress 心跳,视为死任务 */ + private static final int STALE_RUNNING_MINUTES = 30; + private static final ExecutorService EXECUTOR = Executors.newFixedThreadPool(2, r -> { Thread t = new Thread(r, "sm-hotel-sync"); t.setDaemon(true); @@ -74,11 +77,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { @Override public R startListSyncAsync(boolean autoImport, String triggerType) { + abortStaleRunningSyncLogs(); Long running = syncLogMapper.selectCount(Wrappers.lambdaQuery() .eq(SmHotelSyncLogEntity::getSyncType, TYPE_LIST) .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)); if (running != null && running > 0) { - return R.failed("已有列表同步任务进行中,请稍候查看进度"); + return R.failed("已有列表同步任务进行中,请稍候查看进度(若长时间卡住请在同步日志点「终止」)"); } SmHotelSyncLogEntity logEntity = newLog(TYPE_LIST, triggerType); logEntity.setStage("QUEUED"); @@ -102,11 +106,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { if (!properties.isPriceRollDailyEnabled() && !"MANUAL".equalsIgnoreCase(triggerType)) { return R.failed("滚价未启用"); } + abortStaleRunningSyncLogs(); Long running = syncLogMapper.selectCount(Wrappers.lambdaQuery() .eq(SmHotelSyncLogEntity::getSyncType, TYPE_ROLL) .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)); if (running != null && running > 0) { - return R.failed("已有滚价任务进行中"); + return R.failed("已有滚价任务进行中(若长时间卡住请在同步日志点「终止」)"); } SmHotelSyncLogEntity logEntity = newLog(TYPE_ROLL, triggerType); logEntity.setStage("QUEUED"); @@ -125,11 +130,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { @Override public R startDetailPullAsync(Integer limit, String triggerType) { + abortStaleRunningSyncLogs(); Long running = syncLogMapper.selectCount(Wrappers.lambdaQuery() .eq(SmHotelSyncLogEntity::getSyncType, TYPE_DETAIL) .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)); if (running != null && running > 0) { - return R.failed("已有全量详情拉取任务进行中,请打开「同步日志」查看进度"); + return R.failed("已有全量详情拉取任务进行中,请打开「同步日志」查看进度(若长时间卡住请点「终止」)"); } String quotaErr = checkFullPriceSyncQuota(); if (quotaErr != null) { @@ -176,11 +182,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { if (CollUtil.isEmpty(codes)) { return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因)"); } + abortStaleRunningSyncLogs(); Long running = syncLogMapper.selectCount(Wrappers.lambdaQuery() .eq(SmHotelSyncLogEntity::getSyncType, syncType) .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)); if (running != null && running > 0) { - return R.failed("已有同类型任务进行中,请稍后再试"); + return R.failed("已有同类型任务进行中,请稍后再试(若卡住请先终止)"); } SmHotelSyncLogEntity logEntity = newLog(syncType, "RETRY"); @@ -212,6 +219,51 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { return R.ok(e); } + @Override + public R abortSyncLog(Long id) { + if (id == null) { + return R.failed("日志ID不能为空"); + } + SmHotelSyncLogEntity e = syncLogMapper.selectById(id); + if (e == null) { + return R.failed("同步记录不存在"); + } + if (!STATUS_RUNNING.equals(e.getStatus())) { + return R.failed("仅能终止「进行中」的任务,当前状态=" + e.getStatus()); + } + failLog(id, "人工终止死任务(线程已中断或进程重启后残留)", + e.getCreateTime() == null ? 0L + : Math.max(0L, java.time.Duration.between(e.getCreateTime(), LocalDateTime.now()).toMillis())); + log.warn("|酒店同步|人工终止|logId={}|type={}|stage={}|pct={}|", id, e.getSyncType(), e.getStage(), + e.getProgressPct()); + return R.ok(Boolean.TRUE, "已终止并标记失败,可重新发起同步"); + } + + @Override + public int abortStaleRunningSyncLogs() { + LocalDateTime staleBefore = LocalDateTime.now().minusMinutes(STALE_RUNNING_MINUTES); + var list = syncLogMapper.selectList(Wrappers.lambdaQuery() + .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING) + .and(w -> w.lt(SmHotelSyncLogEntity::getUpdateTime, staleBefore) + .or() + .isNull(SmHotelSyncLogEntity::getUpdateTime) + .lt(SmHotelSyncLogEntity::getCreateTime, staleBefore)) + .last("limit 50")); + int n = 0; + for (SmHotelSyncLogEntity e : list) { + failLog(e.getId(), + "超时无进度(>" + STALE_RUNNING_MINUTES + " 分钟),自动标记失败以免挡后续任务", + e.getCreateTime() == null ? 0L + : Math.max(0L, + java.time.Duration.between(e.getCreateTime(), LocalDateTime.now()).toMillis())); + n++; + } + if (n > 0) { + log.warn("|酒店同步|自动清理卡死 RUNNING|count={}|staleMinutes={}|", n, STALE_RUNNING_MINUTES); + } + return n; + } + @Override public Page pageSyncLogs(Page page, SmHotelSyncLogEntity query) {