调整
This commit is contained in:
1 parent
a3c7699716
commit
9013f51e2a
16 files changed
+740
-151
No files matched your search
+15
@@ -66,4 +66,19 @@ public class CanteenHotelEntity extends Model<CanteenHotelEntity> {
|
||||
@TableLogic
|
||||
@TableField(fill = FieldFill.INSERT)
|
||||
private String delFlag;
|
||||
|
||||
/** 列表展示:房型数量(非表字段) */
|
||||
@TableField(exist = false)
|
||||
@Schema(description = "房型数量")
|
||||
private Integer roomCount;
|
||||
|
||||
/** 列表展示:产品数量(非表字段) */
|
||||
@TableField(exist = false)
|
||||
@Schema(description = "产品数量")
|
||||
private Integer productCount;
|
||||
|
||||
/** 列表展示:日历条数(非表字段) */
|
||||
@TableField(exist = false)
|
||||
@Schema(description = "日历数量")
|
||||
private Integer dailyCount;
|
||||
}
|
||||
+19
-2
@@ -93,6 +93,20 @@ public class ShangyouHotelProperties {
|
||||
/** 单条回调任务最大尝试次数(含首次) */
|
||||
private int callbackMaxAttempts = 3;
|
||||
|
||||
/** 回调队列每处理完一家后的间隔毫秒(有 price-api-qps 限流时可略小;默认 50) */
|
||||
private int callbackTaskIntervalMs = 50;
|
||||
|
||||
/** 连续失败达到该次数后,额外冷却毫秒(默认 5 次后冷却 15 秒) */
|
||||
private int callbackFailPauseAfter = 5;
|
||||
|
||||
private int callbackFailPauseMs = 15000;
|
||||
|
||||
/**
|
||||
* 房价接口 getRoomPrice 限流(对方约定约 20 QPS)。
|
||||
* ≤0 表示不限制;多实例部署时请按实例数均分(如 2 实例各配 10)。
|
||||
*/
|
||||
private double priceApiQps = 20;
|
||||
|
||||
// ---------- 日更 / 差集(可配置)----------
|
||||
|
||||
/** 酒店列表日同步:是否允许定时/XXL 执行(false 则跳过) */
|
||||
@@ -131,14 +145,17 @@ public class ShangyouHotelProperties {
|
||||
/** 滚价每批处理本地酒店数(按 id 游标,不装全表) */
|
||||
private int priceRollBatchSize = 100;
|
||||
|
||||
/** 滚价每家间隔毫秒,减轻上游限流/繁忙(默认 30) */
|
||||
private int priceRollIntervalMs = 30;
|
||||
|
||||
/** 文档:全量房价房量每天最多次数 */
|
||||
private int fullPriceSyncMaxPerDay = 3;
|
||||
|
||||
/** 文档:全量房价两次间隔最少小时数 */
|
||||
private int fullPriceSyncMinIntervalHours = 6;
|
||||
|
||||
/** 全量拉详情:每家酒店间隔毫秒(防打爆上游;落库已批量,默认 0) */
|
||||
private int detailPullIntervalMs = 0;
|
||||
/** 全量拉详情:每家酒店间隔毫秒(防打爆上游;落库已批量,默认 50) */
|
||||
private int detailPullIntervalMs = 50;
|
||||
|
||||
/** 全量拉详情单次最多处理家数;0=不限制(慎用,酒店多时耗时长) */
|
||||
private int detailPullMaxHotels = 0;
|
||||
|
||||
+18
@@ -194,4 +194,22 @@ public class CanteenHotelMidController {
|
||||
data.put("message", "已提交,请轮询 /mid/syncLog/{id} 或打开同步日志");
|
||||
return R.ok(data, started.getMsg());
|
||||
}
|
||||
|
||||
@SysLog("重试同步失败酒店")
|
||||
@Operation(summary = "根据某次同步日志的失败明细,只重试失败酒店(详情/滚价)")
|
||||
@PostMapping("/mid/retryFailedFromLog")
|
||||
@PreAuthorize("@pms.hasPermission('canteen_canteenHotel_edit')")
|
||||
public R<Map<String, Object>> retryFailedFromLog(@RequestParam Long logId) {
|
||||
R<CanteenHotelSyncLogEntity> started = syncService.startRetryFailedAsync(logId);
|
||||
if (started.getCode() != 0) {
|
||||
return R.failed(started.getMsg());
|
||||
}
|
||||
Map<String, Object> data = new LinkedHashMap<>();
|
||||
data.put("async", true);
|
||||
data.put("logId", started.getData().getId());
|
||||
data.put("status", started.getData().getStatus());
|
||||
data.put("sourceLogId", logId);
|
||||
data.put("message", started.getMsg());
|
||||
return R.ok(data, started.getMsg());
|
||||
}
|
||||
}
|
||||
+25
-1
@@ -51,6 +51,7 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
|
||||
}
|
||||
|
||||
private void loop() {
|
||||
int consecutiveFail = 0;
|
||||
while (running.get()) {
|
||||
try {
|
||||
String taskId = redisTemplate.opsForList()
|
||||
@@ -66,7 +67,30 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
|
||||
log.warn("|回调队列|非法 taskId={}|", taskId);
|
||||
continue;
|
||||
}
|
||||
callbackTaskService.processQueuedTask(id);
|
||||
boolean ok = callbackTaskService.processQueuedTask(id);
|
||||
int interval = Math.max(0, properties.getCallbackTaskIntervalMs());
|
||||
if (interval > 0) {
|
||||
Thread.sleep(interval);
|
||||
}
|
||||
if (ok) {
|
||||
consecutiveFail = 0;
|
||||
}
|
||||
else {
|
||||
consecutiveFail++;
|
||||
int pauseAfter = Math.max(1, properties.getCallbackFailPauseAfter());
|
||||
if (consecutiveFail >= pauseAfter) {
|
||||
int pauseMs = Math.max(0, properties.getCallbackFailPauseMs());
|
||||
log.warn("|回调队列|连续失败 {} 次,冷却 {}ms 后再继续|", consecutiveFail, pauseMs);
|
||||
if (pauseMs > 0) {
|
||||
Thread.sleep(pauseMs);
|
||||
}
|
||||
consecutiveFail = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (InterruptedException ie) {
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
}
|
||||
catch (Exception e) {
|
||||
if (!running.get()) {
|
||||
|
||||
+3
@@ -57,4 +57,7 @@ public final class ShangyouHotelApi {
|
||||
|
||||
/** 上次全量房价同步时间 epochMs */
|
||||
public static final String REDIS_FULL_PRICE_SYNC_LAST = "canteen:shangyou:hotel:fullprice:last";
|
||||
|
||||
/** 房价接口 QPS 秒窗计数前缀,完整 key = 前缀 + ":" + epochSecond */
|
||||
public static final String REDIS_PRICE_QPS_PREFIX = "canteen:shangyou:hotel:price:qps";
|
||||
}
|
||||
+189
-20
@@ -61,6 +61,12 @@ public class ShangyouHotelClient {
|
||||
return getAccessToken(false);
|
||||
}
|
||||
|
||||
/**
|
||||
* 取 Token。forceRefresh=true 时先删缓存再换新。
|
||||
* <p>
|
||||
* 用 Redis 锁互斥刷新;抢不到锁时等待其他线程写好缓存,避免并发各自去打 oauth、
|
||||
* 或「删掉旧 Token 后还没写上新 Token」窗口里读到空/脏值。
|
||||
*/
|
||||
public String getAccessToken(boolean forceRefresh) {
|
||||
ensureEnabled();
|
||||
if (!forceRefresh) {
|
||||
@@ -74,20 +80,60 @@ public class ShangyouHotelClient {
|
||||
}
|
||||
Boolean locked = redisTemplate.opsForValue()
|
||||
.setIfAbsent(ShangyouHotelApi.REDIS_TOKEN_LOCK, "1", 30, TimeUnit.SECONDS);
|
||||
try {
|
||||
if (!forceRefresh) {
|
||||
if (Boolean.TRUE.equals(locked)) {
|
||||
try {
|
||||
if (!forceRefresh) {
|
||||
String cached = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
if (StrUtil.isNotBlank(cached)) {
|
||||
return cached;
|
||||
}
|
||||
}
|
||||
return fetchAndCacheToken();
|
||||
}
|
||||
finally {
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_LOCK);
|
||||
}
|
||||
}
|
||||
// 未抢到锁:等持锁线程把 Token 写入 Redis
|
||||
String waited = waitForCachedToken(10_000L);
|
||||
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 {
|
||||
String cached = redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
if (StrUtil.isNotBlank(cached)) {
|
||||
return cached;
|
||||
}
|
||||
return fetchAndCacheToken();
|
||||
}
|
||||
return fetchAndCacheToken();
|
||||
}
|
||||
finally {
|
||||
if (Boolean.TRUE.equals(locked)) {
|
||||
finally {
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_LOCK);
|
||||
}
|
||||
}
|
||||
throw new CheckedException("获取上游 Token 超时(刷新锁竞争)");
|
||||
}
|
||||
|
||||
/** 轮询 Redis 直到出现 Token 或超时 */
|
||||
private String waitForCachedToken(long timeoutMs) {
|
||||
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)) {
|
||||
return cached;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(150L);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
return redisTemplate.opsForValue().get(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
}
|
||||
|
||||
private String fetchAndCacheToken() {
|
||||
@@ -378,6 +424,7 @@ public class ShangyouHotelClient {
|
||||
|
||||
/**
|
||||
* 房价房量。返回完整上游响应(含 code/message/content)。
|
||||
* QPS 限流在 postBiz 内按次生效(含重试)。
|
||||
*/
|
||||
public JSONObject getRoomPriceRaw(String hotelId, String checkInDate, String checkOutDate,
|
||||
List<String> roomTypeIds) {
|
||||
@@ -399,6 +446,57 @@ public class ShangyouHotelClient {
|
||||
return postBiz(path, body);
|
||||
}
|
||||
|
||||
/**
|
||||
* 房价接口限流:本地匀速 + Redis 秒窗硬顶(多实例共享)。
|
||||
* 对方约定约 20 QPS;配置 price-api-qps≤0 关闭。
|
||||
*/
|
||||
private void acquirePriceApiPermit() {
|
||||
double qps = properties.getPriceApiQps();
|
||||
if (qps <= 0) {
|
||||
return;
|
||||
}
|
||||
// 1) 本进程匀速:约 1000/qps 毫秒一把
|
||||
long intervalMs = Math.max(1L, Math.round(1000.0 / qps));
|
||||
synchronized (this) {
|
||||
long now = System.currentTimeMillis();
|
||||
if (priceNextPermitAtMs > now) {
|
||||
sleepQuiet(priceNextPermitAtMs - now);
|
||||
now = System.currentTimeMillis();
|
||||
}
|
||||
priceNextPermitAtMs = Math.max(now, priceNextPermitAtMs) + intervalMs;
|
||||
}
|
||||
// 2) Redis 秒窗:跨实例不超过 ceil(qps)(配置已按实例均分时更稳)
|
||||
int maxPerSec = Math.max(1, (int) Math.ceil(qps));
|
||||
for (int i = 0; i < 30; i++) {
|
||||
long sec = System.currentTimeMillis() / 1000L;
|
||||
String key = ShangyouHotelApi.REDIS_PRICE_QPS_PREFIX + ":" + sec;
|
||||
Long n = redisTemplate.opsForValue().increment(key);
|
||||
if (n != null && n == 1L) {
|
||||
redisTemplate.expire(key, 3, TimeUnit.SECONDS);
|
||||
}
|
||||
if (n == null || n <= maxPerSec) {
|
||||
return;
|
||||
}
|
||||
// 当前秒已满,等到下一秒
|
||||
long wait = 1000L - (System.currentTimeMillis() % 1000L) + 5L;
|
||||
sleepQuiet(wait);
|
||||
}
|
||||
log.warn("shangyou price QPS wait timeout, continue anyway qps={}", qps);
|
||||
}
|
||||
|
||||
/** 本进程房价下次可发时间 */
|
||||
private volatile long priceNextPermitAtMs = 0L;
|
||||
|
||||
private boolean isRoomPricePath(String path) {
|
||||
if (StrUtil.isBlank(path)) {
|
||||
return false;
|
||||
}
|
||||
String configured = StrUtil.blankToDefault(properties.getRoomPricePath(), "/hotel/getRoomPrice");
|
||||
String p = path.startsWith("/") ? path : "/" + path;
|
||||
String c = configured.startsWith("/") ? configured : "/" + configured;
|
||||
return p.equalsIgnoreCase(c) || p.toLowerCase().contains("getroomprice");
|
||||
}
|
||||
|
||||
public JSONObject getRoomPrice(String hotelId, String checkInDate, String checkOutDate,
|
||||
List<String> roomTypeIds) {
|
||||
JSONObject resp = getRoomPriceRaw(hotelId, checkInDate, checkOutDate, roomTypeIds);
|
||||
@@ -578,17 +676,25 @@ public class ShangyouHotelClient {
|
||||
}
|
||||
|
||||
private JSONObject postBiz(String path, JSONObject body) {
|
||||
return postBiz(path, body, false);
|
||||
return postBiz(path, body, 0, false);
|
||||
}
|
||||
|
||||
/**
|
||||
* 业务调用。文档约定业务码 504 = token 失效,需重新取 token 后再调一次。
|
||||
* 业务调用。
|
||||
* <p>
|
||||
* 文档:504=token 失效。但 getRoomPrice 高峰常把限流也打成 504(同一 Token 刚调通
|
||||
* getHotelComplexList 却对房价接口报 504)。策略:先同 Token 退避重试,再强制刷新,
|
||||
* 繁忙同样退避;最多 4 轮。
|
||||
*/
|
||||
private JSONObject postBiz(String path, JSONObject body, boolean retried) {
|
||||
private JSONObject postBiz(String path, JSONObject body, int retryRound, boolean forceToken) {
|
||||
ensureEnabled();
|
||||
String token = getAccessToken(retried);
|
||||
if (isRoomPricePath(path)) {
|
||||
acquirePriceApiPermit();
|
||||
}
|
||||
String token = getAccessToken(forceToken);
|
||||
String url = properties.resolveBizBaseUrl() + path;
|
||||
log.info("shangyou request path={} retried={} bodyKeys={}", path, retried, body.keySet());
|
||||
log.info("shangyou request path={} retryRound={} forceToken={} bodyKeys={}", path, retryRound, forceToken,
|
||||
body.keySet());
|
||||
try (HttpResponse response = HttpRequest.post(url)
|
||||
.header("Authorization", "Bearer " + token)
|
||||
.header("Accept", "*/*")
|
||||
@@ -598,11 +704,25 @@ public class ShangyouHotelClient {
|
||||
.timeout(properties.getTimeoutMs())
|
||||
.execute()) {
|
||||
String respBody = response.body();
|
||||
if (response.getStatus() == 401 || isTokenExpiredBiz(respBody)) {
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
if (!retried) {
|
||||
log.warn("shangyou token expired, refresh and retry path={}", path);
|
||||
return postBiz(path, body, true);
|
||||
boolean httpUnauthorized = response.getStatus() == 401;
|
||||
boolean tokenBizFail = isTokenExpiredBiz(respBody);
|
||||
if (httpUnauthorized || tokenBizFail) {
|
||||
if (retryRound < 4) {
|
||||
boolean refreshNow = httpUnauthorized || retryRound >= 1;
|
||||
if (refreshNow) {
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
}
|
||||
long waitMs = switch (retryRound) {
|
||||
case 0 -> 400L;
|
||||
case 1 -> 1000L;
|
||||
case 2 -> 2000L;
|
||||
default -> 4000L;
|
||||
};
|
||||
sleepQuiet(waitMs);
|
||||
log.warn(
|
||||
"shangyou token/limit code path={} nextRound={} refreshToken={} http401={} snippet={}",
|
||||
path, retryRound + 1, refreshNow, httpUnauthorized, StrUtil.sub(respBody, 0, 200));
|
||||
return postBiz(path, body, retryRound + 1, refreshNow);
|
||||
}
|
||||
throw new CheckedException("上游 Token 失效且重试仍失败 path=" + path + " body=" + respBody);
|
||||
}
|
||||
@@ -612,13 +732,37 @@ public class ShangyouHotelClient {
|
||||
}
|
||||
JSONObject resp = JSON.parseObject(respBody);
|
||||
if (isTokenExpiredBiz(resp)) {
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
if (!retried) {
|
||||
log.warn("shangyou biz code token invalid, refresh and retry path={}", path);
|
||||
return postBiz(path, body, true);
|
||||
if (retryRound < 4) {
|
||||
boolean refreshNow = retryRound >= 1;
|
||||
if (refreshNow) {
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_TOKEN_KEY);
|
||||
}
|
||||
long waitMs = switch (retryRound) {
|
||||
case 0 -> 400L;
|
||||
case 1 -> 1000L;
|
||||
case 2 -> 2000L;
|
||||
default -> 4000L;
|
||||
};
|
||||
sleepQuiet(waitMs);
|
||||
log.warn("shangyou biz token/limit path={} nextRound={} refreshToken={} msg={}", path,
|
||||
retryRound + 1, refreshNow, resp.getString("message"));
|
||||
return postBiz(path, body, retryRound + 1, refreshNow);
|
||||
}
|
||||
throw new CheckedException("上游 Token 失效且重试仍失败 path=" + path + " body=" + respBody);
|
||||
}
|
||||
// 上游瞬时繁忙:短暂退避后重试(不刷新 Token)
|
||||
if (isUpstreamBusy(resp) && retryRound < 4) {
|
||||
long waitMs = switch (retryRound) {
|
||||
case 0 -> 800L;
|
||||
case 1 -> 1500L;
|
||||
case 2 -> 3000L;
|
||||
default -> 5000L;
|
||||
};
|
||||
sleepQuiet(waitMs);
|
||||
log.warn("shangyou busy, retry path={} nextRound={} msg={}", path, retryRound + 1,
|
||||
resp.getString("message"));
|
||||
return postBiz(path, body, retryRound + 1, false);
|
||||
}
|
||||
ensureBizOk(resp, path);
|
||||
// 酒店列表很大,只打数量,避免日志/字符串截取拖慢
|
||||
if (ShangyouHotelApi.HOTEL_LIST.equals(path)) {
|
||||
@@ -663,6 +807,31 @@ public class ShangyouHotelClient {
|
||||
|| StrUtil.containsIgnoreCase(msg, "失效") || StrUtil.containsIgnoreCase(msg, "为空"));
|
||||
}
|
||||
|
||||
/** 上游瞬时繁忙(滚价高峰常见) */
|
||||
private static boolean isUpstreamBusy(JSONObject resp) {
|
||||
if (resp == null) {
|
||||
return false;
|
||||
}
|
||||
String code = firstNonBlank(resp.getString("code"), resp.getString("Code"));
|
||||
String msg = firstNonBlank(resp.getString("message"), resp.getString("msg"));
|
||||
if ("500".equals(code) && StrUtil.containsIgnoreCase(msg, "繁忙")) {
|
||||
return true;
|
||||
}
|
||||
return StrUtil.containsIgnoreCase(msg, "请稍后再试") || StrUtil.containsIgnoreCase(msg, "稍候再试");
|
||||
}
|
||||
|
||||
private static void sleepQuiet(long ms) {
|
||||
if (ms <= 0) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(ms);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
private void ensureBizOk(JSONObject resp, String path) {
|
||||
if (resp == null) {
|
||||
throw new CheckedException("上游返回空 path=" + path);
|
||||
|
||||
+13
@@ -5,9 +5,12 @@ import com.bsd.pjcloud.common.data.datascope.PjcloudBaseMapper;
|
||||
import org.apache.ibatis.annotations.Delete;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@Mapper
|
||||
public interface CanteenHotelDailyMapper extends PjcloudBaseMapper<CanteenHotelDailyEntity> {
|
||||
@@ -19,4 +22,14 @@ public interface CanteenHotelDailyMapper extends PjcloudBaseMapper<CanteenHotelD
|
||||
int physicalDeleteFromDate(@Param("hotelId") Long hotelId, @Param("fromDate") LocalDate fromDate);
|
||||
|
||||
int insertBatch(@Param("list") List<CanteenHotelDailyEntity> list);
|
||||
|
||||
@Select("""
|
||||
<script>
|
||||
SELECT hotel_id AS hotelId, COUNT(1) AS cnt FROM canteen_hotel_daily
|
||||
WHERE del_flag = '0' AND hotel_id IN
|
||||
<foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
|
||||
GROUP BY hotel_id
|
||||
</script>
|
||||
""")
|
||||
List<Map<String, Object>> countGroupByHotelIds(@Param("ids") Collection<Long> ids);
|
||||
}
|
||||
+13
@@ -4,11 +4,24 @@ import com.bsd.pjcloud.canteen.entity.CanteenHotelProductEntity;
|
||||
import com.bsd.pjcloud.common.data.datascope.PjcloudBaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@Mapper
|
||||
public interface CanteenHotelProductMapper extends PjcloudBaseMapper<CanteenHotelProductEntity> {
|
||||
|
||||
int insertBatch(@Param("list") List<CanteenHotelProductEntity> list);
|
||||
|
||||
@Select("""
|
||||
<script>
|
||||
SELECT hotel_id AS hotelId, COUNT(1) AS cnt FROM canteen_hotel_product
|
||||
WHERE del_flag = '0' AND hotel_id IN
|
||||
<foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
|
||||
GROUP BY hotel_id
|
||||
</script>
|
||||
""")
|
||||
List<Map<String, Object>> countGroupByHotelIds(@Param("ids") Collection<Long> ids);
|
||||
}
|
||||
+13
@@ -4,11 +4,24 @@ import com.bsd.pjcloud.canteen.entity.CanteenHotelRoomEntity;
|
||||
import com.bsd.pjcloud.common.data.datascope.PjcloudBaseMapper;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@Mapper
|
||||
public interface CanteenHotelRoomMapper extends PjcloudBaseMapper<CanteenHotelRoomEntity> {
|
||||
|
||||
int insertBatch(@Param("list") List<CanteenHotelRoomEntity> list);
|
||||
|
||||
@Select("""
|
||||
<script>
|
||||
SELECT hotel_id AS hotelId, COUNT(1) AS cnt FROM canteen_hotel_room
|
||||
WHERE del_flag = '0' AND hotel_id IN
|
||||
<foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
|
||||
GROUP BY hotel_id
|
||||
</script>
|
||||
""")
|
||||
List<Map<String, Object>> countGroupByHotelIds(@Param("ids") Collection<Long> ids);
|
||||
}
|
||||
+2
-2
@@ -26,8 +26,8 @@ public interface CanteenHotelCallbackTaskService {
|
||||
/** 失败/跳过任务重新入队 */
|
||||
R<Boolean> retry(Long id);
|
||||
|
||||
/** 消费一条队列任务(Worker 调用) */
|
||||
void processQueuedTask(Long taskId);
|
||||
/** 消费一条队列任务(Worker 调用);返回 true=本次成功/跳过,false=失败或待再试 */
|
||||
boolean processQueuedTask(Long taskId);
|
||||
|
||||
/** 启动时把滞留 PENDING 重新入队 */
|
||||
int requeueStalePending();
|
||||
|
||||
+5
@@ -39,6 +39,11 @@ public interface CanteenHotelSyncService {
|
||||
*/
|
||||
R<CanteenHotelSyncLogEntity> startDetailPullAsync(Integer limit, String triggerType);
|
||||
|
||||
/**
|
||||
* 根据某次同步日志的 failDetail,只重试失败酒店(支持 DETAIL_PULL / PRICE_ROLL)。
|
||||
*/
|
||||
R<CanteenHotelSyncLogEntity> startRetryFailedAsync(Long sourceLogId);
|
||||
|
||||
/** 兼容旧同步入口(内部转异步) */
|
||||
R<Map<String, Object>> syncHotelListAndDiff(boolean autoImport);
|
||||
|
||||
|
||||
+19
-14
@@ -140,12 +140,13 @@ public class CanteenHotelCallbackTaskServiceImpl implements CanteenHotelCallback
|
||||
}
|
||||
|
||||
@Override
|
||||
public void processQueuedTask(Long taskId) {
|
||||
public boolean processQueuedTask(Long taskId) {
|
||||
if (taskId == null) {
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
|
||||
TenantBroker.runAs(tenantId, tid -> doProcess(taskId));
|
||||
Boolean ok = TenantBroker.applyAs(tenantId, tid -> doProcess(taskId));
|
||||
return Boolean.TRUE.equals(ok);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -158,7 +159,8 @@ public class CanteenHotelCallbackTaskServiceImpl implements CanteenHotelCallback
|
||||
.last("limit 200"));
|
||||
int n = 0;
|
||||
for (CanteenHotelCallbackTaskEntity t : list) {
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId()));
|
||||
// 队尾恢复,避免一启动就顶着打
|
||||
redisTemplate.opsForList().rightPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId()));
|
||||
n++;
|
||||
}
|
||||
if (n > 0) {
|
||||
@@ -168,15 +170,16 @@ public class CanteenHotelCallbackTaskServiceImpl implements CanteenHotelCallback
|
||||
});
|
||||
}
|
||||
|
||||
private void doProcess(Long taskId) {
|
||||
/** @return true 成功或跳过;false 失败/待重试 */
|
||||
private boolean doProcess(Long taskId) {
|
||||
CanteenHotelCallbackTaskEntity task = taskMapper.selectById(taskId);
|
||||
if (task == null) {
|
||||
log.warn("|回调队列|任务不存在|taskId={}|", taskId);
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
if (!STATUS_PENDING.equals(task.getStatus())) {
|
||||
log.info("|回调队列|跳过非 PENDING|taskId={}|status={}|", taskId, task.getStatus());
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
int attempts = task.getAttemptCount() == null ? 0 : task.getAttemptCount();
|
||||
int max = Math.max(1, properties.getCallbackMaxAttempts());
|
||||
@@ -193,7 +196,7 @@ public class CanteenHotelCallbackTaskServiceImpl implements CanteenHotelCallback
|
||||
: JSONUtil.parseObj(task.getReqJson());
|
||||
Map<String, Object> result = orderService.handleOrderStatusNotify(body);
|
||||
finish(task, STATUS_SUCCESS, "订单状态更新: " + JSONUtil.toJsonStr(result), t0);
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
if (task.getHotelId() == null && StrUtil.isNotBlank(task.getHotelCode())) {
|
||||
CanteenHotelEntity hotel = hotelMapper.selectOne(Wrappers.<CanteenHotelEntity>lambdaQuery()
|
||||
@@ -207,22 +210,23 @@ public class CanteenHotelCallbackTaskServiceImpl implements CanteenHotelCallback
|
||||
}
|
||||
if (task.getHotelId() == null) {
|
||||
finish(task, STATUS_SKIPPED, "本地无匹配酒店(hotelCode=" + task.getHotelCode() + "),请先导入/授权", t0);
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
R<Map<String, Object>> pull = midService.pullFromUpstream(task.getHotelId());
|
||||
if (pull != null && pull.getCode() == 0) {
|
||||
Object data = pull.getData();
|
||||
finish(task, STATUS_SUCCESS,
|
||||
"拉取成功" + (data == null ? "" : ": " + StrUtil.maxLength(JSONUtil.toJsonStr(data), 800)), t0);
|
||||
return true;
|
||||
}
|
||||
else {
|
||||
String msg = pull == null ? "拉取返回空" : StrUtil.blankToDefault(pull.getMsg(), "拉取失败");
|
||||
handleFailOrRetry(task, msg, t0, max);
|
||||
}
|
||||
String msg = pull == null ? "拉取返回空" : StrUtil.blankToDefault(pull.getMsg(), "拉取失败");
|
||||
handleFailOrRetry(task, msg, t0, max);
|
||||
return false;
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("|回调队列|处理异常|taskId={}|", taskId, e);
|
||||
handleFailOrRetry(task, StrUtil.maxLength(e.getMessage(), 800), t0, max);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -234,7 +238,8 @@ public class CanteenHotelCallbackTaskServiceImpl implements CanteenHotelCallback
|
||||
task.setCostMs(System.currentTimeMillis() - t0);
|
||||
task.setUpdateTime(LocalDateTime.now());
|
||||
taskMapper.updateById(task);
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(task.getId()));
|
||||
// 放到队尾,避免刚失败立刻再打同一家(原先 leftPush 会顶到队头)
|
||||
redisTemplate.opsForList().rightPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(task.getId()));
|
||||
return;
|
||||
}
|
||||
finish(task, STATUS_FAILED, msg, t0);
|
||||
|
||||
+58
-12
@@ -776,8 +776,10 @@ public class CanteenHotelOrderServiceImpl extends ServiceImpl<CanteenHotelOrderM
|
||||
req.put("checkOutDate", order.getCheckOutDate() == null ? null : order.getCheckOutDate().toString());
|
||||
req.put("roomCount", order.getCount() == null ? 1 : order.getCount());
|
||||
|
||||
// 文档必填:联系人用 linkFirstName / linkMobile(不是 linkName)
|
||||
req.put("linkFirstName", order.getContactName());
|
||||
// 文档必填:联系人 linkLastName + linkFirstName(中文须拆姓/名,不能整名塞 FirstName)
|
||||
NameParts link = splitPersonName(order.getContactName());
|
||||
req.put("linkLastName", link.lastName());
|
||||
req.put("linkFirstName", link.firstName());
|
||||
req.put("linkMobile", order.getContactPhone());
|
||||
|
||||
boolean isGroup = resolveOrderIsGroup(downstreamBody);
|
||||
@@ -794,20 +796,29 @@ public class CanteenHotelOrderServiceImpl extends ServiceImpl<CanteenHotelOrderM
|
||||
// 文档:bookerPersonId 必填;无员工号时用手机号兜底
|
||||
req.put("bookerPersonId",
|
||||
StrUtil.blankToDefault(str(downstreamBody.get("bookerPersonId")), order.getContactPhone()));
|
||||
req.put("bookerPersonFirstName", order.getContactName());
|
||||
NameParts booker = splitPersonName(order.getContactName());
|
||||
req.put("bookerPersonLastName", booker.lastName());
|
||||
req.put("bookerPersonFirstName", booker.firstName());
|
||||
req.put("bookerPersonMobile", order.getContactPhone());
|
||||
req.put("adults", downstreamBody.get("adults") == null ? 2
|
||||
: Integer.parseInt(String.valueOf(downstreamBody.get("adults"))));
|
||||
|
||||
String guestName = order.getContactName();
|
||||
String guestRawName = order.getContactName();
|
||||
String guestMobile = order.getContactPhone();
|
||||
String guestLastOverride = null;
|
||||
String guestFirstOverride = null;
|
||||
if (StrUtil.isNotBlank(order.getGuestJson())) {
|
||||
try {
|
||||
JSONArray guests = JSONUtil.parseArray(order.getGuestJson());
|
||||
if (guests != null && !guests.isEmpty()) {
|
||||
JSONObject g0 = guests.getJSONObject(0);
|
||||
guestName = StrUtil.blankToDefault(
|
||||
StrUtil.blankToDefault(g0.getStr("guestFirstName"), g0.getStr("name")), guestName);
|
||||
guestLastOverride = g0.getStr("guestLastName");
|
||||
guestFirstOverride = g0.getStr("guestFirstName");
|
||||
guestRawName = StrUtil.blankToDefault(g0.getStr("name"),
|
||||
StrUtil.blankToDefault(guestFirstOverride, guestRawName));
|
||||
if (StrUtil.isNotBlank(guestLastOverride) && StrUtil.isNotBlank(guestFirstOverride)) {
|
||||
guestRawName = guestLastOverride + guestFirstOverride;
|
||||
}
|
||||
guestMobile = StrUtil.blankToDefault(
|
||||
StrUtil.blankToDefault(g0.getStr("guestPhone"), g0.getStr("mobile")), guestMobile);
|
||||
}
|
||||
@@ -815,13 +826,16 @@ public class CanteenHotelOrderServiceImpl extends ServiceImpl<CanteenHotelOrderM
|
||||
catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
// 入住人:文档字段 guestInfoList(可选整体为空;传了则 guestFirstName/guestRoomNo 必填)
|
||||
NameParts guest = (StrUtil.isNotBlank(guestLastOverride) && StrUtil.isNotBlank(guestFirstOverride))
|
||||
? new NameParts(guestLastOverride, guestFirstOverride)
|
||||
: splitPersonName(guestRawName);
|
||||
com.alibaba.fastjson.JSONArray guestInfoList = new com.alibaba.fastjson.JSONArray();
|
||||
com.alibaba.fastjson.JSONObject guest = new com.alibaba.fastjson.JSONObject();
|
||||
guest.put("guestFirstName", guestName);
|
||||
guest.put("guestPhone", guestMobile);
|
||||
guest.put("guestRoomNo", 1);
|
||||
guestInfoList.add(guest);
|
||||
com.alibaba.fastjson.JSONObject guestRow = new com.alibaba.fastjson.JSONObject();
|
||||
guestRow.put("guestLastName", guest.lastName());
|
||||
guestRow.put("guestFirstName", guest.firstName());
|
||||
guestRow.put("guestPhone", guestMobile);
|
||||
guestRow.put("guestRoomNo", 1);
|
||||
guestInfoList.add(guestRow);
|
||||
req.put("guestInfoList", guestInfoList);
|
||||
|
||||
Object remark = downstreamBody.get("remark");
|
||||
@@ -848,6 +862,38 @@ public class CanteenHotelOrderServiceImpl extends ServiceImpl<CanteenHotelOrderM
|
||||
return shangyouProperties.isOrderIsGroup();
|
||||
}
|
||||
|
||||
/**
|
||||
* 上游姓名格式:LastName=姓 / FirstName=名。
|
||||
* <ul>
|
||||
* <li>中文:首字为姓,其余为名(复姓未特殊处理;单字名时名为 "/")</li>
|
||||
* <li>英文空格分隔:按 First Last 或 Last First 取两段</li>
|
||||
* <li>其它:整段进 LastName,FirstName="/"</li>
|
||||
* </ul>
|
||||
*/
|
||||
private static NameParts splitPersonName(String raw) {
|
||||
String name = StrUtil.trim(raw);
|
||||
if (StrUtil.isBlank(name)) {
|
||||
return new NameParts("测", "试");
|
||||
}
|
||||
// 去掉中间空格后的纯中文
|
||||
String compact = name.replace(" ", "").replace(" ", "");
|
||||
if (compact.chars().allMatch(ch -> Character.UnicodeScript.of(ch) == Character.UnicodeScript.HAN)) {
|
||||
if (compact.length() == 1) {
|
||||
return new NameParts(compact, "/");
|
||||
}
|
||||
return new NameParts(compact.substring(0, 1), compact.substring(1));
|
||||
}
|
||||
// 英文:含空格则拆成两段(名 姓 → first last;若多段取首为名、末为姓)
|
||||
String[] parts = name.trim().split("\\s+");
|
||||
if (parts.length >= 2) {
|
||||
return new NameParts(parts[parts.length - 1], parts[0]);
|
||||
}
|
||||
return new NameParts(name, "/");
|
||||
}
|
||||
|
||||
private record NameParts(String lastName, String firstName) {
|
||||
}
|
||||
|
||||
/**
|
||||
* 按住店夜补齐 dailyPriceList;本地日历缺夜则用产品底价兜底。
|
||||
*/
|
||||
|
||||
+51
-1
@@ -15,7 +15,10 @@ import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
||||
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
||||
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
||||
import com.bsd.pjcloud.canteen.entity.CanteenHotelEntity;
|
||||
import com.bsd.pjcloud.canteen.mapper.CanteenHotelDailyMapper;
|
||||
import com.bsd.pjcloud.canteen.mapper.CanteenHotelMapper;
|
||||
import com.bsd.pjcloud.canteen.mapper.CanteenHotelProductMapper;
|
||||
import com.bsd.pjcloud.canteen.mapper.CanteenHotelRoomMapper;
|
||||
import com.bsd.pjcloud.canteen.service.CanteenHotelService;
|
||||
import com.bsd.pjcloud.canteen.vo.CanteenHotelImportResultVO;
|
||||
import com.bsd.pjcloud.common.core.constant.CommonConstants;
|
||||
@@ -53,10 +56,57 @@ public class CanteenHotelServiceImpl extends ServiceImpl<CanteenHotelMapper, Can
|
||||
private static final int SKIP_SAMPLE_LIMIT = 50;
|
||||
private static final ConcurrentHashMap<String, CanteenHotelImportResultVO> IMPORT_TASKS = new ConcurrentHashMap<>();
|
||||
|
||||
private final CanteenHotelRoomMapper roomMapper;
|
||||
|
||||
private final CanteenHotelProductMapper productMapper;
|
||||
|
||||
private final CanteenHotelDailyMapper dailyMapper;
|
||||
|
||||
@Override
|
||||
public Page<CanteenHotelEntity> pageScope(Page<CanteenHotelEntity> page,
|
||||
LambdaQueryWrapper<CanteenHotelEntity> wrapper) {
|
||||
return baseMapper.selectPageByScope(page, wrapper, DataScope.of());
|
||||
Page<CanteenHotelEntity> result = baseMapper.selectPageByScope(page, wrapper, DataScope.of());
|
||||
fillDetailCounts(result.getRecords());
|
||||
return result;
|
||||
}
|
||||
|
||||
/** 为本页酒店批量补齐房型/产品/日历数量,避免 N+1 */
|
||||
private void fillDetailCounts(List<CanteenHotelEntity> records) {
|
||||
if (CollUtil.isEmpty(records)) {
|
||||
return;
|
||||
}
|
||||
List<Long> ids = records.stream().map(CanteenHotelEntity::getId).filter(id -> id != null).toList();
|
||||
if (ids.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<Long, Integer> roomMap = toCountMap(roomMapper.countGroupByHotelIds(ids));
|
||||
Map<Long, Integer> productMap = toCountMap(productMapper.countGroupByHotelIds(ids));
|
||||
Map<Long, Integer> dailyMap = toCountMap(dailyMapper.countGroupByHotelIds(ids));
|
||||
for (CanteenHotelEntity h : records) {
|
||||
Long id = h.getId();
|
||||
h.setRoomCount(roomMap.getOrDefault(id, 0));
|
||||
h.setProductCount(productMap.getOrDefault(id, 0));
|
||||
h.setDailyCount(dailyMap.getOrDefault(id, 0));
|
||||
}
|
||||
}
|
||||
|
||||
private static Map<Long, Integer> toCountMap(List<Map<String, Object>> rows) {
|
||||
Map<Long, Integer> map = new HashMap<>();
|
||||
if (CollUtil.isEmpty(rows)) {
|
||||
return map;
|
||||
}
|
||||
for (Map<String, Object> row : rows) {
|
||||
Object hid = row.get("hotelId");
|
||||
if (hid == null) {
|
||||
hid = row.get("hotel_id");
|
||||
}
|
||||
Object cnt = row.get("cnt");
|
||||
if (hid == null || cnt == null) {
|
||||
continue;
|
||||
}
|
||||
map.put(Long.valueOf(String.valueOf(hid)), Integer.valueOf(String.valueOf(cnt)));
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+196
-2
@@ -157,6 +157,53 @@ public class CanteenHotelSyncServiceImpl implements CanteenHotelSyncService {
|
||||
return R.ok(logEntity, "已提交全量详情拉取");
|
||||
}
|
||||
|
||||
@Override
|
||||
public R<CanteenHotelSyncLogEntity> startRetryFailedAsync(Long sourceLogId) {
|
||||
if (sourceLogId == null) {
|
||||
return R.failed("请指定失败来源日志 logId");
|
||||
}
|
||||
CanteenHotelSyncLogEntity src = syncLogMapper.selectById(sourceLogId);
|
||||
if (src == null) {
|
||||
return R.failed("同步记录不存在");
|
||||
}
|
||||
if (STATUS_RUNNING.equals(src.getStatus())) {
|
||||
return R.failed("原任务仍在执行,请结束后再重试失败项");
|
||||
}
|
||||
String syncType = src.getSyncType();
|
||||
if (!TYPE_DETAIL.equals(syncType) && !TYPE_ROLL.equals(syncType)) {
|
||||
return R.failed("仅支持「全量详情」或「滚价」的失败重试");
|
||||
}
|
||||
List<String> codes = parseFailHotelCodes(src.getFailDetail());
|
||||
if (CollUtil.isEmpty(codes)) {
|
||||
return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因)");
|
||||
}
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<CanteenHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(CanteenHotelSyncLogEntity::getSyncType, syncType)
|
||||
.eq(CanteenHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
return R.failed("已有同类型任务进行中,请稍后再试");
|
||||
}
|
||||
|
||||
CanteenHotelSyncLogEntity logEntity = newLog(syncType, "RETRY");
|
||||
logEntity.setStage("QUEUED");
|
||||
logEntity.setStageMsg("重试失败酒店:来源日志 " + sourceLogId + ",共 " + codes.size() + " 家"
|
||||
+ (TYPE_ROLL.equals(syncType) ? "(滚价日=" + resolveRollDay(src) + ")" : ""));
|
||||
logEntity.setProgressPct(1);
|
||||
logEntity.setLocalCount((long) codes.size());
|
||||
syncLogMapper.insert(logEntity);
|
||||
|
||||
Long tenantId = TenantContextHolder.getTenantId();
|
||||
if (tenantId == null) {
|
||||
tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
|
||||
}
|
||||
final Long tid = tenantId;
|
||||
final Long logId = logEntity.getId();
|
||||
final List<String> codeList = codes;
|
||||
final LocalDate rollDay = TYPE_ROLL.equals(syncType) ? resolveRollDay(src) : null;
|
||||
EXECUTOR.execute(() -> TenantBroker.runAs(tid, id -> runRetryFailed(logId, syncType, codeList, rollDay)));
|
||||
return R.ok(logEntity, "已提交失败重试,共 " + codes.size() + " 家");
|
||||
}
|
||||
|
||||
@Override
|
||||
public R<CanteenHotelSyncLogEntity> getSyncLog(Long id) {
|
||||
CanteenHotelSyncLogEntity e = syncLogMapper.selectById(id);
|
||||
@@ -311,6 +358,16 @@ public class CanteenHotelSyncServiceImpl implements CanteenHotelSyncService {
|
||||
fail++;
|
||||
appendFail(failLines, h, e.getMessage());
|
||||
}
|
||||
int intervalMs = Math.max(0, properties.getPriceRollIntervalMs());
|
||||
if (intervalMs > 0) {
|
||||
try {
|
||||
Thread.sleep(intervalMs);
|
||||
}
|
||||
catch (InterruptedException ie) {
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
int pct = total <= 0 ? 50 : Math.min(95, 5 + (int) (processed * 90L / total));
|
||||
progress(logId, pct, "ROLL",
|
||||
@@ -452,12 +509,149 @@ public class CanteenHotelSyncServiceImpl implements CanteenHotelSyncService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 仅重试某次失败明细里的酒店码(详情拉取或滚价)。
|
||||
*/
|
||||
private void runRetryFailed(Long logId, String syncType, List<String> hotelCodes, LocalDate rollDay) {
|
||||
long t0 = System.currentTimeMillis();
|
||||
try {
|
||||
progress(logId, 5, "SCAN", "正在按失败明细匹配本地酒店…");
|
||||
List<CanteenHotelEntity> hotels = new ArrayList<>();
|
||||
int batch = 200;
|
||||
for (int i = 0; i < hotelCodes.size(); i += batch) {
|
||||
List<String> slice = hotelCodes.subList(i, Math.min(i + batch, hotelCodes.size()));
|
||||
List<CanteenHotelEntity> part = hotelMapper.selectList(Wrappers.<CanteenHotelEntity>lambdaQuery()
|
||||
.select(CanteenHotelEntity::getId, CanteenHotelEntity::getHotelCode,
|
||||
CanteenHotelEntity::getHotelName)
|
||||
.in(CanteenHotelEntity::getHotelCode, slice));
|
||||
if (CollUtil.isNotEmpty(part)) {
|
||||
hotels.addAll(part);
|
||||
}
|
||||
}
|
||||
int target = hotels.size();
|
||||
int missing = hotelCodes.size() - target;
|
||||
int ok = 0;
|
||||
int fail = 0;
|
||||
int skip = missing;
|
||||
int processed = 0;
|
||||
List<String> failLines = new ArrayList<>();
|
||||
int intervalMs = TYPE_ROLL.equals(syncType)
|
||||
? Math.max(0, properties.getPriceRollIntervalMs())
|
||||
: Math.max(0, properties.getDetailPullIntervalMs());
|
||||
LocalDate day = rollDay != null ? rollDay : LocalDate.now().plusDays(Math.max(1, properties.getPriceRollDayOffset()));
|
||||
LocalDate end = day.plusDays(1);
|
||||
|
||||
progress(logId, 8, "RETRY", "开始重试 " + target + " 家"
|
||||
+ (missing > 0 ? "(明细中有 " + missing + " 家本地未匹配)" : "")
|
||||
+ (TYPE_ROLL.equals(syncType) ? " day=" + day : "") + "…");
|
||||
|
||||
for (CanteenHotelEntity h : hotels) {
|
||||
processed++;
|
||||
if (StrUtil.isBlank(h.getHotelCode())) {
|
||||
skip++;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
R<Map<String, Object>> r;
|
||||
if (TYPE_ROLL.equals(syncType)) {
|
||||
r = midService.pullPriceRange(h.getId(), day, end);
|
||||
}
|
||||
else {
|
||||
r = midService.pullFromUpstream(h.getId());
|
||||
}
|
||||
if (r != null && r.getCode() == 0) {
|
||||
ok++;
|
||||
}
|
||||
else {
|
||||
fail++;
|
||||
appendFail(failLines, h, r == null ? "返回为空" : StrUtil.blankToDefault(r.getMsg(), "业务失败"));
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
fail++;
|
||||
appendFail(failLines, h, e.getMessage());
|
||||
}
|
||||
int pct = target <= 0 ? 50 : Math.min(95, 8 + (int) (processed * 87L / target));
|
||||
progress(logId, pct, "RETRY", "重试中 " + processed + "/" + target + " 成功 " + ok + " 失败 " + fail
|
||||
+ " 当前 " + StrUtil.blankToDefault(h.getHotelName(), h.getHotelCode()));
|
||||
if (intervalMs > 0) {
|
||||
try {
|
||||
Thread.sleep(intervalMs);
|
||||
}
|
||||
catch (InterruptedException ie) {
|
||||
Thread.currentThread().interrupt();
|
||||
failLog(logId, "任务被中断", System.currentTimeMillis() - t0);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
CanteenHotelSyncLogEntity done = syncLogMapper.selectById(logId);
|
||||
if (done == null) {
|
||||
return;
|
||||
}
|
||||
done.setStatus(STATUS_SUCCESS);
|
||||
done.setStage("DONE");
|
||||
done.setStageMsg("失败重试完成" + (TYPE_ROLL.equals(syncType) ? " day=" + day : ""));
|
||||
done.setProgressPct(100);
|
||||
done.setSuccessCount(ok);
|
||||
done.setFailCount(fail);
|
||||
done.setSkipCount(skip);
|
||||
done.setLocalCount((long) hotelCodes.size());
|
||||
applyFailDetail(done, failLines, fail);
|
||||
done.setCostMs(System.currentTimeMillis() - t0);
|
||||
done.setFinishTime(LocalDateTime.now());
|
||||
done.setUpdateTime(LocalDateTime.now());
|
||||
syncLogMapper.updateById(done);
|
||||
log.info("|失败重试|完成|logId={}|type={}|ok={}|fail={}|skip={}|", logId, syncType, ok, fail, skip);
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("|失败重试|失败|logId={}|", logId, e);
|
||||
failLog(logId, e.getMessage(), System.currentTimeMillis() - t0);
|
||||
}
|
||||
}
|
||||
|
||||
/** 从 failDetail 解析 hotelCode(每行 code|name|reason) */
|
||||
private static List<String> parseFailHotelCodes(String failDetail) {
|
||||
List<String> codes = new ArrayList<>();
|
||||
if (StrUtil.isBlank(failDetail)) {
|
||||
return codes;
|
||||
}
|
||||
java.util.LinkedHashSet<String> uniq = new java.util.LinkedHashSet<>();
|
||||
for (String line : StrUtil.splitTrim(failDetail, '\n')) {
|
||||
if (StrUtil.isBlank(line) || line.startsWith("…") || line.startsWith("...")) {
|
||||
continue;
|
||||
}
|
||||
String code = StrUtil.subBefore(line, "|", false);
|
||||
if (StrUtil.isNotBlank(code) && !code.contains(" ")) {
|
||||
uniq.add(code.trim());
|
||||
}
|
||||
}
|
||||
codes.addAll(uniq);
|
||||
return codes;
|
||||
}
|
||||
|
||||
/** 从滚价日志文案解析日期;解析不到则用配置偏移 */
|
||||
private LocalDate resolveRollDay(CanteenHotelSyncLogEntity src) {
|
||||
String msg = StrUtil.blankToDefault(src.getStageMsg(), "") + " " + StrUtil.blankToDefault(src.getErrorMsg(), "");
|
||||
java.util.regex.Matcher m = java.util.regex.Pattern.compile("day=(\\d{4}-\\d{2}-\\d{2})").matcher(msg);
|
||||
if (m.find()) {
|
||||
try {
|
||||
return LocalDate.parse(m.group(1));
|
||||
}
|
||||
catch (Exception ignored) {
|
||||
// fall through
|
||||
}
|
||||
}
|
||||
return LocalDate.now().plusDays(Math.max(1, properties.getPriceRollDayOffset()));
|
||||
}
|
||||
|
||||
private static void appendFail(List<String> failLines, CanteenHotelEntity h, String reason) {
|
||||
if (failLines == null) {
|
||||
return;
|
||||
}
|
||||
// 明细最多保留 500 行,避免撑爆字段
|
||||
if (failLines.size() >= 500) {
|
||||
// 明细最多保留 2000 行,保证失败重试能覆盖大部分码
|
||||
if (failLines.size() >= 2000) {
|
||||
return;
|
||||
}
|
||||
String code = StrUtil.blankToDefault(h.getHotelCode(), String.valueOf(h.getId()));
|
||||
|
||||
@@ -30,100 +30,104 @@ security:
|
||||
default-encrypt-type: AES # 可选:AES、SM4,默认 AES
|
||||
aes-key: pigxpigxpigxpigx # AES 密钥必须为 16 位
|
||||
|
||||
# ========== 玉编电云 / TMC CRS 酒店上游(docs/shangyou)==========
|
||||
# 生产对方通常只给一个根地址:https://onlink.yubdianyun.cn/link/provider
|
||||
# 此时只配 base-url 即可:鉴权 POST {base-url}/oauth/token ,业务 POST {base-url}/hotel/...
|
||||
# 测试若鉴权与业务拆成两个地址,再单独配 auth-base-url / biz-base-url
|
||||
shangyou-hotel:
|
||||
# 是否启用上游调用(false 时拉酒店/房价/下单都会直接报「未启用」)
|
||||
enabled: true
|
||||
# 上游根地址(生产常用这一项即可,同时用于鉴权+业务)
|
||||
base-url: https://onlink.yubdianyun.cn/link/provider
|
||||
# 鉴权根地址;一般留空,自动用 base-url。仅当测试环境鉴权与业务分离时才填,例 https://xxx/link
|
||||
# auth-base-url: ""
|
||||
# 业务根地址;一般留空,自动用 base-url。仅当业务地址与 base-url 不同时才填
|
||||
# biz-base-url: ""
|
||||
# OAuth clientId(对方开通账号时下发,用于换 access_token)
|
||||
client-id: F210AE02B74C48A3901B
|
||||
# OAuth clientSecret(与 client-id 配对,Basic 鉴权换 token)
|
||||
client-secret: 99165116A43F413CB799C91714861408
|
||||
# 业务卡号 cardId(账号),请求体 content 里会带
|
||||
card-id: 9082727358de43be96fe
|
||||
# 业务秘钥 cardPassWord,请求体 content 里会带;也可兼作回调验签 Secret
|
||||
card-password: 5bffd3c42e5a42219872bace3092983d
|
||||
# 房价/房量回调验签 Secret:sign=MD5(Secret+time);留空则用上面的 card-password
|
||||
callback-secret: 99165116A43F413CB799C91714861408
|
||||
# 下单预订人 bookerId;留空则用 card-id;以对方开通为准
|
||||
booker-id: ""
|
||||
# 是否总对总下单(企业中转一般 true;现付调试可 false)
|
||||
order-is-group: true
|
||||
# 担保模式:NONE / FIRST_NIGHT / FULL_AMOUNT,以对方开卡为准
|
||||
order-guarantee-mode: NONE
|
||||
# 房价接口路径。生产必须用 /hotel/getRoomPrice(/hotel/getTMCRoomPrice 会 404)
|
||||
room-price-path: /hotel/getRoomPrice
|
||||
# 会员等级,逗号分隔;对方要求传时再填,空则不传该字段
|
||||
member-levels: ""
|
||||
# 调上游 HTTP 超时(毫秒),含拉酒店列表/房价等
|
||||
timeout-ms: 60000
|
||||
# token 提前作废秒数(对方 token 默认约 21600 秒);避免边界时刻用到已失效 token
|
||||
token-skew-seconds: 120
|
||||
# 「拉取上游」时拉未来多少天房价;建议 1~90;接口单次约 31 天,代码会自动按段切片
|
||||
price-days: 90
|
||||
# 授权酒店列表 Redis 缓存分钟数(HASH+ZSET,不落 JVM 全量,支持十几万级)
|
||||
hotel-list-cache-minutes: 30
|
||||
# 授权酒店数量告警阈值(仅打日志告警,不截断);十几万级强制刷新建议低峰做
|
||||
hotel-list-warn-size: 100000
|
||||
# 回调无 TENANT-ID:写入任务/匹配本地酒店用的默认租户(酒店中转一般单租户)
|
||||
callback-tenant-id: 1
|
||||
# 是否启用回调异步队列 Worker(Redis BRPOP 消费后 pullFromUpstream)
|
||||
callback-queue-enabled: true
|
||||
# 单条回调任务最大尝试次数(含首次;失败会自动重入队)
|
||||
callback-max-attempts: 3
|
||||
# ----- 日更 / 差集 -----
|
||||
# 酒店列表日同步开关(false 则定时/XXL 都跳过)
|
||||
hotel-list-daily-sync-enabled: true
|
||||
# 是否启用本服务 Spring Cron(改下面 cron 即可生效;已用 XXL 请设 false,避免一天跑两次)
|
||||
spring-schedule-enabled: true
|
||||
# 列表日更时间:默认每天 02:00(Spring Cron 6 段:秒 分 时 日 月 周)
|
||||
hotel-list-daily-cron: "0 0 2 * * ?"
|
||||
# 滚价日更时间:默认每天 03:00
|
||||
price-roll-daily-cron: "0 0 3 * * ?"
|
||||
# 列表同步后是否自动把「上游有本地无」写入本地主数据
|
||||
auto-import-new-hotels: true
|
||||
# 自动导入单次上限(剩余留在差集,可再点导入)
|
||||
auto-import-batch-size: 5000
|
||||
# 每日滚第 N 天房价开关
|
||||
price-roll-daily-enabled: true
|
||||
# 滚价偏移:today + N(文档第 91 天 → 90)
|
||||
price-roll-day-offset: 90
|
||||
# 滚价按本地酒店 id 游标每批数量(防 OOM)
|
||||
price-roll-batch-size: 100
|
||||
# 全量房价房量:每天最多次数 / 最小间隔小时(「拉取全部启用详情」受此限制)
|
||||
full-price-sync-max-per-day: 3
|
||||
full-price-sync-min-interval-hours: 6
|
||||
# 全量拉详情:每家间隔毫秒(落库已批量,默认 0;上游压力大时可调大)
|
||||
detail-pull-interval-ms: 0
|
||||
# 全量拉详情单次最多家数;0=不限制(酒店很多时建议先设 100 试跑)
|
||||
detail-pull-max-hotels: 0
|
||||
# 差集扫描 / 重建本地 code SET 批大小
|
||||
diff-scan-batch-size: 500
|
||||
# 同步后:本地有上游无 → 自动标 OFFLINE 停用
|
||||
auto-mark-offline-hotels: true
|
||||
# 同步后:已停用但上游又出现 → 自动恢复 ACTIVE
|
||||
auto-reactivate-online-hotels: true
|
||||
|
||||
# ========== 下游销售方推送(《第三方数据同步接口文档》)==========
|
||||
# 本系统作为中转 A,主动把加价后的酒店/房价推到下游 B(与上面「上游」无关,可单独开关)
|
||||
hotel-downstream:
|
||||
# 是否启用向下游推送
|
||||
enabled: false
|
||||
# 下游 B 系统根地址(不含尾斜杠),实际请求 = base-url + /third/party/api/...
|
||||
base-url: https://b.example.com
|
||||
# B 系统分配给我们的 appKey(请求头/签名用)
|
||||
app-key: TP_xxx
|
||||
# B 系统分配的 appSecret(仅本地算签名,不落报文)
|
||||
app-secret: your-secret
|
||||
# 推下游 HTTP 超时(毫秒)
|
||||
timeout-ms: 30000
|
||||
# 兼容字段;实际上游拉取天数以 shangyou-hotel.price-days 为准
|
||||
daily-days: 90
|
||||
## ========== 玉编电云 / TMC CRS 酒店上游(docs/shangyou)==========
|
||||
## 生产对方通常只给一个根地址:https://onlink.yubdianyun.cn/link/provider
|
||||
## 此时只配 base-url 即可:鉴权 POST {base-url}/oauth/token ,业务 POST {base-url}/hotel/...
|
||||
## 测试若鉴权与业务拆成两个地址,再单独配 auth-base-url / biz-base-url
|
||||
#shangyou-hotel:
|
||||
# # 是否启用上游调用(false 时拉酒店/房价/下单都会直接报「未启用」)
|
||||
# enabled: true
|
||||
# # 上游根地址(生产常用这一项即可,同时用于鉴权+业务)
|
||||
# base-url: https://onlink.yubdianyun.cn/link/provider
|
||||
# # 鉴权根地址;一般留空,自动用 base-url。仅当测试环境鉴权与业务分离时才填,例 https://xxx/link
|
||||
# # auth-base-url: ""
|
||||
# # 业务根地址;一般留空,自动用 base-url。仅当业务地址与 base-url 不同时才填
|
||||
# # biz-base-url: ""
|
||||
# # OAuth clientId(对方开通账号时下发,用于换 access_token)
|
||||
# client-id: F210AE02B74C48A3901B
|
||||
# # OAuth clientSecret(与 client-id 配对,Basic 鉴权换 token)
|
||||
# client-secret: 99165116A43F413CB799C91714861408
|
||||
# # 业务卡号 cardId(账号),请求体 content 里会带
|
||||
# card-id: 9082727358de43be96fe
|
||||
# # 业务秘钥 cardPassWord,请求体 content 里会带;也可兼作回调验签 Secret
|
||||
# card-password: 5bffd3c42e5a42219872bace3092983d
|
||||
# # 房价/房量回调验签 Secret:sign=MD5(Secret+time);留空则用上面的 card-password
|
||||
# callback-secret: 99165116A43F413CB799C91714861408
|
||||
# # 下单预订人 bookerId;留空则用 card-id;以对方开通为准
|
||||
# booker-id: ""
|
||||
# # 是否总对总下单(企业中转一般 true;现付调试可 false)
|
||||
# order-is-group: true
|
||||
# # 担保模式:NONE / FIRST_NIGHT / FULL_AMOUNT,以对方开卡为准
|
||||
# order-guarantee-mode: NONE
|
||||
# # 房价接口路径。生产必须用 /hotel/getRoomPrice(/hotel/getTMCRoomPrice 会 404)
|
||||
# room-price-path: /hotel/getRoomPrice
|
||||
# # 会员等级,逗号分隔;对方要求传时再填,空则不传该字段
|
||||
# member-levels: ""
|
||||
# # 调上游 HTTP 超时(毫秒),含拉酒店列表/房价等
|
||||
# timeout-ms: 60000
|
||||
# # token 提前作废秒数(对方 token 默认约 21600 秒);避免边界时刻用到已失效 token
|
||||
# token-skew-seconds: 120
|
||||
# # 「拉取上游」时拉未来多少天房价;建议 1~90;接口单次约 31 天,代码会自动按段切片
|
||||
# price-days: 90
|
||||
# # 授权酒店列表 Redis 缓存分钟数(HASH+ZSET,不落 JVM 全量,支持十几万级)
|
||||
# hotel-list-cache-minutes: 30
|
||||
# # 授权酒店数量告警阈值(仅打日志告警,不截断);十几万级强制刷新建议低峰做
|
||||
# hotel-list-warn-size: 100000
|
||||
# # 回调无 TENANT-ID:写入任务/匹配本地酒店用的默认租户(酒店中转一般单租户)
|
||||
# callback-tenant-id: 1
|
||||
# # 是否启用回调异步队列 Worker(Redis BRPOP 消费后 pullFromUpstream)
|
||||
# callback-queue-enabled: true
|
||||
# # 单条回调任务最大尝试次数(含首次;失败会自动重入队)
|
||||
# callback-max-attempts: 3
|
||||
# # 回调队列处理完一家后的间隔毫秒(房价另有 QPS 限流,这里可较小)
|
||||
# callback-task-interval-ms: 50
|
||||
# # 房价 getRoomPrice 限流(对方约定约 20 QPS)。多实例请均分,如 2 实例各配 10;≤0 关闭
|
||||
# price-api-qps: 20
|
||||
# # ----- 日更 / 差集 -----
|
||||
# # 酒店列表日同步开关(false 则定时/XXL 都跳过)
|
||||
# hotel-list-daily-sync-enabled: true
|
||||
# # 是否启用本服务 Spring Cron(改下面 cron 即可生效;已用 XXL 请设 false,避免一天跑两次)
|
||||
# spring-schedule-enabled: true
|
||||
# # 列表日更时间:默认每天 02:00(Spring Cron 6 段:秒 分 时 日 月 周)
|
||||
# hotel-list-daily-cron: "0 0 2 * * ?"
|
||||
# # 滚价日更时间:默认每天 03:00
|
||||
# price-roll-daily-cron: "0 0 3 * * ?"
|
||||
# # 列表同步后是否自动把「上游有本地无」写入本地主数据
|
||||
# auto-import-new-hotels: true
|
||||
# # 自动导入单次上限(剩余留在差集,可再点导入)
|
||||
# auto-import-batch-size: 5000
|
||||
# # 每日滚第 N 天房价开关
|
||||
# price-roll-daily-enabled: true
|
||||
# # 滚价偏移:today + N(文档第 91 天 → 90)
|
||||
# price-roll-day-offset: 90
|
||||
# # 滚价按本地酒店 id 游标每批数量(防 OOM)
|
||||
# price-roll-batch-size: 100
|
||||
# # 全量房价房量:每天最多次数 / 最小间隔小时(「拉取全部启用详情」受此限制)
|
||||
# full-price-sync-max-per-day: 3
|
||||
# full-price-sync-min-interval-hours: 6
|
||||
# # 全量拉详情:每家间隔毫秒(落库已批量,默认 0;上游压力大时可调大)
|
||||
# detail-pull-interval-ms: 0
|
||||
# # 全量拉详情单次最多家数;0=不限制(酒店很多时建议先设 100 试跑)
|
||||
# detail-pull-max-hotels: 0
|
||||
# # 差集扫描 / 重建本地 code SET 批大小
|
||||
# diff-scan-batch-size: 500
|
||||
# # 同步后:本地有上游无 → 自动标 OFFLINE 停用
|
||||
# auto-mark-offline-hotels: true
|
||||
# # 同步后:已停用但上游又出现 → 自动恢复 ACTIVE
|
||||
# auto-reactivate-online-hotels: true
|
||||
#
|
||||
## ========== 下游销售方推送(《第三方数据同步接口文档》)==========
|
||||
## 本系统作为中转 A,主动把加价后的酒店/房价推到下游 B(与上面「上游」无关,可单独开关)
|
||||
#hotel-downstream:
|
||||
# # 是否启用向下游推送
|
||||
# enabled: false
|
||||
# # 下游 B 系统根地址(不含尾斜杠),实际请求 = base-url + /third/party/api/...
|
||||
# base-url: https://b.example.com
|
||||
# # B 系统分配给我们的 appKey(请求头/签名用)
|
||||
# app-key: TP_xxx
|
||||
# # B 系统分配的 appSecret(仅本地算签名,不落报文)
|
||||
# app-secret: your-secret
|
||||
# # 推下游 HTTP 超时(毫秒)
|
||||
# timeout-ms: 30000
|
||||
# # 兼容字段;实际上游拉取天数以 shangyou-hotel.price-days 为准
|
||||
# daily-days: 90
|
||||
Reference in new issue
Block a user