酒店修复优化

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 → 异步拉取。
+ * <p>
+ * 另起看门狗线程：即使 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();
+		}
 	}
 }

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";

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 失效）。
-	 * <p>
-	 * 多线程保护：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 锁。规则：
+	 * <ul>
+	 * <li>缓存为空 → oauth</li>
+	 * <li>指定了 invalidToken 且缓存仍等于它 → oauth（必要时等冷却）</li>
+	 * <li>缓存已是别的票 → 直接复用</li>
+	 * <li>盲目刷新但冷却期内 → 复用</li>
+	 * </ul>
 	 */
-	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 {
 	 * <p>
 	 * 文档：业务码 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);

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.<SmHotelCallbackTaskEntity>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++;