diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelAppController.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelAppController.java index c464d936..f68502d8 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelAppController.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelAppController.java @@ -89,7 +89,7 @@ public class SmHotelAppController { private static final String CANDIDATE_CACHE_OVERFLOW = "OVERFLOW"; /** 列表页结果短缓存(同条件重复进入/下拉) */ - private static final String PAGE_CACHE_PREFIX = "sm:hotel:app:page:v9:"; + private static final String PAGE_CACHE_PREFIX = "sm:hotel:app:page:v10:"; private static final long PAGE_CACHE_SECONDS = 45L; @@ -248,79 +248,12 @@ public class SmHotelAppController { return R.ok(cached); } - boolean hasScopeFilter = districtAdcode != null || StrUtil.isNotBlank(district) || StrUtil.isNotBlank(city) - || StrUtil.isNotBlank(keyword) || StrUtil.isNotBlank(bizArea) || !starList.isEmpty() - || minScore != null || !facilityList.isEmpty(); - List scopedIds = null; - if (hasScopeFilter) { - scopedIds = loadScopedHotelIds(keyword, city, district, districtAdcode, bizArea, starList, minScore, - facilityList); - if (scopedIds != null && scopedIds.isEmpty()) { - Page> empty = emptyPage(page); - writePageCache(pageCacheKey, empty); - return R.ok(empty); - } - } - boolean smallScope = scopedIds != null && !scopedIds.isEmpty() && scopedIds.size() <= MAX_SCOPED_HOTEL_IDS; - Page raw; - int candSize = -1; - if (smallScope) { - List candidateIds = scopedIds; - if (needCap > 1) { - List capHotels = roomMapper.selectDistinctHotelIdsByMinCapacityInHotels(needCap, candidateIds); - if (capHotels == null || capHotels.isEmpty()) { - Page> empty = emptyPage(page); - writePageCache(pageCacheKey, empty); - return R.ok(empty); - } - Set capSet = new HashSet<>(capHotels); - candidateIds = candidateIds.stream().filter(capSet::contains).toList(); - if (candidateIds.isEmpty()) { - Page> empty = emptyPage(page); - writePageCache(pageCacheKey, empty); - return R.ok(empty); - } - } - Map productMinMap = queryProductMinPricesByIds(candidateIds); - if (minPrice != null || maxPrice != null) { - candidateIds = candidateIds.stream().filter(id -> inProductMinRange(productMinMap.get(id), minPrice, maxPrice)) - .toList(); - if (candidateIds.isEmpty()) { - Page> empty = emptyPage(page); - writePageCache(pageCacheKey, empty); - return R.ok(empty); - } - } - candSize = candidateIds.size(); - List matched = hotelService.list(baseOnlineListWrapper(keyword, city, district, - districtAdcode, bizArea, starList, minScore, facilityList) - .select(SmHotelEntity::getId, SmHotelEntity::getStarLevel) - .in(SmHotelEntity::getId, candidateIds)); - if (matched == null || matched.isEmpty()) { - Page> empty = emptyPage(page); - writePageCache(pageCacheKey, empty); - return R.ok(empty); - } - matched = new ArrayList<>(matched); - matched.sort(Comparator - .comparing((SmHotelEntity h) -> h.getStarLevel() == null ? 0 : h.getStarLevel()) - .reversed() - .thenComparing((SmHotelEntity h) -> productMinMap.getOrDefault(h.getId(), BigDecimal.ZERO), - Comparator.reverseOrder()) - .thenComparing((SmHotelEntity h) -> h.getId() == null ? 0L : h.getId(), Comparator.reverseOrder())); - raw = sliceWithoutCount(page, matched); - List pageIds = raw.getRecords().stream().map(SmHotelEntity::getId).filter(Objects::nonNull).toList(); - raw.setRecords(loadListRowsInOrder(pageIds)); - } - else { - // 全国或匹配过多:只取一页,禁止 list 全表。排序用星级+id,避免产品相关子查询把请求拖死。 - var wrapper = applyListColumns(baseOnlineListWrapper(keyword, city, district, districtAdcode, bizArea, - starList, minScore, facilityList)); - applyGuestCapacityFilter(wrapper, needCap); - applyMinProductPriceFilter(wrapper, minPrice, maxPrice); - wrapper.last("ORDER BY IFNULL(star_level, 0) DESC, id DESC"); - raw = pageWithoutCount(page, wrapper); - } + var wrapper = applyListColumns(baseOnlineListWrapper(keyword, city, district, districtAdcode, bizArea, + starList, minScore, facilityList)); + applyGuestCapacityFilter(wrapper, needCap); + applyMinProductPriceFilter(wrapper, minPrice, maxPrice); + wrapper.last("ORDER BY IFNULL(star_level, 0) DESC, id DESC"); + Page raw = pageWithoutCount(page, wrapper); List ids = raw.getRecords().stream().map(SmHotelEntity::getId).filter(Objects::nonNull).toList(); LocalDate priceFrom = stayFrom != null ? stayFrom : LocalDate.now(); LocalDate priceToEx = stayToExclusive; @@ -334,9 +267,8 @@ public class SmHotelAppController { .map(h -> toListRow(h, minPriceMap.get(h.getId()))) .toList()); writePageCache(pageCacheKey, out); - log.info("|酒店App|page|current={}|size={}|total={}|scope={}|cand={}|costMs={}|", page.getCurrent(), - page.getSize(), raw.getTotal(), scopedIds == null ? -1 : scopedIds.size(), candSize, - System.currentTimeMillis() - t0); + log.info("|酒店App|page|current={}|size={}|total={}|costMs={}|", page.getCurrent(), page.getSize(), + raw.getTotal(), System.currentTimeMillis() - t0); return R.ok(out); } 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 2c77ead8..68c9a2d9 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 @@ -99,12 +99,16 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner { log.warn("|回调队列|非法 taskId={}|", taskId); continue; } - boolean ok = callbackTaskService.processQueuedTask(id); + int result = callbackTaskService.processQueuedTask(id); + if (result == 0) { + // 残留 SUCCESS 副本:不限流、不打间隔,尽快排空以免占连接 + continue; + } int interval = Math.max(0, p.getCallbackTaskIntervalMs()); if (interval > 0) { Thread.sleep(interval); } - if (ok) { + if (result > 0) { consecutiveFail = 0; } else { 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 c226d329..a1ed9497 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 @@ -44,6 +44,9 @@ public final class ShangyouHotelApi { /** 回调异步队列(LIST,元素=taskId):房价/房量等普通优先级 */ public static final String REDIS_CALLBACK_QUEUE = "supermarket:shangyou:hotel:callback:queue"; + /** 已在队列中的 taskId(防重复 leftPush,避免 SUCCESS 副本每次启动再扫一遍) */ + public static final String REDIS_CALLBACK_QUEUED = "supermarket:shangyou:hotel:callback:queued"; + /** 回调高优先级队列:订单状态等,Worker 始终先消费本队列 */ public static final String REDIS_CALLBACK_QUEUE_HIGH = "supermarket:shangyou:hotel:callback:queue:high"; 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 2f8918f3..b7ca241b 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 @@ -50,8 +50,11 @@ public interface SmHotelCallbackTaskService { /** 清空全部回调任务,并清空 Redis 消费队列 */ R clearAllTasks(); - /** 消费一条队列任务(Worker 调用);返回 true=本次成功/跳过,false=失败或待再试 */ - boolean processQueuedTask(Long taskId); + /** + * 消费一条队列任务(Worker 调用)。 + * @return 1 已处理成功,0 跳过(非 PENDING / 不存在),-1 失败待重试 + */ + int processQueuedTask(Long taskId); /** 启动/人工:把全部 PENDING + 超时 RUNNING 重新入队 */ int requeueStalePending(); 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 5b02b50f..579337fa 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 @@ -63,6 +63,9 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic private final StringRedisTemplate redisTemplate; + /** Redis 残留 SUCCESS 副本跳过计数(避免启动刷屏) */ + private long skipNonPendingCount; + /** 与 Worker 一致:读渠道库表覆盖后的运行时,不要只读 YAML 默认值 */ private ShangyouHotelProperties props() { return channelConfigService.resolveShangyouRuntime(); @@ -311,6 +314,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic Long qHigh = redisTemplate.opsForList().size(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH); redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUE); redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH); + redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUED); log.warn("|回调任务|清空全部|deleted={}|queueCleared≈{}/high≈{}|", n, qLen, qHigh); return R.ok(n, "已清空 " + n + " 条任务,并清理队列" + (cnt != null && cnt > n ? "(处理中任务未删)" : "")); @@ -339,14 +343,14 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic } @Override - public boolean processQueuedTask(Long taskId) { + public int processQueuedTask(Long taskId) { if (taskId == null) { - return true; + return 0; } ShangyouHotelProperties p = props(); Long tenantId = p.getCallbackTenantId() == null ? 1L : p.getCallbackTenantId(); - Boolean ok = TenantBroker.applyAs(tenantId, tid -> doProcess(taskId)); - return Boolean.TRUE.equals(ok); + Integer r = TenantBroker.applyAs(tenantId, tid -> doProcess(taskId)); + return r == null ? -1 : r; } @Override @@ -433,16 +437,25 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic return n; } - /** @return true 成功或跳过;false 失败/待重试 */ - private boolean doProcess(Long taskId) { + /** @return 1 成功,0 跳过,-1 失败 */ + private int doProcess(Long taskId) { SmHotelCallbackTaskEntity task = taskMapper.selectById(taskId); + // 已从 List 弹出,去掉 queued 标记,失败重试才能再次入队 + unmarkQueued(taskId); if (task == null) { log.warn("|回调队列|任务不存在|taskId={}|", taskId); - return true; + return 0; } if (!STATUS_PENDING.equals(task.getStatus())) { - log.info("|回调队列|跳过非 PENDING|taskId={}|status={}|", taskId, task.getStatus()); - return true; + skipNonPendingCount++; + if (skipNonPendingCount == 1 || skipNonPendingCount % 200 == 0) { + log.info("|回调队列|跳过非 PENDING(Redis 残留已完成副本)|taskId={}|status={}|skipped={}|", + taskId, task.getStatus(), skipNonPendingCount); + } + else { + log.debug("|回调队列|跳过非 PENDING|taskId={}|status={}|", taskId, task.getStatus()); + } + return 0; } int attempts = task.getAttemptCount() == null ? 0 : task.getAttemptCount(); int max = Math.max(1, props().getCallbackMaxAttempts()); @@ -460,7 +473,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic : JSONUtil.parseObj(task.getReqJson()); Map result = orderService.handleOrderStatusNotify(body); finish(task, STATUS_SUCCESS, "订单状态更新: " + JSONUtil.toJsonStr(result), t0); - return true; + return 1; } if (task.getHotelId() == null && StrUtil.isNotBlank(task.getHotelCode())) { SmHotelEntity hotel = resolveWritableHotelByCode(task.getHotelCode(), SmHotelDataSource.SHANGYOU); @@ -472,7 +485,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic } if (task.getHotelId() == null) { finish(task, STATUS_SKIPPED, "无商游可写酒店(hotelCode=" + task.getHotelCode() + "),本地或其它渠道不会覆盖", t0); - return true; + return 1; } // 消费前把同酒店其余待处理房价/房量并进本条,积压的重复通知不再各拉一次 int folded = foldPendingSiblings(task); @@ -487,7 +500,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic SmHotelDataSource.denyWriteMessage(bound == null ? null : bound.getSheetName(), SmHotelDataSource.SHANGYOU), t0); - return true; + return 1; } // 回调只拉通知日期区间,避免每次全量 90 天把单线程队列堵死 R> pull = pullForCallback(task); @@ -495,11 +508,11 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic Object data = pull.getData(); finish(task, STATUS_SUCCESS, "拉取成功" + (data == null ? "" : ": " + StrUtil.maxLength(JSONUtil.toJsonStr(data), 800)), t0); - return true; + return 1; } String msg = pull == null ? "拉取返回空" : StrUtil.blankToDefault(pull.getMsg(), "拉取失败"); handleFailOrRetry(task, msg, t0, max); - return false; + return -1; } catch (Exception e) { String msg = StrUtil.maxLength(e.getMessage(), 800); @@ -511,7 +524,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic log.error("|回调队列|处理异常|taskId={}|", taskId, e); } handleFailOrRetry(task, msg, t0, max); - return false; + return -1; } } @@ -616,6 +629,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic task.setFinishTime(LocalDateTime.now()); task.setUpdateTime(LocalDateTime.now()); taskMapper.updateById(task); + unmarkQueued(task.getId()); log.info("|回调队列|完成|taskId={}|status={}|costMs={}|msg={}|", task.getId(), status, task.getCostMs(), task.getResultMsg()); } @@ -765,7 +779,14 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic /** 订单状态回调最高优先:进高优队列,Worker 始终先于房价/房量消费 */ private void enqueueTaskId(Long taskId, String notifyType) { + if (taskId == null) { + return; + } String id = String.valueOf(taskId); + Long added = redisTemplate.opsForSet().add(ShangyouHotelApi.REDIS_CALLBACK_QUEUED, id); + if (added != null && added == 0L) { + return; + } if (isHighPriority(notifyType)) { redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, id); } @@ -774,6 +795,18 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic } } + private void unmarkQueued(Long taskId) { + if (taskId == null) { + return; + } + try { + redisTemplate.opsForSet().remove(ShangyouHotelApi.REDIS_CALLBACK_QUEUED, String.valueOf(taskId)); + } + catch (Exception e) { + log.debug("|回调队列|清除 queued 标记失败|taskId={}|{}", taskId, e.getMessage()); + } + } + private static boolean isHighPriority(String notifyType) { return "ORDER_STATUS".equalsIgnoreCase(notifyType); }