This commit is contained in:
zjf committed 2026-11-09 18:54:49 +08:00
1 parent fdd33d3e0a
commit ef4bc42ddb
4 files changed
+80 -4

No files matched your search

@@ -107,6 +107,9 @@ public class SmHotelDownstreamChannelConfigDTO {
@Schema(description = "本伙伴近窗天数;≤0 用上游 near-detail-pull-days")
private Integer nearDetailPullDays;
@Schema(description = "近窗是否边拉边推(单店拉成功即推本伙伴);默认 false")
private Boolean nearDetailPullThenPush;
@Schema(description = "本地改价/库存/房态后是否自动推本伙伴;大批量补数可临时 false,默认 true(未配则跟进程级)")
private Boolean resourceAutoPushEnabled;
}
@@ -164,6 +164,13 @@ public class HotelDownstreamProperties {
*/
private int nearDetailPullDays = 0;
/**
* 近窗详情是否边拉边推:单店拉上游成功后立刻推本伙伴(拉一家推一家)。
* 仅对「单伙伴近窗」任务生效;开放应用(拉模式)请保持 false。
* false(默认):只刷本地。
*/
private boolean nearDetailPullThenPush = false;
/** 进程级:是否跑资源日更时钟;改 cron 需重启 */
private boolean resourcePushScheduleEnabled = true;
@@ -403,6 +403,9 @@ public class SmHotelChannelConfigServiceImpl extends ServiceImpl<SmHotelChannelC
if (dto.getNearDetailPullDays() != null) {
target.setNearDetailPullDays(dto.getNearDetailPullDays());
}
if (dto.getNearDetailPullThenPush() != null) {
target.setNearDetailPullThenPush(dto.getNearDetailPullThenPush());
}
if (dto.getResourceAutoPushEnabled() != null) {
target.setResourceAutoPushEnabled(dto.getResourceAutoPushEnabled());
}
@@ -2063,6 +2063,8 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
AtomicInteger fail = new AtomicInteger(Math.max(0, baseFail));
AtomicInteger skip = new AtomicInteger(Math.max(0, baseSkip));
AtomicInteger processed = new AtomicInteger(Math.max(0, baseProcessed));
AtomicInteger pushOk = new AtomicInteger(0);
AtomicInteger pushFail = new AtomicInteger(0);
List<String> failLines = baseFailLines == null ? new ArrayList<>() : new ArrayList<>(baseFailLines);
final Object failLock = new Object();
ExecutorService pullPool = null;
@@ -2080,10 +2082,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
tenantId = props().getCallbackTenantId() == null ? 1L : props().getCallbackTenantId();
}
final Long tid = tenantId;
final String thenPushPartner = resolveNearThenPushPartner(logId);
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 : "")
+ "…",
ok.get(), fail.get(), skip.get(), total, failLines);
@@ -2126,6 +2130,10 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
priceDaysParam);
if (r != null && r.getCode() == 0) {
ok.incrementAndGet();
if (StrUtil.isNotBlank(thenPushPartner)) {
pushOneAfterNearPull(h, thenPushPartner, pushOk, pushFail,
failLines, failLock);
}
}
else {
fail.incrementAndGet();
@@ -2157,9 +2165,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
if (doneCnt % 20 == 0 || doneCnt >= total) {
int pct = total <= 0 ? 50
: Math.min(95, 5 + (int) (doneCnt * 90L / total));
String pushHint = StrUtil.isBlank(thenPushPartner) ? ""
: " 推成功 " + pushOk.get() + " 推失败 " + pushFail.get();
progressSnapshot(logId, pct, "PULL",
"拉取中 " + doneCnt + "/" + total + " 成功 " + ok.get()
+ " 失败 " + fail.get() + " 并发=" + concurrency,
+ " 失败 " + fail.get() + pushHint
+ " 并发=" + concurrency,
ok.get(), fail.get(), skip.get(), total, failLines);
}
}
@@ -2195,7 +2206,11 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
}
done.setStatus(STATUS_SUCCESS);
done.setStage("DONE");
done.setStageMsg("范围详情拉取完成(并发=" + concurrency + ",priceDays=" + effectivePriceDays + ")");
String stageMsg = "范围详情拉取完成(并发=" + concurrency + ",priceDays=" + effectivePriceDays + ")";
if (StrUtil.isNotBlank(thenPushPartner)) {
stageMsg += ";边拉边推→" + thenPushPartner + " 成功 " + pushOk.get() + " 失败 " + pushFail.get();
}
done.setStageMsg(stageMsg);
done.setProgressPct(100);
done.setSuccessCount(ok.get());
done.setFailCount(fail.get());
@@ -2207,8 +2222,9 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
done.setUpdateTime(LocalDateTime.now());
syncLogMapper.updateById(done);
clearResumeCkpt(logId);
log.info("|范围详情|完成|logId={}|ok={}|fail={}|skip={}|concurrency={}|priceDays={}|costMs={}|", logId,
ok.get(), fail.get(), skip.get(), concurrency, effectivePriceDays, done.getCostMs());
log.info("|范围详情|完成|logId={}|ok={}|fail={}|skip={}|pushOk={}|pushFail={}|partner={}|concurrency={}|priceDays={}|costMs={}|",
logId, ok.get(), fail.get(), skip.get(), pushOk.get(), pushFail.get(), thenPushPartner,
concurrency, effectivePriceDays, done.getCostMs());
}
catch (Throwable e) {
log.error("|范围详情|失败|logId={}|", logId, e);
@@ -2231,6 +2247,53 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
}
}
/** 近窗单伙伴且 YAML 开了 then-push 时,返回目标伙伴编码;否则 null。 */
private String resolveNearThenPushPartner(Long logId) {
if (logId == null) {
return null;
}
SmHotelSyncLogEntity logEntity = syncLogMapper.selectById(logId);
if (logEntity == null || !TYPE_DETAIL_NEAR.equals(logEntity.getSyncType())) {
return null;
}
String partner = SmHotelDataSource.normalize(logEntity.getDataSource()).toLowerCase();
if (StrUtil.isBlank(partner) || SCOPE_UNION_DS.equals(partner)) {
return null;
}
HotelDownstreamProperties ds = channelConfigService.resolveDownstreamRuntime(partner);
if (ds == null || !ds.isEnabled() || !ds.isNearDetailPullThenPush()) {
return null;
}
return partner;
}
/** 单店近窗拉成功后立刻推该下游;推失败不影响拉取成功计数,仅记 failDetail。 */
private void pushOneAfterNearPull(SmHotelEntity hotel, String partner, AtomicInteger pushOk,
AtomicInteger pushFail, List<String> failLines, Object failLock) {
if (hotel == null || hotel.getId() == null || StrUtil.isBlank(partner)) {
return;
}
try {
R<Map<String, Object>> push = midService.pushToDownstream(hotel.getId(), partner);
if (push != null && push.getCode() == 0) {
pushOk.incrementAndGet();
return;
}
pushFail.incrementAndGet();
synchronized (failLock) {
appendFail(failLines, hotel, "近窗后推下游失败:"
+ (push == null ? "null" : StrUtil.blankToDefault(push.getMsg(), "业务失败")));
}
}
catch (Exception e) {
pushFail.incrementAndGet();
synchronized (failLock) {
appendFail(failLines, hotel, "近窗后推下游异常:" + e.getMessage());
}
log.warn("|近窗边拉边推|失败|hotelId={}|partner={}|err={}|", hotel.getId(), partner, e.getMessage());
}
}
/** 全量详情与范围详情互斥(不含近窗)。 */
private boolean hasRunningHeavyDetailPull() {
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()