From 0f2a4961882f940eeea7a0182d5bde90d7cdc351 Mon Sep 17 00:00:00 2001 From: zjf <1799650030@qq.com> Date: Wed, 4 Nov 2026 14:19:41 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=8F=E5=A6=82=E9=85=92=E5=BA=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../admin/hotel/HotelZaxController.java | 5 +- .../hotel/ZaxHotelOpenProperties.java | 10 +- .../service/hotel/HotelZaxSyncService.java | 6 + .../hotel/HotelZaxSyncServiceImpl.java | 221 +++++++++++++++--- 4 files changed, 206 insertions(+), 36 deletions(-) diff --git a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/controller/admin/hotel/HotelZaxController.java b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/controller/admin/hotel/HotelZaxController.java index 5023181..b7b18c3 100644 --- a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/controller/admin/hotel/HotelZaxController.java +++ b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/controller/admin/hotel/HotelZaxController.java @@ -220,11 +220,12 @@ public class HotelZaxController { } @PostMapping("/sync") - @Operation(summary = "从资安心开放平台同步到本地（默认全量，适配授权范围变更）") + @Operation(summary = "从资安心开放平台异步同步到本地（立刻返回，进度见同步水位；默认全量）") @PreAuthorize("@ss.hasPermission('product:hotel-zax:sync')") public CommonResult> 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") diff --git a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/framework/hotel/ZaxHotelOpenProperties.java b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/framework/hotel/ZaxHotelOpenProperties.java index 496529c..fec2054 100644 --- a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/framework/hotel/ZaxHotelOpenProperties.java +++ b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/framework/hotel/ZaxHotelOpenProperties.java @@ -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; /** * 供应方回调落库使用的租户（回调无租户头时用此值）。 diff --git a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncService.java b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncService.java index d192ffe..6e3ebe5 100644 --- a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncService.java +++ b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncService.java @@ -18,6 +18,12 @@ public interface HotelZaxSyncService { */ Map syncOnce(boolean full); + /** + * 后台提交异步全量/增量同步（立刻返回，避免 HTTP 超时）。 + * @return accepted=true；若已有任务在跑则 accepted=false 并带 running 说明 + */ + Map submitSyncAsync(boolean full); + /** * 单店拉取详情并落库（资源回调触发）。 */ diff --git a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncServiceImpl.java b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncServiceImpl.java index ea0643a..a4e1ddd 100644 --- a/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncServiceImpl.java +++ b/yudao-module-mall/yudao-module-product-biz/src/main/java/cn/iocoder/yudao/module/product/service/hotel/HotelZaxSyncServiceImpl.java @@ -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 syncOnce() { // 管理端「同步资安心」默认全量：改授权范围后增量水位拉不到新授权酒店 return syncOnce(true); } @Override - @Transactional(rollbackFor = Exception.class) public Map syncOnce(boolean full) { + if (!SYNC_RUNNING.compareAndSet(false, true)) { + Map 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 submitSyncAsync(boolean full) { + if (!openProperties.isEnabled()) { + throw new IllegalStateException("酒店中转未启用"); + } + if (!SYNC_RUNNING.compareAndSet(false, true)) { + Map 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 out = new LinkedHashMap<>(6); + out.put("accepted", true); + out.put("async", true); + out.put("full", full); + out.put("message", "已提交后台同步，请刷新「同步水位」查看进度（十万级可能需数十分钟）"); + return out; + } + + /** + * 实际同步逻辑：不加整库大事务（否则几十家详情+日历就会拖垮连接/超时）。 + * 列表分页串行；单页内详情+日历按配置并发拉取。 + */ + private Map 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 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 req = new HashMap<>(); req.put("current", pageNo); @@ -111,6 +187,7 @@ public class HotelZaxSyncServiceImpl implements HotelZaxSyncService { listComplete = true; break; } + List 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 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 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> 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，避免「先撤权再授权」一直下架。