From e7d1df5ef9572af5015a24887ffd235418ec63b0 Mon Sep 17 00:00:00 2001 From: zjf <1799650030@qq.com> Date: Tue, 10 Nov 2026 10:39:33 +0800 Subject: [PATCH] =?UTF-8?q?=E9=85=92=E5=BA=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../dto/SmHotelShangyouChannelConfigDTO.java | 8 +- .../SmHotelDownstreamPushLogEntity.java | 8 + .../config/HotelDownstreamProperties.java | 15 + .../config/ShangyouHotelProperties.java | 14 +- .../SmHotelDownstreamPushLogController.java | 43 ++- ...HotelDownstreamPushLogCleanupSchedule.java | 15 +- .../SmHotelDownstreamPushLogMapper.java | 74 +++++ .../SmHotelDownstreamPushLogService.java | 18 +- .../service/SmHotelMidService.java | 5 + .../impl/SmHotelChannelConfigServiceImpl.java | 6 + .../SmHotelDownstreamPushLogServiceImpl.java | 304 +++++++++++++++++- .../service/impl/SmHotelMidServiceImpl.java | 43 ++- .../service/impl/SmHotelSyncServiceImpl.java | 78 ++++- 13 files changed, 580 insertions(+), 51 deletions(-) diff --git a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelShangyouChannelConfigDTO.java b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelShangyouChannelConfigDTO.java index dd88b3136..1168dbb1a 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelShangyouChannelConfigDTO.java +++ b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelShangyouChannelConfigDTO.java @@ -112,9 +112,15 @@ public class SmHotelShangyouChannelConfigDTO { private Integer detailPullIntervalMs; - /** 全量详情酒店并发；建议 4~8 */ + /** 全量/范围详情酒店并发；建议 4~8 */ private Integer detailPullConcurrency; + /** 近窗详情酒店并发；默认 1，边拉边推时硬顶 1 */ + private Integer nearDetailPullConcurrency; + + /** 近窗每家拉完间隔毫秒；默认 200，给业务查询让路 */ + private Integer nearDetailPullIntervalMs; + /** 全量详情房价天数；≤0 用 priceDays；建议 15~30 */ private Integer detailPullPriceDays; diff --git a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelDownstreamPushLogEntity.java b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelDownstreamPushLogEntity.java index fa108d2c4..d79c7e371 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelDownstreamPushLogEntity.java +++ b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelDownstreamPushLogEntity.java @@ -75,4 +75,12 @@ public class SmHotelDownstreamPushLogEntity extends Model diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/ShangyouHotelProperties.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/ShangyouHotelProperties.java index 27c459a4f..841e664a7 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/ShangyouHotelProperties.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/ShangyouHotelProperties.java @@ -225,11 +225,23 @@ public class ShangyouHotelProperties { private int detailPullIntervalMs = 0; /** - * 全量拉详情并发度（酒店级并行）。对方房价约 20 QPS，建议 4~8；1=串行旧行为。 + * 全量/范围拉详情并发度（酒店级并行）。对方房价约 20 QPS，建议 4~8；1=串行旧行为。 * 多实例时请降低，避免合计超限。 */ private int detailPullConcurrency = 6; + /** + * 近窗详情并发度（默认 1，低于全量）。近窗常带边拉边推，并发过高会占满库连接导致整站变卡。 + * ≤0 时按 1；边拉边推时硬顶 1，且推送串行。 + */ + private int nearDetailPullConcurrency = 1; + + /** + * 近窗详情：每家酒店拉完后的间隔毫秒（给业务查询让路）。 + * 默认 200；0=不额外等待。比全量 detail-pull-interval-ms 更积极节流。 + */ + private int nearDetailPullIntervalMs = 200; + /** * 全量拉详情房价天数；≤0 则用 {@link #priceDays}。 * 圣诺只推约 15 天时可配 15~30，比默认 90 天少 2/3 次 getRoomPrice。 diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java index c533b82e2..3110e5bb5 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java @@ -15,10 +15,17 @@ import io.swagger.v3.oas.annotations.security.SecurityRequirement; import io.swagger.v3.oas.annotations.tags.Tag; import lombok.RequiredArgsConstructor; import org.springdoc.core.annotations.ParameterObject; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.*; +import java.io.File; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; @@ -93,11 +100,45 @@ public class SmHotelDownstreamPushLogController { } @SysLog("按时间区间删除下游推送流水") - @Operation(summary = "按创建时间区间删除推送流水") + @Operation(summary = "按创建时间区间分批物理删除推送流水（大批量请多次点）") @PostMapping("/range") @PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_del')") public R removeRange(@RequestParam(required = false) String createTimeBegin, @RequestParam(required = false) String createTimeEnd) { return pushLogService.removeByCreateTimeRange(createTimeBegin, createTimeEnd); } + + @SysLog("归档并清理早于N天的下游推送流水") + @Operation(summary = "过期流水先归档 CSV.GZ 再物理删除（默认保留 15 天；单次约最多 200 万，可多次）") + @PostMapping("/cleanupOlder") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_del')") + public R> cleanupOlder(@RequestParam(defaultValue = "15") int retainDays) { + if (retainDays < 1) { + return R.failed("retainDays 至少为 1"); + } + return pushLogService.cleanupOlderThanDays(retainDays); + } + + @Operation(summary = "归档文件列表（本地 CSV.GZ）") + @GetMapping("/archives") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_view')") + public R>> listArchives() { + return R.ok(pushLogService.listArchiveFiles()); + } + + @Operation(summary = "下载归档文件") + @GetMapping("/archives/download") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_view')") + public ResponseEntity downloadArchive(@RequestParam String fileName) { + File file = pushLogService.resolveArchiveFile(fileName); + if (file == null) { + return ResponseEntity.notFound().build(); + } + String encoded = URLEncoder.encode(file.getName(), StandardCharsets.UTF_8).replace("+", "%20"); + return ResponseEntity.ok() + .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename*=UTF-8''" + encoded) + .contentType(MediaType.APPLICATION_OCTET_STREAM) + .contentLength(file.length()) + .body(new FileSystemResource(file)); + } } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/job/SmHotelDownstreamPushLogCleanupSchedule.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/job/SmHotelDownstreamPushLogCleanupSchedule.java index 21feebcef..e5a8f08f5 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/job/SmHotelDownstreamPushLogCleanupSchedule.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/job/SmHotelDownstreamPushLogCleanupSchedule.java @@ -34,9 +34,18 @@ public class SmHotelDownstreamPushLogCleanupSchedule { log.debug("|下游推送流水|清理跳过|retainDays={}|", retainDays); return; } - int n = pushLogService.cleanupOlderThanDays(retainDays); - if (n > 0) { - log.info("|下游推送流水|清理完成|retainDays={}|rows={}|", retainDays, n); + var r = pushLogService.cleanupOlderThanDays(retainDays); + if (r != null && r.getCode() == 0 && r.getData() != null) { + Object deleted = r.getData().get("deletedRows"); + Object archived = r.getData().get("archivedRows"); + if ((deleted instanceof Number n && n.intValue() > 0) + || (archived instanceof Number a && a.intValue() > 0)) { + log.info("|下游推送流水|归档清理完成|retainDays={}|archived={}|deleted={}|msg={}|", + retainDays, archived, deleted, r.getMsg()); + } + } + else if (r != null && r.getCode() != 0) { + log.warn("|下游推送流水|归档清理失败|retainDays={}|msg={}|", retainDays, r.getMsg()); } }); } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/mapper/SmHotelDownstreamPushLogMapper.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/mapper/SmHotelDownstreamPushLogMapper.java index a4a9c5285..2afd243fc 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/mapper/SmHotelDownstreamPushLogMapper.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/mapper/SmHotelDownstreamPushLogMapper.java @@ -2,15 +2,89 @@ package com.bsd.pjcloud.supermarket.mapper; import com.bsd.pjcloud.common.data.datascope.PjcloudBaseMapper; import com.bsd.pjcloud.supermarket.entity.SmHotelDownstreamPushLogEntity; +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.LocalDateTime; import java.util.List; @Mapper public interface SmHotelDownstreamPushLogMapper extends PjcloudBaseMapper { + /** 归档用：按 id 升序取早于 before 的一批（含逻辑删行，绕过 @TableLogic） */ + @Select(""" + SELECT id, tenant_id, partner_code, sync_log_id, hotel_id, ext_hotel_id, action, + request_url, request_body, response_body, http_status, biz_code, success, + error_msg, cost_ms, create_by, create_time, update_by, update_time, del_flag + FROM canteen_hotel_downstream_push_log + WHERE create_time < #{before} + ORDER BY id ASC + LIMIT #{limit} + """) + List listOldestBefore(@Param("before") LocalDateTime before, + @Param("limit") int limit); + + /** 归档用：创建时间闭区间一批 */ + @Select(""" + SELECT id, tenant_id, partner_code, sync_log_id, hotel_id, ext_hotel_id, action, + request_url, request_body, response_body, http_status, biz_code, success, + error_msg, cost_ms, create_by, create_time, update_by, update_time, del_flag + FROM canteen_hotel_downstream_push_log + WHERE create_time >= #{begin} AND create_time <= #{end} + ORDER BY id ASC + LIMIT #{limit} + """) + List listOldestBetween(@Param("begin") LocalDateTime begin, + @Param("end") LocalDateTime end, @Param("limit") int limit); + + /** 物理删除单条（绕过逻辑删） */ + @Delete("DELETE FROM canteen_hotel_downstream_push_log WHERE id = #{id}") + int physicalDeleteById(@Param("id") Long id); + + /** 物理删除一批 id（归档成功后调用） */ + @Delete(""" + + """) + int physicalDeleteByIds(@Param("ids") List ids); + + /** 物理删除：早于 before 的流水（含已逻辑删），分批 LIMIT */ + @Delete(""" + DELETE FROM canteen_hotel_downstream_push_log + WHERE create_time < #{before} + LIMIT #{limit} + """) + int physicalDeleteBefore(@Param("before") LocalDateTime before, @Param("limit") int limit); + + /** 物理删除：创建时间落在区间内，分批 LIMIT */ + @Delete(""" + DELETE FROM canteen_hotel_downstream_push_log + WHERE create_time >= #{begin} AND create_time <= #{end} + LIMIT #{limit} + """) + int physicalDeleteBetween(@Param("begin") LocalDateTime begin, @Param("end") LocalDateTime end, + @Param("limit") int limit); + + /** 物理删除：仅一端起 */ + @Delete(""" + DELETE FROM canteen_hotel_downstream_push_log + WHERE create_time >= #{begin} + LIMIT #{limit} + """) + int physicalDeleteFrom(@Param("begin") LocalDateTime begin, @Param("limit") int limit); + + /** 物理删除：仅一端止 */ + @Delete(""" + DELETE FROM canteen_hotel_downstream_push_log + WHERE create_time <= #{end} + LIMIT #{limit} + """) + int physicalDeleteTo(@Param("end") LocalDateTime end, @Param("limit") int limit); + /** * 仅取本页去重酒店 ID（走 sync_log_id+success+hotel_id 索引），避免整表 GROUP BY MAX。 */ diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java index 4bc812ce9..371e96720 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java @@ -7,6 +7,7 @@ import com.bsd.pjcloud.supermarket.dto.SmHotelDownstreamGapExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelDownstreamPushLogEntity; import java.util.List; +import java.util.Map; public interface SmHotelDownstreamPushLogService extends IService { @@ -16,15 +17,24 @@ public interface SmHotelDownstreamPushLogService extends IService> cleanupOlderThanDays(int retainDays); - /** 删除单条推送流水 */ + /** 删除单条推送流水（物理删） */ R removeLog(Long id); - /** 按创建时间区间删除（至少指定一端） */ + /** 按创建时间区间分批处理（默认先归档再物理删）；超大批量可能需点多次 */ R removeByCreateTimeRange(String createTimeBegin, String createTimeEnd); + /** 归档文件列表（文件名/大小/修改时间） */ + List> listArchiveFiles(); + + /** 解析可下载的归档文件（防路径穿越）；不存在返回 null */ + java.io.File resolveArchiveFile(String fileName); + /** * 按某次下游全量推同步日志导出完善清单（不限行数）： * ① 圣诺 getInfo 无店/失败；② getInfo 成功但房型/产品编码匹配不上。 diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java index 47e7603d2..24d23314b 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelMidService.java @@ -24,6 +24,11 @@ public interface SmHotelMidService { */ R> pullFromUpstream(Long hotelId, Integer priceDaysOverride); + /** + * @param scheduleAutoPush false 时不触发下游防抖自动推（近窗边拉边推已自行推送） + */ + R> pullFromUpstream(Long hotelId, Integer priceDaysOverride, boolean scheduleAutoPush); + /** * 仅调用 getHotelComplexList 落房型（不拉房价），用于单独验证综合信息接口。 */ diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelChannelConfigServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelChannelConfigServiceImpl.java index cfda6355f..a006654ac 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelChannelConfigServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelChannelConfigServiceImpl.java @@ -526,6 +526,12 @@ public class SmHotelChannelConfigServiceImpl extends ServiceImpl 0) { + target.setNearDetailPullConcurrency(dto.getNearDetailPullConcurrency()); + } + if (dto.getNearDetailPullIntervalMs() != null && dto.getNearDetailPullIntervalMs() >= 0) { + target.setNearDetailPullIntervalMs(dto.getNearDetailPullIntervalMs()); + } if (dto.getAutoImportNewHotels() != null) { target.setAutoImportNewHotels(dto.getAutoImportNewHotels()); } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java index 7be819d55..00556aa67 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java @@ -13,6 +13,7 @@ import com.bsd.pjcloud.supermarket.dto.SmHotelDownstreamGapExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelDownstreamPushLogEntity; import com.bsd.pjcloud.supermarket.entity.SmHotelEntity; import com.bsd.pjcloud.supermarket.entity.SmHotelSyncLogEntity; +import com.bsd.pjcloud.supermarket.config.HotelDownstreamProperties; import com.bsd.pjcloud.supermarket.mapper.SmHotelDownstreamPushLogMapper; import com.bsd.pjcloud.supermarket.mapper.SmHotelMapper; import com.bsd.pjcloud.supermarket.mapper.SmHotelSyncLogMapper; @@ -21,16 +22,27 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import java.io.BufferedWriter; +import java.io.File; +import java.io.OutputStreamWriter; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; +import java.util.zip.GZIPOutputStream; @Slf4j @Service @@ -43,6 +55,19 @@ public class SmHotelDownstreamPushLogServiceImpl private final SmHotelMapper hotelMapper; + private final HotelDownstreamProperties downstreamProperties; + + /** 列表未带时间时默认只查近 N 天，避免 百万级 COUNT/排序拖垮页面 */ + private static final int DEFAULT_PAGE_DAYS = 3; + + /** 单次归档+删除每批行数 */ + private static final int PHYSICAL_DELETE_BATCH = 5000; + + /** 单次接口最多跑多少批（约 200 万行），超出提示再点 */ + private static final int PHYSICAL_DELETE_MAX_ROUNDS = 400; + + private static final DateTimeFormatter ARCHIVE_TS = DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss"); + @Override public Page pageLogs(Page page, SmHotelDownstreamPushLogEntity query) { @@ -53,6 +78,12 @@ public class SmHotelDownstreamPushLogServiceImpl String extHotelId = query == null ? null : query.getExtHotelId(); String action = query == null ? null : query.getAction(); String success = query == null ? null : query.getSuccess(); + LocalDateTime begin = parseDateTime(query == null ? null : query.getCreateTimeBegin(), true); + LocalDateTime end = parseDateTime(query == null ? null : query.getCreateTimeEnd(), false); + // 无时间窗：默认近 3 天（有 syncLogId/hotelId 精确定位时不强制，避免查不到历史） + if (begin == null && end == null && syncLogId == null && hotelId == null) { + begin = LocalDateTime.now().minusDays(DEFAULT_PAGE_DAYS); + } return baseMapper.selectPage(page, Wrappers.lambdaQuery() .select(SmHotelDownstreamPushLogEntity.class, info -> { String col = info.getColumn(); @@ -64,7 +95,9 @@ public class SmHotelDownstreamPushLogServiceImpl .eq(StrUtil.isNotBlank(extHotelId), SmHotelDownstreamPushLogEntity::getExtHotelId, extHotelId) .eq(StrUtil.isNotBlank(action), SmHotelDownstreamPushLogEntity::getAction, action) .eq(StrUtil.isNotBlank(success), SmHotelDownstreamPushLogEntity::getSuccess, success) - .orderByDesc(SmHotelDownstreamPushLogEntity::getCreateTime)); + .ge(begin != null, SmHotelDownstreamPushLogEntity::getCreateTime, begin) + .le(end != null, SmHotelDownstreamPushLogEntity::getCreateTime, end) + .orderByDesc(SmHotelDownstreamPushLogEntity::getId)); } @Override @@ -82,14 +115,12 @@ public class SmHotelDownstreamPushLogServiceImpl } @Override - public int cleanupOlderThanDays(int retainDays) { + public R> cleanupOlderThanDays(int retainDays) { if (retainDays <= 0) { - return 0; + return R.failed("retainDays 须大于 0"); } LocalDateTime before = LocalDateTime.now().minusDays(retainDays); - // @TableLogic：delete 实际为逻辑删除 - return baseMapper.delete(Wrappers.lambdaQuery() - .lt(SmHotelDownstreamPushLogEntity::getCreateTime, before)); + return archiveThenDeleteBefore(before, "cleanup_" + retainDays + "d"); } @Override @@ -101,7 +132,7 @@ public class SmHotelDownstreamPushLogServiceImpl if (row == null) { return R.failed("流水不存在"); } - baseMapper.deleteById(id); + baseMapper.physicalDeleteById(id); return R.ok(Boolean.TRUE, "已删除"); } @@ -115,17 +146,258 @@ public class SmHotelDownstreamPushLogServiceImpl if (begin != null && end != null && end.isBefore(begin)) { return R.failed("结束时间不能早于开始时间"); } - var qw = Wrappers.lambdaQuery(); - if (begin != null) { - qw.ge(SmHotelDownstreamPushLogEntity::getCreateTime, begin); + R> r; + if (begin != null && end != null) { + r = archiveThenDeleteBetween(begin, end, "range"); } - if (end != null) { - qw.le(SmHotelDownstreamPushLogEntity::getCreateTime, end); + else if (end != null) { + r = archiveThenDeleteBefore(end.plusSeconds(1), "to_" + end.toLocalDate()); } - Long cnt = baseMapper.selectCount(qw); - int n = baseMapper.delete(qw); - log.info("|下游推送流水|按时间删除|begin={}|end={}|deleted={}|matched={}|", begin, end, n, cnt); - return R.ok(n, "已删除 " + n + " 条"); + else { + r = archiveThenDeleteBetween(begin, LocalDateTime.now(), "from_" + begin.toLocalDate()); + } + if (r.getCode() != 0) { + return R.failed(r.getMsg()); + } + int deleted = r.getData() == null ? 0 + : ((Number) r.getData().getOrDefault("deletedRows", 0)).intValue(); + String msg = r.getData() == null ? null : String.valueOf(r.getData().getOrDefault("msg", "")); + return R.ok(deleted, StrUtil.blankToDefault(msg, "已处理 " + deleted + " 条")); + } + + @Override + public List> listArchiveFiles() { + File dir = resolveArchiveDir().toFile(); + if (!dir.isDirectory()) { + return List.of(); + } + File[] files = dir.listFiles((d, name) -> name != null && name.endsWith(".csv.gz")); + if (files == null || files.length == 0) { + return List.of(); + } + Arrays.sort(files, Comparator.comparingLong(File::lastModified).reversed()); + List> out = new ArrayList<>(files.length); + for (File f : files) { + Map m = new LinkedHashMap<>(6); + m.put("fileName", f.getName()); + m.put("sizeBytes", f.length()); + m.put("sizeMb", Math.round(f.length() / 1024.0 / 1024.0 * 100) / 100.0); + m.put("lastModified", f.lastModified()); + out.add(m); + } + return out; + } + + @Override + public File resolveArchiveFile(String fileName) { + if (StrUtil.isBlank(fileName) || fileName.contains("..") || fileName.contains("/") || fileName.contains("\\") + || !fileName.endsWith(".csv.gz")) { + return null; + } + Path base = resolveArchiveDir().toAbsolutePath().normalize(); + Path target = base.resolve(fileName).normalize(); + if (!target.startsWith(base) || !Files.isRegularFile(target)) { + return null; + } + return target.toFile(); + } + + /** + * 归档 create_time < before 的数据后物理删除。 + */ + private R> archiveThenDeleteBefore(LocalDateTime before, String tag) { + boolean archiveOn = downstreamProperties == null || downstreamProperties.isPushLogArchiveEnabled(); + int archived = 0; + int deleted = 0; + List files = new ArrayList<>(); + boolean truncated = false; + try { + Files.createDirectories(resolveArchiveDir()); + } + catch (Exception e) { + return R.failed("无法创建归档目录：" + e.getMessage()); + } + for (int i = 0; i < PHYSICAL_DELETE_MAX_ROUNDS; i++) { + List batch = baseMapper.listOldestBefore(before, PHYSICAL_DELETE_BATCH); + if (CollUtil.isEmpty(batch)) { + break; + } + List ids = batch.stream().map(SmHotelDownstreamPushLogEntity::getId).filter(id -> id != null) + .toList(); + if (ids.isEmpty()) { + break; + } + if (archiveOn) { + try { + String name = writeArchiveCsvGz(batch, tag); + files.add(name); + archived += batch.size(); + } + catch (Exception e) { + log.error("|下游推送流水|归档失败，中止删除以免丢历史|err={}|", e.getMessage(), e); + return R.failed("归档失败已中止删除：" + e.getMessage() + "（已删 " + deleted + "，已归档 " + + archived + "）"); + } + } + int n = baseMapper.physicalDeleteByIds(ids); + deleted += Math.max(0, n); + if (batch.size() < PHYSICAL_DELETE_BATCH) { + break; + } + if (i == PHYSICAL_DELETE_MAX_ROUNDS - 1) { + truncated = true; + } + } + Map data = new LinkedHashMap<>(8); + data.put("archivedRows", archived); + data.put("deletedRows", deleted); + data.put("files", files); + data.put("archiveEnabled", archiveOn); + data.put("archiveDir", resolveArchiveDir().toAbsolutePath().toString()); + data.put("truncated", truncated); + data.put("before", before.toString()); + String msg = archiveOn + ? ("已归档 " + archived + " 条到 " + files.size() + " 个文件，并删除 " + deleted + " 条" + + (truncated ? "（未清完请再点）" : "")) + : ("归档关闭，已直接删除 " + deleted + " 条" + (truncated ? "（未清完请再点）" : "")); + data.put("msg", msg); + log.info("|下游推送流水|归档清理|tag={}|before={}|archived={}|deleted={}|files={}|truncated={}|", tag, before, + archived, deleted, files.size(), truncated); + return R.ok(data, msg); + } + + private R> archiveThenDeleteBetween(LocalDateTime begin, LocalDateTime end, String tag) { + boolean archiveOn = downstreamProperties == null || downstreamProperties.isPushLogArchiveEnabled(); + int archived = 0; + int deleted = 0; + List files = new ArrayList<>(); + boolean truncated = false; + try { + Files.createDirectories(resolveArchiveDir()); + } + catch (Exception e) { + return R.failed("无法创建归档目录：" + e.getMessage()); + } + for (int i = 0; i < PHYSICAL_DELETE_MAX_ROUNDS; i++) { + List batch = baseMapper.listOldestBetween(begin, end, PHYSICAL_DELETE_BATCH); + if (CollUtil.isEmpty(batch)) { + break; + } + List ids = batch.stream().map(SmHotelDownstreamPushLogEntity::getId).filter(id -> id != null) + .toList(); + if (ids.isEmpty()) { + break; + } + if (archiveOn) { + try { + files.add(writeArchiveCsvGz(batch, tag)); + archived += batch.size(); + } + catch (Exception e) { + log.error("|下游推送流水|区间归档失败|err={}|", e.getMessage(), e); + return R.failed("归档失败已中止删除：" + e.getMessage() + "（已删 " + deleted + "）"); + } + } + deleted += Math.max(0, baseMapper.physicalDeleteByIds(ids)); + if (batch.size() < PHYSICAL_DELETE_BATCH) { + break; + } + if (i == PHYSICAL_DELETE_MAX_ROUNDS - 1) { + truncated = true; + } + } + Map data = new LinkedHashMap<>(8); + data.put("archivedRows", archived); + data.put("deletedRows", deleted); + data.put("files", files); + data.put("truncated", truncated); + String msg = "已归档 " + archived + " / 删除 " + deleted + (truncated ? "（未清完请再点）" : ""); + data.put("msg", msg); + return R.ok(data, msg); + } + + private Path resolveArchiveDir() { + String configured = downstreamProperties == null ? null : downstreamProperties.getPushLogArchiveDir(); + if (StrUtil.isNotBlank(configured)) { + return Path.of(configured.trim()); + } + return Path.of("data", "sm-hotel-push-log-archive"); + } + + private String writeArchiveCsvGz(List rows, String tag) throws Exception { + String ts = LocalDateTime.now().format(ARCHIVE_TS); + Long minId = rows.get(0).getId(); + Long maxId = rows.get(rows.size() - 1).getId(); + String safeTag = StrUtil.blankToDefault(tag, "batch").replaceAll("[^a-zA-Z0-9_-]", "_"); + String fileName = "push_log_" + safeTag + "_" + ts + "_" + minId + "_" + maxId + "_" + rows.size() + + ".csv.gz"; + Path target = resolveArchiveDir().resolve(fileName); + int bodyMax = downstreamProperties == null ? 4096 : downstreamProperties.getPushLogArchiveBodyMaxChars(); + try (GZIPOutputStream gzip = new GZIPOutputStream(Files.newOutputStream(target, StandardOpenOption.CREATE_NEW)); + BufferedWriter w = new BufferedWriter(new OutputStreamWriter(gzip, StandardCharsets.UTF_8))) { + // Excel 友好 BOM + w.write('\uFEFF'); + w.write( + "id,tenant_id,partner_code,sync_log_id,hotel_id,ext_hotel_id,action,request_url,http_status,biz_code,success,error_msg,cost_ms,create_by,create_time,del_flag,request_body,response_body"); + w.newLine(); + for (SmHotelDownstreamPushLogEntity r : rows) { + w.write(csv(r.getId())); + w.write(','); + w.write(csv(r.getTenantId())); + w.write(','); + w.write(csv(r.getPartnerCode())); + w.write(','); + w.write(csv(r.getSyncLogId())); + w.write(','); + w.write(csv(r.getHotelId())); + w.write(','); + w.write(csv(r.getExtHotelId())); + w.write(','); + w.write(csv(r.getAction())); + w.write(','); + w.write(csv(r.getRequestUrl())); + w.write(','); + w.write(csv(r.getHttpStatus())); + w.write(','); + w.write(csv(r.getBizCode())); + w.write(','); + w.write(csv(r.getSuccess())); + w.write(','); + w.write(csv(r.getErrorMsg())); + w.write(','); + w.write(csv(r.getCostMs())); + w.write(','); + w.write(csv(r.getCreateBy())); + w.write(','); + w.write(csv(r.getCreateTime() == null ? null : r.getCreateTime().toString())); + w.write(','); + w.write(csv(r.getDelFlag())); + w.write(','); + w.write(csv(truncate(r.getRequestBody(), bodyMax))); + w.write(','); + w.write(csv(truncate(r.getResponseBody(), bodyMax))); + w.newLine(); + } + } + return fileName; + } + + private static String truncate(String s, int max) { + if (s == null || max <= 0 || s.length() <= max) { + return s; + } + return s.substring(0, max) + "...(truncated)"; + } + + private static String csv(Object v) { + if (v == null) { + return ""; + } + String s = String.valueOf(v); + if (s.indexOf(',') >= 0 || s.indexOf('"') >= 0 || s.indexOf('\n') >= 0 || s.indexOf('\r') >= 0) { + return '"' + s.replace("\"", "\"\"") + '"'; + } + return s; } @Override diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java index f618e8923..ee90c6b4b 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelMidServiceImpl.java @@ -46,7 +46,9 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Lazy; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import java.math.BigDecimal; import java.time.LocalDate; @@ -124,6 +126,9 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { private final SmHotelMerchantScopeHelper hotelMerchantScopeHelper; + /** 拉上游：HTTP 在事务外，仅落库走短事务，避免长时间占连接池拖慢整站 */ + private final TransactionTemplate hotelPullTx; + public SmHotelMidServiceImpl(ShangyouHotelClient upstreamClient, ShangyouHotelProperties upstreamProperties, DownstreamHotelClient downstreamClient, HotelDownstreamProperties downstreamProperties, SmHotelPriceConfigService priceConfigService, SmHotelChannelConfigService channelConfigService, @@ -135,7 +140,7 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { StringRedisTemplate stringRedisTemplate, ShengnuoDownstreamHotelAdapter shengnuoAdapter, ShengnuoIdMappingCache shengnuoIdMappingCache, HotelUpstreamCancelPolicyRegistry upstreamCancelPolicyRegistry, - SmHotelMerchantScopeHelper hotelMerchantScopeHelper) { + SmHotelMerchantScopeHelper hotelMerchantScopeHelper, PlatformTransactionManager transactionManager) { this.upstreamClient = upstreamClient; this.upstreamProperties = upstreamProperties; this.downstreamClient = downstreamClient; @@ -156,6 +161,7 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { this.dailyMapper = dailyMapper; this.stringRedisTemplate = stringRedisTemplate; this.hotelMerchantScopeHelper = hotelMerchantScopeHelper; + this.hotelPullTx = new TransactionTemplate(transactionManager); } @Override @@ -165,8 +171,16 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { } @Override - @Transactional(rollbackFor = Exception.class) public R> pullFromUpstream(Long hotelId, Integer priceDaysOverride) { + return pullFromUpstream(hotelId, priceDaysOverride, true); + } + + /** + * @param scheduleAutoPush false：近窗边拉边推等场景已自行推送，避免再防抖自动推加重库压 + */ + @Override + public R> pullFromUpstream(Long hotelId, Integer priceDaysOverride, + boolean scheduleAutoPush) { SmHotelEntity hotel = requireChannelWritableHotel(hotelId, SmHotelDataSource.SHANGYOU); String hotelCode = hotel.getHotelCode(); List upstreamTips = new ArrayList<>(); @@ -174,7 +188,7 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { log.info("|酒店拉取上游|开始|hotelId={}|hotelCode={}|hotelName={}|dataSource={}|", hotelId, hotelCode, hotel.getHotelName(), hotel.getSheetName()); - // 1) 综合信息：静态酒店 + 房型 + // 1) 综合信息 HTTP（事务外，不占库连接） JSONObject complexResp = upstreamClient.getHotelComplexListRaw(Collections.singletonList(hotelCode)); String complexMsg = complexResp.getString("message"); JSONArray complexList = asContentArray(complexResp.get("content")); @@ -193,10 +207,8 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { } applyHotelComplex(hotel, complex); stampChannelIfBlank(hotel, SmHotelDataSource.SHANGYOU); - hotelMapper.updateById(hotel); - int roomCount = upsertRoomsFromComplex(hotel, complex); - // 2) 房价房量：按 30 天切片拉取，合并产品与日历（全量详情可缩窗减少切片次数） + // 2) 房价房量 HTTP：按 30 天切片，仅内存合并 int configuredDays = upstreamProperties.getPriceDays() > 0 ? upstreamProperties.getPriceDays() : downstreamProperties.getDailyDays(); if (priceDaysOverride != null && priceDaysOverride > 0) { @@ -235,9 +247,18 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { cursor = chunkEnd; } - ensureRoomsBatch(hotel, roomTypesFromPrice); - int productCount = persistProducts(hotel, productMap); - int dailyCount = persistDailies(hotel, dailyDrafts, start, endExclusive); + // 3) 短事务只落库（房型/产品/日历），尽快归还连接 + final int[] counts = new int[3]; + hotelPullTx.executeWithoutResult(status -> { + hotelMapper.updateById(hotel); + counts[0] = upsertRoomsFromComplex(hotel, complex); + ensureRoomsBatch(hotel, roomTypesFromPrice); + counts[1] = persistProducts(hotel, productMap); + counts[2] = persistDailies(hotel, dailyDrafts, start, endExclusive); + }); + int roomCount = counts[0]; + int productCount = counts[1]; + int dailyCount = counts[2]; if (roomCount == 0 && productCount == 0) { try { @@ -267,7 +288,9 @@ public class SmHotelMidServiceImpl implements SmHotelMidService { result.put("upstreamTips", upstreamTips); log.info("|酒店拉取上游|结束|hotelCode={}|room={}|product={}|daily={}|tips={}|", hotelCode, roomCount, productCount, dailyCount, upstreamTips); - downstreamAutoPushService.scheduleAfterHotelChanged(hotelId); + if (scheduleAutoPush) { + downstreamAutoPushService.scheduleAfterHotelChanged(hotelId); + } availabilityNotifyService.scheduleCheckAfterHotelChanged(hotelId); return R.ok(result); } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java index bbbefd558..28afa60c5 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java @@ -50,6 +50,7 @@ import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -2069,8 +2070,14 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { final Object failLock = new Object(); ExecutorService pullPool = null; try { - int concurrency = Math.max(1, props().getDetailPullConcurrency()); - int intervalMs = Math.max(0, props().getDetailPullIntervalMs()); + SmHotelSyncLogEntity jobMeta = syncLogMapper.selectById(logId); + boolean nearJob = jobMeta != null && TYPE_DETAIL_NEAR.equals(jobMeta.getSyncType()); + final String thenPushPartner = resolveNearThenPushPartner(jobMeta); + // 近窗默认低并发；边拉边推硬顶 1，避免占满连接池导致整站卡顿 + int concurrency = resolveScopeOrNearConcurrency(nearJob, StrUtil.isNotBlank(thenPushPartner)); + // 近窗用更积极的间隔给业务让路；全量/范围仍用 detail-pull-interval-ms + int intervalMs = nearJob ? Math.max(0, props().getNearDetailPullIntervalMs()) + : Math.max(0, props().getDetailPullIntervalMs()); Integer daysOverride = priceDaysOverride; if (daysOverride == null || daysOverride <= 0) { daysOverride = props().getDetailPullPriceDays() > 0 ? props().getDetailPullPriceDays() : null; @@ -2082,12 +2089,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { tenantId = props().getCallbackTenantId() == null ? 1L : props().getCallbackTenantId(); } final Long tid = tenantId; - final String thenPushPartner = resolveNearThenPushPartner(logId); + final Semaphore pushGate = StrUtil.isNotBlank(thenPushPartner) ? new Semaphore(1) : null; progressSnapshot(logId, Math.min(95, 5 + (int) (processed.get() * 90L / Math.max(1, total))), "PULL", (baseProcessed > 0 ? "继续" : "开始") + "详情拉取，目标 " + total + " 家上游，已处理 " + processed.get() + "，并发=" + concurrency + " price-days=" + effectivePriceDays - + (StrUtil.isNotBlank(thenPushPartner) ? "，边拉边推→" + thenPushPartner : "") + + (StrUtil.isNotBlank(thenPushPartner) ? "，边拉边推→" + thenPushPartner + "（推送串行）" : "") + "…", ok.get(), fail.get(), skip.get(), total, failLines); @@ -2100,7 +2107,8 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { th.setDaemon(true); return th; }); - int batchSize = Math.max(concurrency * 2, 20); + // 近窗小批次，方便中途让出 CPU/连接；全量仍用较大批 + int batchSize = nearJob ? Math.max(concurrency, 4) : Math.max(concurrency * 2, 20); for (int i = 0; i < hotelIds.size(); i += batchSize) { if (handleStopIfNeeded(logId, t0, total, ok, fail, skip, processed, failLines)) { return; @@ -2126,12 +2134,14 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { skip.incrementAndGet(); return; } + // 边拉边推时关闭防抖自动推，避免拉完又排队推一遍 + boolean scheduleAuto = StrUtil.isBlank(thenPushPartner); R> r = midService.pullFromUpstream(h.getId(), - priceDaysParam); + priceDaysParam, scheduleAuto); if (r != null && r.getCode() == 0) { ok.incrementAndGet(); if (StrUtil.isNotBlank(thenPushPartner)) { - pushOneAfterNearPull(h, thenPushPartner, pushOk, pushFail, + pushOneAfterNearPull(h, thenPushPartner, pushGate, pushOk, pushFail, failLines, failLock); } } @@ -2193,6 +2203,16 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { "拉取中 " + processed.get() + "/" + total + " 成功 " + ok.get() + " 失败 " + fail.get() + " 跳过 " + skip.get(), ok.get(), fail.get(), skip.get(), total, failLines); + // 近窗批次间再让出一小段，减轻业务高峰抢连接 + if (nearJob && intervalMs > 0 && i + batchSize < hotelIds.size()) { + try { + Thread.sleep(Math.min(1000, intervalMs)); + } + catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } + } } } @@ -2247,12 +2267,23 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { } } - /** 近窗单伙伴且 YAML 开了 then-push 时，返回目标伙伴编码；否则 null。 */ - private String resolveNearThenPushPartner(Long logId) { - if (logId == null) { - return null; + /** 近窗用低并发；边拉边推硬顶 1，避免占满库连接拖慢整站。 */ + private int resolveScopeOrNearConcurrency(boolean nearJob, boolean thenPush) { + if (nearJob) { + int near = props().getNearDetailPullConcurrency(); + if (near <= 0) { + near = 1; + } + if (thenPush) { + return 1; + } + return Math.max(1, Math.min(2, near)); } - SmHotelSyncLogEntity logEntity = syncLogMapper.selectById(logId); + return Math.max(1, props().getDetailPullConcurrency()); + } + + /** 近窗单伙伴且 YAML 开了 then-push 时，返回目标伙伴编码；否则 null。 */ + private String resolveNearThenPushPartner(SmHotelSyncLogEntity logEntity) { if (logEntity == null || !TYPE_DETAIL_NEAR.equals(logEntity.getSyncType())) { return null; } @@ -2267,13 +2298,18 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { return partner; } - /** 单店近窗拉成功后立刻推该下游；推失败不影响拉取成功计数，仅记 failDetail。 */ - private void pushOneAfterNearPull(SmHotelEntity hotel, String partner, AtomicInteger pushOk, - AtomicInteger pushFail, List failLines, Object failLock) { + /** 单店近窗拉成功后立刻推该下游（推送串行）；推失败不影响拉取成功计数。 */ + private void pushOneAfterNearPull(SmHotelEntity hotel, String partner, Semaphore pushGate, + AtomicInteger pushOk, AtomicInteger pushFail, List failLines, Object failLock) { if (hotel == null || hotel.getId() == null || StrUtil.isBlank(partner)) { return; } + boolean acquired = false; try { + if (pushGate != null) { + pushGate.acquire(); + acquired = true; + } R> push = midService.pushToDownstream(hotel.getId(), partner); if (push != null && push.getCode() == 0) { pushOk.incrementAndGet(); @@ -2285,6 +2321,13 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { + (push == null ? "null" : StrUtil.blankToDefault(push.getMsg(), "业务失败"))); } } + catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + pushFail.incrementAndGet(); + synchronized (failLock) { + appendFail(failLines, hotel, "近窗后推下游中断"); + } + } catch (Exception e) { pushFail.incrementAndGet(); synchronized (failLock) { @@ -2292,6 +2335,11 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { } log.warn("|近窗边拉边推|失败|hotelId={}|partner={}|err={}|", hotel.getId(), partner, e.getMessage()); } + finally { + if (acquired) { + pushGate.release(); + } + } } /** 全量详情与范围详情互斥（不含近窗）。 */