小如酒店

2026-11-04 14:19:41 +08:00
1 parent 87a2b76250
commit 0f2a496188
@@ -220,11 +220,12 @@ public class HotelZaxController {
 	}

 	@PostMapping("/sync")
-	@Operation(summary = "从资安心开放平台同步到本地（默认全量，适配授权范围变更）")
+	@Operation(summary = "从资安心开放平台异步同步到本地（立刻返回，进度见同步水位；默认全量）")
 	@PreAuthorize("@ss.hasPermission('product:hotel-zax:sync')")
 	public CommonResult<Map<String, Object>> sync(
 			@RequestParam(value = "full", required = false, defaultValue = "true") Boolean full) {
-		return success(hotelZaxSyncService.syncOnce(Boolean.TRUE.equals(full)));
+		// 异步提交：避免几十/上万家酒店时 HTTP 直接超时
+		return success(hotelZaxSyncService.submitSyncAsync(Boolean.TRUE.equals(full)));
 	}

 	@PostMapping("/reset-cursor")
@@ -29,8 +29,14 @@ public class ZaxHotelOpenProperties {
 	/** 连接超时毫秒 */
 	private int connectTimeoutMs = 5000;

-	/** 读超时毫秒 */
-	private int readTimeoutMs = 20000;
+	/** 读超时毫秒（单次 HTTP；全量同步改为异步后可适当加大） */
+	private int readTimeoutMs = 60000;
+
+	/**
+	 * 同步时单页内并行拉取详情+日历的并发数。
+	 * 过大可能触发供应方限流；过小十万级会极慢。建议 3~8。
+	 */
+	private int syncDetailConcurrency = 4;

 	/**
 	 * 供应方回调落库使用的租户（回调无租户头时用此值）。
@@ -18,6 +18,12 @@ public interface HotelZaxSyncService {
 	 */
 	Map<String, Object> syncOnce(boolean full);

+	/**
+	 * 后台提交异步全量/增量同步（立刻返回，避免 HTTP 超时）。
+	 * @return accepted=true；若已有任务在跑则 accepted=false 并带 running 说明
+	 */
+	Map<String, Object> submitSyncAsync(boolean full);
+
 	/**
 	 * 单店拉取详情并落库（资源回调触发）。
 	 */
@@ -15,6 +15,8 @@ import cn.iocoder.yudao.module.product.dal.mysql.hotel.HotelZaxMapper;
 import cn.iocoder.yudao.module.product.dal.mysql.hotel.HotelZaxProductMapper;
 import cn.iocoder.yudao.module.product.dal.mysql.hotel.HotelZaxRoomMapper;
 import cn.iocoder.yudao.module.product.dal.mysql.hotel.HotelZaxSyncCursorMapper;
+import cn.iocoder.yudao.framework.tenant.core.context.TenantContextHolder;
+import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
 import cn.iocoder.yudao.module.product.framework.hotel.ZaxHotelOpenClient;
 import cn.iocoder.yudao.module.product.framework.hotel.ZaxHotelOpenProperties;
 import lombok.RequiredArgsConstructor;
@@ -27,12 +29,18 @@ import java.math.BigDecimal;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;

 /**
 * 开放平台 hotel/sync + detail → 本地 hotel_zax / room / product（结构对齐供应侧）。
@@ -46,6 +54,15 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService {

 	private static final DateTimeFormatter DT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

+	/** 后台/定时共用：同一时刻只允许一条全量/增量同步，避免互相踩水位 */
+	private static final AtomicBoolean SYNC_RUNNING = new AtomicBoolean(false);
+
+	private static final ExecutorService SYNC_ASYNC = Executors.newSingleThreadExecutor(r -> {
+		Thread t = new Thread(r, "hotel-zax-sync-async");
+		t.setDaemon(true);
+		return t;
+	});
+
 	private final ZaxHotelOpenClient openClient;

 	private final ZaxHotelOpenProperties openProperties;
@@ -61,15 +78,72 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService {
 	private final HotelZaxSyncCursorMapper cursorMapper;

 	@Override
-	@Transactional(rollbackFor = Exception.class)
 	public Map<String, Object> syncOnce() {
 		// 管理端「同步资安心」默认全量：改授权范围后增量水位拉不到新授权酒店
 		return syncOnce(true);
 	}

 	@Override
-	@Transactional(rollbackFor = Exception.class)
 	public Map<String, Object> syncOnce(boolean full) {
+		if (!SYNC_RUNNING.compareAndSet(false, true)) {
+			Map<String, Object> busy = new LinkedHashMap<>(4);
+			busy.put("accepted", false);
+			busy.put("running", true);
+			busy.put("message", "已有同步任务在执行，请稍后查看同步水位");
+			return busy;
+		}
+		try {
+			return doSyncOnce(full);
+		}
+		finally {
+			SYNC_RUNNING.set(false);
+		}
+	}
+
+	@Override
+	public Map<String, Object> submitSyncAsync(boolean full) {
+		if (!openProperties.isEnabled()) {
+			throw new IllegalStateException("酒店中转未启用");
+		}
+		if (!SYNC_RUNNING.compareAndSet(false, true)) {
+			Map<String, Object> busy = new LinkedHashMap<>(4);
+			busy.put("accepted", false);
+			busy.put("async", true);
+			busy.put("running", true);
+			busy.put("message", "已有同步任务在执行，请稍后刷新水位查看进度");
+			return busy;
+		}
+		Long tenantId = TenantContextHolder.getTenantId();
+		if (tenantId == null) {
+			tenantId = openProperties.getCallbackTenantId() == null ? 1L : openProperties.getCallbackTenantId();
+		}
+		final Long tid = tenantId;
+		markCursorRunning(full);
+		SYNC_ASYNC.execute(() -> TenantUtils.execute(tid, () -> {
+			try {
+				doSyncOnce(full);
+			}
+			catch (Exception e) {
+				log.warn("|hotel-zax-sync|异步同步失败|{}", e.getMessage(), e);
+				markCursorFailed(e.getMessage());
+			}
+			finally {
+				SYNC_RUNNING.set(false);
+			}
+		}));
+		Map<String, Object> out = new LinkedHashMap<>(6);
+		out.put("accepted", true);
+		out.put("async", true);
+		out.put("full", full);
+		out.put("message", "已提交后台同步，请刷新「同步水位」查看进度（十万级可能需数十分钟）");
+		return out;
+	}
+
+	/**
+	 * 实际同步逻辑：不加整库大事务（否则几十家详情+日历就会拖垮连接/超时）。
+	 * 列表分页串行；单页内详情+日历按配置并发拉取。
+	 */
+	private Map<String, Object> doSyncOnce(boolean full) {
 		if (!openProperties.isEnabled()) {
 			throw new IllegalStateException("酒店中转未启用");
 		}
@@ -78,18 +152,20 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService {
 		String updatedSince = full ? null : cursor.getUpdatedSince();
 		int pageNo = 1;
 		int upsert = 0;
-		int detailOk = 0;
-		int dailyOk = 0;
-		int dailyFail = 0;
-		int dailyRows = 0;
+		AtomicInteger detailOk = new AtomicInteger();
+		AtomicInteger dailyOk = new AtomicInteger();
+		AtomicInteger dailyFail = new AtomicInteger();
+		AtomicInteger dailyRows = new AtomicInteger();
 		int offlineOutOfScope = 0;
 		String nextSince = updatedSince;
 		boolean hasMore = true;
 		// 必须整页拉完（hasMore=false）才允许按授权范围清库，避免中途失败/触顶把大量酒店误标未授权
 		boolean listComplete = false;
 		Set<String> syncedCodes = new HashSet<>();
-		// 单页 50，上限覆盖约 2.5 万家；触顶仍不 listComplete
-		final int maxPages = 500;
+		Long tenantId = TenantContextHolder.getTenantId();
+		int concurrency = Math.max(1, openProperties.getSyncDetailConcurrency());
+		// 单页 50，上限覆盖约 2.5 万家；触顶仍不 listComplete（十万级需加大 maxPages）
+		final int maxPages = 3000;
 		while (hasMore && pageNo <= maxPages) {
 			Map<String, Object> req = new HashMap<>();
 			req.put("current", pageNo);
@@ -111,6 +187,7 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService {
 				listComplete = true;
 				break;
 			}
+			List<String> needDetail = new ArrayList<>();
 			for (int i = 0; i < records.size(); i++) {
 				JSONObject row = records.getJSONObject(i);
 				if (row == null || StrUtil.isBlank(row.getStr("hotelCode"))) {
@@ -124,25 +201,10 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService {
 				if (isOfflineStatus(row.getStr("status"))) {
 					continue;
 				}
-				try {
-					DetailPullResult dr = pullDetailResult(code);
-					if (dr.ok) {
-						detailOk++;
-					}
-					dailyRows += dr.dailyRows;
-					if (dr.dailyFail) {
-						dailyFail++;
-					}
-					else if (dr.ok) {
-						dailyOk++;
-					}
-					sleepQuiet(80L);
-				}
-				catch (Exception e) {
-					log.warn("|hotel-zax-sync|detail失败|code={}|{}", code, e.getMessage());
-					dailyFail++;
-				}
+				needDetail.add(code);
 			}
+			pullDetailsConcurrent(needDetail, concurrency, tenantId, detailOk, dailyOk, dailyFail, dailyRows);
+			markCursorProgress(full, pageNo, upsert, detailOk.get(), dailyRows.get(), dailyFail.get());
 			String ns = data.getStr("nextUpdatedSince");
 			if (StrUtil.isNotBlank(ns)) {
 				nextSince = ns;
@@ -164,25 +226,120 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService {
 				nextSince = LocalDateTime.now().format(DT);
 			}
 		}
+		cursor = getOrCreateCursor();
 		cursor.setUpdatedSince(nextSince);
 		cursor.setLastSyncAt(LocalDateTime.now());
-		cursor.setLastMsg((full ? "full," : "incr,") + "upsert=" + upsert + ",detail=" + detailOk
-				+ ",dailyOk=" + dailyOk + ",dailyRows=" + dailyRows + ",dailyFail=" + dailyFail
+		cursor.setLastMsg((full ? "full," : "incr,") + "upsert=" + upsert + ",detail=" + detailOk.get()
+				+ ",dailyOk=" + dailyOk.get() + ",dailyRows=" + dailyRows.get() + ",dailyFail=" + dailyFail.get()
 				+ ",offlineOut=" + offlineOutOfScope + ",listComplete=" + listComplete);
 		cursorMapper.updateById(cursor);

 		Map<String, Object> out = new LinkedHashMap<>();
+		out.put("accepted", true);
 		out.put("full", full);
 		out.put("upsert", upsert);
-		out.put("detailOk", detailOk);
-		out.put("dailyOk", Integer.valueOf(dailyOk));
-		out.put("dailyRows", Integer.valueOf(dailyRows));
-		out.put("dailyFail", Integer.valueOf(dailyFail));
+		out.put("detailOk", detailOk.get());
+		out.put("dailyOk", Integer.valueOf(dailyOk.get()));
+		out.put("dailyRows", Integer.valueOf(dailyRows.get()));
+		out.put("dailyFail", Integer.valueOf(dailyFail.get()));
 		out.put("offlineOutOfScope", offlineOutOfScope);
 		out.put("updatedSince", nextSince);
 		return out;
 	}

+	private void pullDetailsConcurrent(List<String> codes, int concurrency, Long tenantId,
+			AtomicInteger detailOk, AtomicInteger dailyOk, AtomicInteger dailyFail, AtomicInteger dailyRows) {
+		if (codes == null || codes.isEmpty()) {
+			return;
+		}
+		if (concurrency <= 1 || codes.size() == 1) {
+			for (String code : codes) {
+				pullOneDetailCounted(code, detailOk, dailyOk, dailyFail, dailyRows);
+			}
+			return;
+		}
+		ExecutorService pool = Executors.newFixedThreadPool(Math.min(concurrency, codes.size()), r -> {
+			Thread t = new Thread(r, "hotel-zax-detail");
+			t.setDaemon(true);
+			return t;
+		});
+		try {
+			List<CompletableFuture<Void>> futures = new ArrayList<>(codes.size());
+			for (String code : codes) {
+				futures.add(CompletableFuture.runAsync(() -> {
+					Runnable task = () -> pullOneDetailCounted(code, detailOk, dailyOk, dailyFail, dailyRows);
+					if (tenantId != null) {
+						TenantUtils.execute(tenantId, task);
+					}
+					else {
+						task.run();
+					}
+				}, pool));
+			}
+			CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+		}
+		finally {
+			pool.shutdown();
+		}
+	}
+
+	private void pullOneDetailCounted(String code, AtomicInteger detailOk, AtomicInteger dailyOk,
+			AtomicInteger dailyFail, AtomicInteger dailyRows) {
+		try {
+			DetailPullResult dr = pullDetailResult(code);
+			if (dr.ok) {
+				detailOk.incrementAndGet();
+			}
+			dailyRows.addAndGet(dr.dailyRows);
+			if (dr.dailyFail) {
+				dailyFail.incrementAndGet();
+			}
+			else if (dr.ok) {
+				dailyOk.incrementAndGet();
+			}
+		}
+		catch (Exception e) {
+			log.warn("|hotel-zax-sync|detail失败|code={}|{}", code, e.getMessage());
+			dailyFail.incrementAndGet();
+		}
+	}
+
+	private void markCursorRunning(boolean full) {
+		try {
+			HotelZaxSyncCursorDO cursor = getOrCreateCursor();
+			cursor.setLastMsg((full ? "running,full" : "running,incr") + ",startedAt=" + LocalDateTime.now().format(DT));
+			cursorMapper.updateById(cursor);
+		}
+		catch (Exception e) {
+			log.warn("|hotel-zax-sync|标记running失败|{}", e.getMessage());
+		}
+	}
+
+	private void markCursorProgress(boolean full, int pageNo, int upsert, int detailOk, int dailyRows, int dailyFail) {
+		try {
+			HotelZaxSyncCursorDO cursor = getOrCreateCursor();
+			cursor.setLastMsg((full ? "running,full" : "running,incr") + ",page=" + pageNo
+					+ ",upsert=" + upsert + ",detail=" + detailOk
+					+ ",dailyRows=" + dailyRows + ",dailyFail=" + dailyFail);
+			cursorMapper.updateById(cursor);
+		}
+		catch (Exception ignored) {
+			// 进度写入失败不影响主流程
+		}
+	}
+
+	private void markCursorFailed(String msg) {
+		try {
+			HotelZaxSyncCursorDO cursor = getOrCreateCursor();
+			cursor.setLastSyncAt(LocalDateTime.now());
+			cursor.setLastMsg("fail," + StrUtil.maxLength(StrUtil.blankToDefault(msg, "unknown"), 400));
+			cursorMapper.updateById(cursor);
+		}
+		catch (Exception e) {
+			log.warn("|hotel-zax-sync|标记fail失败|{}", e.getMessage());
+		}
+	}
+
 	/**
 	 * 全量同步后：不在本次授权结果集中的本地酒店标为 OUT_OF_SCOPE（非 LOCAL_OFFLINE）。
 	 * 重新授权后再同步可恢复 ACTIVE，避免「先撤权再授权」一直下架。