This commit is contained in:
zjf committed 2026-11-10 10:39:33 +08:00
1 parent 44c5e38a34
commit e7d1df5ef9
13 files changed
+580 -51

No files matched your search

@@ -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;
@@ -75,4 +75,12 @@ public class SmHotelDownstreamPushLogEntity extends Model<SmHotelDownstreamPushL
@TableLogic
@TableField(fill = FieldFill.INSERT)
private String delFlag;
/** 列表筛选:创建时间起(非表字段) */
@TableField(exist = false)
private String createTimeBegin;
/** 列表筛选:创建时间止(非表字段) */
@TableField(exist = false)
private String createTimeEnd;
}
@@ -128,6 +128,21 @@ public class HotelDownstreamProperties {
/** 推送流水保留天数,定时清理更早数据;≤0 不自动清理 */
private int pushLogRetainDays = 15;
/**
* 过期流水是否先归档再删。归档为本地 CSV.GZ,便于历史留存。
* 目录见 {@link #pushLogArchiveDir}。
*/
private boolean pushLogArchiveEnabled = true;
/**
* 归档目录(绝对或相对进程工作目录)。空则用 ./data/sm-hotel-push-log-archive。
* Nacos:hotel-downstream.push-log-archive-dir
*/
private String pushLogArchiveDir = "";
/** 归档文件中 request/response 最大字符,超出截断;≤0 不截断(文件会很大) */
private int pushLogArchiveBodyMaxChars = 4096;
/**
* 本地/回调变更后是否自动推下游(防抖)。
* <ul>
@@ -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。
@@ -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<Integer> 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<Map<String, Object>> 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<List<Map<String, Object>>> listArchives() {
return R.ok(pushLogService.listArchiveFiles());
}
@Operation(summary = "下载归档文件")
@GetMapping("/archives/download")
@PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_view')")
public ResponseEntity<Resource> 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));
}
}
@@ -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());
}
});
}
@@ -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<SmHotelDownstreamPushLogEntity> {
/** 归档用:按 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<SmHotelDownstreamPushLogEntity> 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<SmHotelDownstreamPushLogEntity> 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("""
<script>
DELETE FROM canteen_hotel_downstream_push_log WHERE id IN
<foreach collection="ids" item="id" open="(" separator="," close=")">#{id}</foreach>
</script>
""")
int physicalDeleteByIds(@Param("ids") List<Long> 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。
*/
@@ -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<SmHotelDownstreamPushLogEntity> {
@@ -16,15 +17,24 @@ public interface SmHotelDownstreamPushLogService extends IService<SmHotelDownstr
/** 异步落库推送流水(失败不影响主流程) */
void record(SmHotelDownstreamPushLogEntity logEntity);
/** 清理早于 retainDays 的流水(逻辑删除);返回影响行数 */
int cleanupOlderThanDays(int retainDays);
/**
* 清理早于 retainDays 的流水:默认先归档 CSV.GZ 再物理删除。
* 返回 archivedRows / deletedRows / files 等。
*/
R<Map<String, Object>> cleanupOlderThanDays(int retainDays);
/** 删除单条推送流水 */
/** 删除单条推送流水(物理删) */
R<Boolean> removeLog(Long id);
/** 按创建时间区间删除(至少指定一端) */
/** 按创建时间区间分批处理(默认先归档再物理删);超大批量可能需点多次 */
R<Integer> removeByCreateTimeRange(String createTimeBegin, String createTimeEnd);
/** 归档文件列表(文件名/大小/修改时间) */
List<Map<String, Object>> listArchiveFiles();
/** 解析可下载的归档文件(防路径穿越);不存在返回 null */
java.io.File resolveArchiveFile(String fileName);
/**
* 按某次下游全量推同步日志导出完善清单(不限行数):
* ① 圣诺 getInfo 无店/失败;② getInfo 成功但房型/产品编码匹配不上。
@@ -24,6 +24,11 @@ public interface SmHotelMidService {
*/
R<Map<String, Object>> pullFromUpstream(Long hotelId, Integer priceDaysOverride);
/**
* @param scheduleAutoPush false 时不触发下游防抖自动推(近窗边拉边推已自行推送)
*/
R<Map<String, Object>> pullFromUpstream(Long hotelId, Integer priceDaysOverride, boolean scheduleAutoPush);
/**
* 仅调用 getHotelComplexList 落房型(不拉房价),用于单独验证综合信息接口。
*/
@@ -526,6 +526,12 @@ public class SmHotelChannelConfigServiceImpl extends ServiceImpl<SmHotelChannelC
if (StrUtil.isNotBlank(dto.getNearDetailPullCron())) {
target.setNearDetailPullCron(dto.getNearDetailPullCron().trim());
}
if (dto.getNearDetailPullConcurrency() != null && dto.getNearDetailPullConcurrency() > 0) {
target.setNearDetailPullConcurrency(dto.getNearDetailPullConcurrency());
}
if (dto.getNearDetailPullIntervalMs() != null && dto.getNearDetailPullIntervalMs() >= 0) {
target.setNearDetailPullIntervalMs(dto.getNearDetailPullIntervalMs());
}
if (dto.getAutoImportNewHotels() != null) {
target.setAutoImportNewHotels(dto.getAutoImportNewHotels());
}
@@ -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<SmHotelDownstreamPushLogEntity> pageLogs(Page<SmHotelDownstreamPushLogEntity> 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.<SmHotelDownstreamPushLogEntity>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<Map<String, Object>> 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.<SmHotelDownstreamPushLogEntity>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.<SmHotelDownstreamPushLogEntity>lambdaQuery();
if (begin != null) {
qw.ge(SmHotelDownstreamPushLogEntity::getCreateTime, begin);
R<Map<String, Object>> 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<Map<String, Object>> 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<Map<String, Object>> out = new ArrayList<>(files.length);
for (File f : files) {
Map<String, Object> 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 &lt; before 的数据后物理删除。
*/
private R<Map<String, Object>> archiveThenDeleteBefore(LocalDateTime before, String tag) {
boolean archiveOn = downstreamProperties == null || downstreamProperties.isPushLogArchiveEnabled();
int archived = 0;
int deleted = 0;
List<String> 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<SmHotelDownstreamPushLogEntity> batch = baseMapper.listOldestBefore(before, PHYSICAL_DELETE_BATCH);
if (CollUtil.isEmpty(batch)) {
break;
}
List<Long> 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<String, Object> 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<Map<String, Object>> archiveThenDeleteBetween(LocalDateTime begin, LocalDateTime end, String tag) {
boolean archiveOn = downstreamProperties == null || downstreamProperties.isPushLogArchiveEnabled();
int archived = 0;
int deleted = 0;
List<String> 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<SmHotelDownstreamPushLogEntity> batch = baseMapper.listOldestBetween(begin, end, PHYSICAL_DELETE_BATCH);
if (CollUtil.isEmpty(batch)) {
break;
}
List<Long> 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<String, Object> 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<SmHotelDownstreamPushLogEntity> 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
@@ -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<Map<String, Object>> pullFromUpstream(Long hotelId, Integer priceDaysOverride) {
return pullFromUpstream(hotelId, priceDaysOverride, true);
}
/**
* @param scheduleAutoPush false:近窗边拉边推等场景已自行推送,避免再防抖自动推加重库压
*/
@Override
public R<Map<String, Object>> pullFromUpstream(Long hotelId, Integer priceDaysOverride,
boolean scheduleAutoPush) {
SmHotelEntity hotel = requireChannelWritableHotel(hotelId, SmHotelDataSource.SHANGYOU);
String hotelCode = hotel.getHotelCode();
List<String> 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);
}
@@ -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<Map<String, Object>> 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<String> failLines, Object failLock) {
/** 单店近窗拉成功后立刻推该下游(推送串行);推失败不影响拉取成功计数。 */
private void pushOneAfterNearPull(SmHotelEntity hotel, String partner, Semaphore pushGate,
AtomicInteger pushOk, AtomicInteger pushFail, List<String> 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<Map<String, Object>> 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();
}
}
}
/** 全量详情与范围详情互斥(不含近窗)。 */