From 8cd0f3bc3dc2c1ffaa804d7887713e2d9dee5caf Mon Sep 17 00:00:00 2001 From: zjf <1799650030@qq.com> Date: Wed, 30 Sep 2026 19:13:59 +0800 Subject: [PATCH] =?UTF-8?q?=E9=85=92=E5=BA=97=E4=BF=AE=E5=A4=8D=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../shangyou/ShangyouCallbackQueueWorker.java | 49 +++-- .../shangyou/ShangyouHotelApi.java | 1 + .../shangyou/ShangyouHotelClient.java | 190 ++++++++++++------ .../impl/SmHotelCallbackTaskServiceImpl.java | 17 +- 4 files changed, 176 insertions(+), 81 deletions(-) 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 ce51c4e8..1aa00406 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 @@ -14,8 +14,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; /** - * 消费 Redis 回调队列：BRPOP → 异步 pullFromUpstream。 - * 启动时把库里滞留的 PENDING 重新入队，避免重启丢任务。 + * 消费 Redis 回调队列：BRPOP → 异步拉取。 + *

+ * 另起看门狗线程：即使 Worker 卡在某次上游调用里，也能把超时 RUNNING 重置入队，避免整队假死。 */ @Slf4j @Component @@ -32,6 +33,8 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner { private Thread worker; + private Thread watchdog; + @Override public void run(ApplicationArguments args) { if (!properties.isCallbackQueueEnabled()) { @@ -47,28 +50,21 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner { worker = new Thread(this::loop, "shangyou-callback-queue"); worker.setDaemon(true); worker.start(); - log.info("|回调队列|Worker 已启动|key={}|", ShangyouHotelApi.REDIS_CALLBACK_QUEUE); + + watchdog = new Thread(this::watchdogLoop, "shangyou-callback-watchdog"); + watchdog.setDaemon(true); + watchdog.start(); + + log.info("|回调队列|Worker+Watchdog 已启动|key={}|", ShangyouHotelApi.REDIS_CALLBACK_QUEUE); } private void loop() { int consecutiveFail = 0; - long lastStaleCheckMs = 0L; while (running.get()) { try { String taskId = redisTemplate.opsForList() .rightPop(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, 5, TimeUnit.SECONDS); if (taskId == null || taskId.isBlank()) { - // 空闲时每分钟扫一次卡死的 RUNNING - long now = System.currentTimeMillis(); - if (now - lastStaleCheckMs >= 60_000L) { - lastStaleCheckMs = now; - try { - callbackTaskService.recoverStaleRunning(); - } - catch (Exception e) { - log.warn("|回调队列|恢复超时 RUNNING 失败|{}", e.getMessage()); - } - } continue; } Long id; @@ -120,11 +116,34 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner { } } + /** 不依赖 Worker 空闲：定时扫卡死的 RUNNING */ + private void watchdogLoop() { + while (running.get()) { + try { + Thread.sleep(30_000L); + if (!running.get()) { + break; + } + callbackTaskService.recoverStaleRunning(); + } + catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + break; + } + catch (Exception e) { + log.warn("|回调队列|Watchdog 恢复失败|{}", e.getMessage()); + } + } + } + @PreDestroy public void shutdown() { running.set(false); if (worker != null) { worker.interrupt(); } + if (watchdog != null) { + watchdog.interrupt(); + } } } 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 14f30d72..af39e08a 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 @@ -24,6 +24,7 @@ public final class ShangyouHotelApi { public static final String ORDER_QUERY_BY_REQUEST_NO = "/order/queryByRequestNo"; + /** 上游 access_token 缓存（全局单票；新票会使旧票失效，必须 compare-and-refresh） */ public static final String REDIS_TOKEN_KEY = "supermarket:shangyou:hotel:token"; public static final String REDIS_TOKEN_LOCK = "supermarket:shangyou:hotel:token:lock"; diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelClient.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelClient.java index 18bdc9c3..87c5e2b0 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelClient.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/integration/shangyou/ShangyouHotelClient.java @@ -62,10 +62,17 @@ public class ShangyouHotelClient { } /** - * 取 Token。forceRefresh=true 时在锁内换新（文档：新 Token 会使旧 Token 失效）。 - *

- * 多线程保护：Redis 锁互斥 oauth；抢不到锁只等待复用。传入 invalidToken 时，若缓存已是 - * 「别人刚换的新 Token」则直接复用，避免连环申请互踢。 + * 同 JVM 内互斥，避免多线程同时抢 Redis 锁 + oauth。 + * 上游约定：新 Token 会使旧 Token 立即失效，因此「只允许在缓存仍是失效 Token 时 oauth 一次」。 + */ + private final Object tokenRefreshMonitor = new Object(); + + /** 两次 oauth 最小间隔，防止 504 风暴下连环换票互踢 */ + private static final long OAUTH_MIN_INTERVAL_MS = 12_000L; + + /** + * 取 Token。forceRefresh=true 时按 invalidToken 做 compare-and-refresh： + * 缓存只要已经不是失效票，一律复用，绝不二次 oauth。 */ public String getAccessToken(boolean forceRefresh) { return getAccessToken(forceRefresh, null); @@ -73,95 +80,142 @@ public class ShangyouHotelClient { public String getAccessToken(boolean forceRefresh, String invalidToken) { ensureEnabled(); + String cached = readCachedToken(); if (!forceRefresh) { - String cached = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY); if (StrUtil.isNotBlank(cached)) { return cached; } } - Boolean locked = redisTemplate.opsForValue() - .setIfAbsent(ShangyouHotelApi.REDIS_TOKEN_LOCK, "1", 30, TimeUnit.SECONDS); - if (Boolean.TRUE.equals(locked)) { - try { - return refreshTokenUnderLock(forceRefresh, invalidToken); + else if (StrUtil.isNotBlank(invalidToken) && StrUtil.isNotBlank(cached) && !cached.equals(invalidToken)) { + // 别人（或上一轮）已经换过新票，本请求绝不能再 oauth，否则会踢掉正在用的新票 + log.info("shangyou reuse rotated token, skip oauth (invalid!=cache)"); + return cached; + } + else if (forceRefresh && StrUtil.isBlank(invalidToken) && StrUtil.isNotBlank(cached)) { + // 未指明失效票时不允许盲目换票 + return cached; + } + synchronized (tokenRefreshMonitor) { + return refreshTokenSynchronized(invalidToken); + } + } + + /** 已持本地锁：再抢 Redis 锁，双检后才 oauth */ + private String refreshTokenSynchronized(String invalidToken) { + // 进锁后再读一次，避免排队期间别人已换好 + String reused = tryReuseValidToken(invalidToken); + if (StrUtil.isNotBlank(reused)) { + return reused; + } + long lockTtlSec = Math.max(45L, properties.getTimeoutMs() / 1000L + 15L); + for (int attempt = 0; attempt < 4; attempt++) { + Boolean locked = redisTemplate.opsForValue() + .setIfAbsent(ShangyouHotelApi.REDIS_TOKEN_LOCK, "1", lockTtlSec, TimeUnit.SECONDS); + if (Boolean.TRUE.equals(locked)) { + try { + return refreshTokenUnderLock(invalidToken); + } + finally { + redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_LOCK); + } } - finally { - redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_LOCK); + String waited = waitForCachedToken(2_000L, invalidToken); + if (StrUtil.isNotBlank(waited)) { + log.info("shangyou wait lock got token from peer, skip oauth"); + return waited; } } - // 未抢到锁：等待他人刷新结果（若指定了失效 token，等到缓存变成另一个） - String waited = waitForCachedToken(10_000L, invalidToken); - if (StrUtil.isNotBlank(waited)) { - return waited; - } - Boolean locked2 = redisTemplate.opsForValue() - .setIfAbsent(ShangyouHotelApi.REDIS_TOKEN_LOCK, "1", 30, TimeUnit.SECONDS); - if (Boolean.TRUE.equals(locked2)) { - try { - return refreshTokenUnderLock(forceRefresh, invalidToken); - } - finally { - redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_LOCK); - } + String last = tryReuseValidToken(invalidToken); + if (StrUtil.isNotBlank(last)) { + return last; } throw new CheckedException("获取上游 Token 超时（刷新锁竞争）"); } /** - * 已持锁。forceRefresh 时：若缓存已不是 invalidToken（他人刚刷新），复用；否则 oauth 换新。 + * 已持 Redis 锁。规则： + *

*/ - private String refreshTokenUnderLock(boolean forceRefresh, String invalidToken) { - String cached = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY); - if (!forceRefresh && StrUtil.isNotBlank(cached)) { - return cached; - } - if (forceRefresh && StrUtil.isNotBlank(cached) && StrUtil.isNotBlank(invalidToken) - && !cached.equals(invalidToken) && wasTokenRefreshedRecently(5_000L)) { - log.info("shangyou reuse token refreshed by other thread, skip oauth"); - return cached; + private String refreshTokenUnderLock(String invalidToken) { + String cached = readCachedToken(); + if (StrUtil.isNotBlank(cached)) { + if (StrUtil.isBlank(invalidToken)) { + // 没有明确的失效票时，绝不主动 oauth（新票会踢旧票） + log.info("shangyou under-lock reuse cache (no invalidToken), skip oauth"); + return cached; + } + if (!cached.equals(invalidToken)) { + log.info("shangyou under-lock reuse rotated token, skip oauth"); + return cached; + } + if (wasTokenRefreshedRecently(OAUTH_MIN_INTERVAL_MS)) { + // 新票刚换出来又立刻 504：绝不能在请求线程睡满冷却（会堵死单线程回调队列）。 + // 直接抛出让上层任务失败入队尾，等冷却过后再由后续任务触发 oauth。 + long remain = oauthCooldownRemainMs(); + throw new CheckedException("上游 Token 刚刷新仍返回504，" + remain + + "ms 内不重复换票（避免互踢/堵队列），请稍后重试"); + } } return fetchAndCacheToken(); } + /** 缓存可用且（未指定失效票 / 已不是失效票）则复用 */ + private String tryReuseValidToken(String invalidToken) { + String cached = readCachedToken(); + if (StrUtil.isBlank(cached)) { + return null; + } + if (StrUtil.isBlank(invalidToken) || !cached.equals(invalidToken)) { + return cached; + } + return null; + } + + private String readCachedToken() { + return redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY); + } + private boolean wasTokenRefreshedRecently(long withinMs) { + return oauthCooldownRemainMs(withinMs) > 0L; + } + + private long oauthCooldownRemainMs() { + return oauthCooldownRemainMs(OAUTH_MIN_INTERVAL_MS); + } + + private long oauthCooldownRemainMs(long withinMs) { String at = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_REFRESHED_AT); if (StrUtil.isBlank(at)) { - return false; + return 0L; } try { long ts = Long.parseLong(at.trim()); - return System.currentTimeMillis() - ts <= withinMs; + long remain = withinMs - (System.currentTimeMillis() - ts); + return Math.max(0L, remain); } catch (Exception e) { - return false; + return 0L; } } /** - * 等待缓存出现 Token。若 invalidToken 非空，等到缓存值与其不同（说明已换新）。 + * 等待缓存出现「可用」Token：非空，且若指定了 invalidToken 则必须已换成另一张。 */ private String waitForCachedToken(long timeoutMs, String invalidToken) { long deadline = System.currentTimeMillis() + Math.max(1000L, timeoutMs); while (System.currentTimeMillis() < deadline) { - String cached = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY); - if (StrUtil.isNotBlank(cached) - && (StrUtil.isBlank(invalidToken) || !cached.equals(invalidToken))) { - return cached; - } - try { - Thread.sleep(150L); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return null; + String ok = tryReuseValidToken(invalidToken); + if (StrUtil.isNotBlank(ok)) { + return ok; } + sleepQuiet(150L); } - String cached = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY); - if (StrUtil.isNotBlank(cached) - && (StrUtil.isBlank(invalidToken) || !cached.equals(invalidToken))) { - return cached; - } - return null; + return tryReuseValidToken(invalidToken); } private String fetchAndCacheToken() { @@ -198,6 +252,8 @@ public class ShangyouHotelClient { redisTemplate.opsForValue() .set(ShangyouHotelApi.REDIS_TOKEN_REFRESHED_AT, String.valueOf(System.currentTimeMillis()), ttl, TimeUnit.SECONDS); + log.info("shangyou oauth ok, token cached ttlSec={} tokenTail={}", ttl, + token.length() > 8 ? token.substring(token.length() - 8) : token); return token; } } @@ -718,6 +774,7 @@ public class ShangyouHotelClient { *

* 文档：业务码 504 = Token 缓存失效，必须重新获取。仅 code=500「繁忙」才按繁忙退避。 * 多线程换 Token 走 Redis 锁，避免并发 oauth 互踢。 + * Token 冷却期内遇 504 会快速失败（不睡满冷却），避免单线程回调队列被堵死。 */ private JSONObject postBiz(String path, JSONObject body, int retryRound, boolean forceToken, String invalidToken) { @@ -742,9 +799,14 @@ public class ShangyouHotelClient { boolean tokenBizFail = isTokenExpiredBiz(respBody); if (httpUnauthorized || tokenBizFail) { if (retryRound < 4) { - sleepQuiet(httpUnauthorized ? 200L : 400L); - log.warn("shangyou 504/token expired, refresh path={} nextRound={} http401={} snippet={}", path, - retryRound + 1, httpUnauthorized, StrUtil.sub(respBody, 0, 200)); + // 先看缓存是否已换新：有则立刻用新票重试，避免本线程再 oauth 互踢 + String cachedNow = readCachedToken(); + boolean alreadyRotated = StrUtil.isNotBlank(cachedNow) && !cachedNow.equals(token); + long waitMs = alreadyRotated ? 50L : (httpUnauthorized ? 200L : 400L); + sleepQuiet(waitMs); + log.warn( + "shangyou 504/token expired, refresh path={} nextRound={} http401={} alreadyRotated={} snippet={}", + path, retryRound + 1, httpUnauthorized, alreadyRotated, StrUtil.sub(respBody, 0, 200)); return postBiz(path, body, retryRound + 1, true, token); } throw new CheckedException("上游 Token 失效且重试仍失败 path=" + path + " body=" + respBody); @@ -756,9 +818,11 @@ public class ShangyouHotelClient { JSONObject resp = JSON.parseObject(respBody); if (isTokenExpiredBiz(resp)) { if (retryRound < 4) { - sleepQuiet(400L); - log.warn("shangyou biz 504/token expired, refresh path={} nextRound={} msg={}", path, - retryRound + 1, resp.getString("message")); + String cachedNow = readCachedToken(); + boolean alreadyRotated = StrUtil.isNotBlank(cachedNow) && !cachedNow.equals(token); + sleepQuiet(alreadyRotated ? 50L : 400L); + log.warn("shangyou biz 504/token expired, refresh path={} nextRound={} alreadyRotated={} msg={}", + path, retryRound + 1, alreadyRotated, resp.getString("message")); return postBiz(path, body, retryRound + 1, true, token); } throw new CheckedException("上游 Token 失效且重试仍失败 path=" + path + " body=" + respBody); 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 95ad633b..f0867d61 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 @@ -181,9 +181,11 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic return TenantBroker.applyAs(tenantId, tid -> resetStaleRunningLocked()); } - /** 超时 RUNNING → PENDING 并入队尾 */ + /** 超时 RUNNING → PENDING 并入队尾；已达最大次数则直接 FAILED，不再堵队列 */ private int resetStaleRunningLocked() { - LocalDateTime staleBefore = LocalDateTime.now().minusMinutes(10); + // 单日区间拉取正常应秒级完成；2 分钟仍 RUNNING 视为卡死（Token 冷却/进程挂起等） + LocalDateTime staleBefore = LocalDateTime.now().minusMinutes(2); + int max = Math.max(1, properties.getCallbackMaxAttempts()); var running = taskMapper.selectList(Wrappers.lambdaQuery() .eq(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING) .and(w -> w.lt(SmHotelCallbackTaskEntity::getStartProcessTime, staleBefore) @@ -194,9 +196,18 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic .last("limit 100")); int n = 0; for (SmHotelCallbackTaskEntity t : running) { + int attempts = t.getAttemptCount() == null ? 0 : t.getAttemptCount(); + t.setUpdateTime(LocalDateTime.now()); + if (attempts >= max) { + t.setStatus(STATUS_FAILED); + t.setFinishTime(LocalDateTime.now()); + t.setResultMsg("超时仍 RUNNING 且已达最大次数(" + attempts + "/" + max + ")，自动失败以免堵队列"); + taskMapper.updateById(t); + n++; + continue; + } t.setStatus(STATUS_PENDING); t.setResultMsg("超时 RUNNING 已重置为 PENDING 并重新入队"); - t.setUpdateTime(LocalDateTime.now()); taskMapper.updateById(t); redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId())); n++;