1.列表只出未删除、非回收的单品
2.选择组成菜品弹窗加了分类筛选(可按商户),并带上当前商户 3.酒店增加删除单条,按时间区间删除,清空redis队列
This commit is contained in:
1 parent
be723cf076
commit
93379302e4
13 files changed
+291
-33
No files matched your search
+2
-1
@@ -643,8 +643,9 @@ public class CanteenOnlineSpuServiceImpl extends ServiceImpl<CanteenOnlineSpuMap
|
||||
.eq(Objects.nonNull(canteenOnlineSpu.getCategoryId()), CanteenOnlineSpuEntity::getCategoryId, canteenOnlineSpu.getCategoryId());
|
||||
merchantScopeHelper.applyMerchantIdScope(wrapper, canteenOnlineSpu.getMerchantId(), CanteenOnlineSpuEntity::getMerchantId);
|
||||
wrapper
|
||||
// 套餐组成只可选单品(spu_type 为空或 1)
|
||||
// 套餐组成只可选单品(spu_type 为空或 1);排除逻辑删除与回收,仅上架
|
||||
.and(w -> w.isNull(CanteenOnlineSpuEntity::getSpuType).or().eq(CanteenOnlineSpuEntity::getSpuType, 1))
|
||||
.and(w -> w.eq(CanteenOnlineSpuEntity::getDelFlag, "0").or().isNull(CanteenOnlineSpuEntity::getDelFlag))
|
||||
.eq(CanteenOnlineSpuEntity::getStatus, 0)
|
||||
.orderByAsc(CanteenOnlineSpuEntity::getSort).orderByDesc(CanteenOnlineSpuEntity::getId);
|
||||
Page<CanteenOnlineSpuEntity> pageData = baseMapper.selectPageByScope(page, wrapper, DataScope.of());
|
||||
|
||||
+3
-1
@@ -569,8 +569,10 @@ public class CanteenSpuServiceImpl extends ServiceImpl<CanteenSpuMapper, Canteen
|
||||
LambdaQueryWrapper<CanteenSpuEntity> wrapper = new LambdaQueryWrapper<>();
|
||||
wrapper.like(StrUtil.isNotBlank(canteenSpu.getName()), CanteenSpuEntity::getName, canteenSpu.getName());
|
||||
wrapper.eq(Objects.nonNull(canteenSpu.getCategoryId()), CanteenSpuEntity::getCategoryId, canteenSpu.getCategoryId());
|
||||
// 套餐组成只可选单品(spu_type 为空或 1)
|
||||
// 套餐组成只可选单品(spu_type 为空或 1);排除逻辑删除与回收
|
||||
wrapper.and(w -> w.isNull(CanteenSpuEntity::getSpuType).or().eq(CanteenSpuEntity::getSpuType, 1));
|
||||
wrapper.and(w -> w.eq(CanteenSpuEntity::getDelFlag, "0").or().isNull(CanteenSpuEntity::getDelFlag));
|
||||
wrapper.ne(CanteenSpuEntity::getStatus, -1);
|
||||
merchantScopeHelper.applyMerchantIdScope(wrapper, canteenSpu.getMerchantId(), CanteenSpuEntity::getMerchantId);
|
||||
wrapper.orderByAsc(CanteenSpuEntity::getSort).orderByDesc(CanteenSpuEntity::getId);
|
||||
|
||||
|
||||
+1
-1
@@ -21,7 +21,7 @@ public class SmHotelCallbackTaskEntity extends Model<SmHotelCallbackTaskEntity>
|
||||
|
||||
private Long tenantId;
|
||||
|
||||
@Schema(description = "PRICE / INVENTORY")
|
||||
@Schema(description = "PRICE / INVENTORY / ORDER_STATUS(ORDER_STATUS 高优先级)")
|
||||
private String notifyType;
|
||||
|
||||
@Schema(description = "上游 hotelId")
|
||||
|
||||
+30
-3
@@ -15,7 +15,7 @@ import org.springframework.security.access.prepost.PreAuthorize;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
|
||||
/**
|
||||
* 上游房价/房量回调任务(后台查询与重试)。
|
||||
* 上游房价/房量回调任务(后台查询、重试、清理)。
|
||||
*/
|
||||
@RestController
|
||||
@RequiredArgsConstructor
|
||||
@@ -30,8 +30,10 @@ public class SmHotelCallbackTaskController {
|
||||
@GetMapping("/page")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_view')")
|
||||
public R<Page<SmHotelCallbackTaskEntity>> page(@ParameterObject Page<SmHotelCallbackTaskEntity> page,
|
||||
@ParameterObject SmHotelCallbackTaskEntity query) {
|
||||
return R.ok(callbackTaskService.pageTasks(page, query));
|
||||
@ParameterObject SmHotelCallbackTaskEntity query,
|
||||
@RequestParam(required = false) String createTimeBegin,
|
||||
@RequestParam(required = false) String createTimeEnd) {
|
||||
return R.ok(callbackTaskService.pageTasks(page, query, createTimeBegin, createTimeEnd));
|
||||
}
|
||||
|
||||
@Operation(summary = "回调任务详情")
|
||||
@@ -48,4 +50,29 @@ public class SmHotelCallbackTaskController {
|
||||
public R<Boolean> retry(@PathVariable Long id) {
|
||||
return callbackTaskService.retry(id);
|
||||
}
|
||||
|
||||
@SysLog("删除酒店回调任务")
|
||||
@Operation(summary = "删除单条回调任务")
|
||||
@DeleteMapping("/{id}")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_del')")
|
||||
public R<Boolean> remove(@PathVariable Long id) {
|
||||
return callbackTaskService.removeTask(id);
|
||||
}
|
||||
|
||||
@SysLog("按时间区间删除酒店回调任务")
|
||||
@Operation(summary = "按创建时间区间删除回调任务")
|
||||
@DeleteMapping("/range")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_del')")
|
||||
public R<Integer> removeRange(@RequestParam(required = false) String createTimeBegin,
|
||||
@RequestParam(required = false) String createTimeEnd) {
|
||||
return callbackTaskService.removeByCreateTimeRange(createTimeBegin, createTimeEnd);
|
||||
}
|
||||
|
||||
@SysLog("清空酒店回调任务")
|
||||
@Operation(summary = "清空全部回调任务并清理 Redis 队列")
|
||||
@DeleteMapping("/clear")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_del')")
|
||||
public R<Integer> clearAll() {
|
||||
return callbackTaskService.clearAllTasks();
|
||||
}
|
||||
}
|
||||
+23
@@ -96,6 +96,13 @@ public class SmHotelMidController {
|
||||
return midService.pullPriceFromUpstream(id);
|
||||
}
|
||||
|
||||
@Operation(summary = "仅拉取上游产品样例(getRoomPrice 今天 1 天落产品)")
|
||||
@PostMapping("/mid/pullProducts/{id}")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_edit')")
|
||||
public R<Map<String, Object>> pullProducts(@PathVariable Long id) {
|
||||
return midService.pullProductsFromUpstream(id);
|
||||
}
|
||||
|
||||
@Operation(summary = "推送到下游销售方")
|
||||
@PostMapping("/mid/push/{id}")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_push')")
|
||||
@@ -159,6 +166,22 @@ public class SmHotelMidController {
|
||||
return syncService.getSyncLog(id);
|
||||
}
|
||||
|
||||
@SysLog("终止酒店同步死任务")
|
||||
@Operation(summary = "终止仍显示进行中的同步日志(死任务)")
|
||||
@PostMapping("/mid/syncLog/{id}/abort")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_edit')")
|
||||
public R<Boolean> abortSyncLog(@PathVariable Long id) {
|
||||
return syncService.abortSyncLog(id);
|
||||
}
|
||||
|
||||
@SysLog("清理酒店同步卡死任务")
|
||||
@Operation(summary = "清理长时间无进度的 RUNNING 同步日志")
|
||||
@PostMapping("/mid/syncLog/abortStale")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_edit')")
|
||||
public R<Integer> abortStaleSyncLogs() {
|
||||
return R.ok(syncService.abortStaleRunningSyncLogs());
|
||||
}
|
||||
|
||||
@Operation(summary = "日更时间配置(只读展示)")
|
||||
@GetMapping("/mid/syncSchedule")
|
||||
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_view')")
|
||||
|
||||
+15
-6
@@ -10,11 +10,12 @@ import org.springframework.boot.ApplicationRunner;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* 消费 Redis 回调队列:BRPOP → 异步拉取。
|
||||
* 消费 Redis 回调队列:BRPOP 高优队列优先于普通队列。
|
||||
* <p>
|
||||
* 另起看门狗线程:即使 Worker 卡在某次上游调用里,也能把超时 RUNNING 重置入队,避免整队假死。
|
||||
*/
|
||||
@@ -55,16 +56,24 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
|
||||
watchdog.setDaemon(true);
|
||||
watchdog.start();
|
||||
|
||||
log.info("|回调队列|Worker+Watchdog 已启动|key={}|", ShangyouHotelApi.REDIS_CALLBACK_QUEUE);
|
||||
log.info("|回调队列|Worker+Watchdog 已启动|high={}|normal={}|",
|
||||
ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, ShangyouHotelApi.REDIS_CALLBACK_QUEUE);
|
||||
}
|
||||
|
||||
private void loop() {
|
||||
int consecutiveFail = 0;
|
||||
while (running.get()) {
|
||||
try {
|
||||
String taskId = redisTemplate.opsForList()
|
||||
.rightPop(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, 5, TimeUnit.SECONDS);
|
||||
if (taskId == null || taskId.isBlank()) {
|
||||
// BRPOP 多 key:先高优(订单状态)再普通;阻塞期间新到的高优也会被优先弹出
|
||||
List<String> popped = redisTemplate.opsForList().rightPop(
|
||||
List.of(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH,
|
||||
ShangyouHotelApi.REDIS_CALLBACK_QUEUE),
|
||||
Duration.ofSeconds(5));
|
||||
if (popped == null || popped.size() < 2 || popped.get(1) == null) {
|
||||
continue;
|
||||
}
|
||||
String taskId = popped.get(1).trim();
|
||||
if (taskId.isBlank() || "null".equalsIgnoreCase(taskId)) {
|
||||
continue;
|
||||
}
|
||||
Long id;
|
||||
|
||||
+4
-1
@@ -41,9 +41,12 @@ public final class ShangyouHotelApi {
|
||||
/** 酒店列表同步锁 */
|
||||
public static final String REDIS_HOTEL_LIST_LOCK = "supermarket:shangyou:hotel:list:lock";
|
||||
|
||||
/** 房价/房量回调异步队列(LIST,元素=taskId) */
|
||||
/** 回调异步队列(LIST,元素=taskId):房价/房量等普通优先级 */
|
||||
public static final String REDIS_CALLBACK_QUEUE = "supermarket:shangyou:hotel:callback:queue";
|
||||
|
||||
/** 回调高优先级队列:订单状态等,Worker 始终先消费本队列 */
|
||||
public static final String REDIS_CALLBACK_QUEUE_HIGH = "supermarket:shangyou:hotel:callback:queue:high";
|
||||
|
||||
/** 上游授权 hotelId 集合(与 HASH/ZSET 同步重建,供 SDIFF) */
|
||||
public static final String REDIS_UPSTREAM_ID_SET = "supermarket:shangyou:hotel:idset:v4";
|
||||
|
||||
|
||||
+14
-2
@@ -17,15 +17,27 @@ public interface SmHotelCallbackTaskService {
|
||||
*/
|
||||
Map<String, Object> enqueueFromNotify(String notifyType, Map<String, Object> body, String time, String sign);
|
||||
|
||||
/** 后台分页 */
|
||||
/** 后台分页(可按回调时间 createTime 区间筛选) */
|
||||
Page<SmHotelCallbackTaskEntity> pageTasks(Page<SmHotelCallbackTaskEntity> page,
|
||||
SmHotelCallbackTaskEntity query);
|
||||
SmHotelCallbackTaskEntity query, String createTimeBegin, String createTimeEnd);
|
||||
|
||||
SmHotelCallbackTaskEntity getDetail(Long id);
|
||||
|
||||
/** 失败/跳过任务重新入队 */
|
||||
R<Boolean> retry(Long id);
|
||||
|
||||
/** 删除单条回调任务 */
|
||||
R<Boolean> removeTask(Long id);
|
||||
|
||||
/**
|
||||
* 按创建时间区间删除(含边界)。begin/end 都空则拒绝。
|
||||
* @return 删除条数
|
||||
*/
|
||||
R<Integer> removeByCreateTimeRange(String createTimeBegin, String createTimeEnd);
|
||||
|
||||
/** 清空全部回调任务,并清空 Redis 消费队列 */
|
||||
R<Integer> clearAllTasks();
|
||||
|
||||
/** 消费一条队列任务(Worker 调用);返回 true=本次成功/跳过,false=失败或待再试 */
|
||||
boolean processQueuedTask(Long taskId);
|
||||
|
||||
|
||||
+5
@@ -26,6 +26,11 @@ public interface SmHotelMidService {
|
||||
*/
|
||||
R<Map<String, Object>> pullPriceFromUpstream(Long hotelId);
|
||||
|
||||
/**
|
||||
* 仅拉取产品样例(默认今天 1 天 getRoomPrice 落产品),用于单独验证产品解析落库。
|
||||
*/
|
||||
R<Map<String, Object>> pullProductsFromUpstream(Long hotelId);
|
||||
|
||||
/**
|
||||
* 仅拉取指定日期区间房价(含房型综合若本地无房型),用于「滚一天」;区间建议 1 天,最长按上游 31 天切片。
|
||||
*/
|
||||
|
||||
+9
@@ -44,6 +44,15 @@ public interface SmHotelSyncService {
|
||||
*/
|
||||
R<SmHotelSyncLogEntity> startRetryFailedAsync(Long sourceLogId);
|
||||
|
||||
/** 人工终止一条仍显示进行中的同步日志(死任务) */
|
||||
R<Boolean> abortSyncLog(Long id);
|
||||
|
||||
/**
|
||||
* 将长时间无进度更新的 RUNNING 日志标为失败,避免挡后续任务。
|
||||
* @return 清理条数
|
||||
*/
|
||||
int abortStaleRunningSyncLogs();
|
||||
|
||||
/** 兼容旧同步入口(内部转异步) */
|
||||
R<Map<String, Object>> syncHotelListAndDiff(boolean autoImport);
|
||||
|
||||
|
||||
+112
-14
@@ -79,11 +79,12 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
SmHotelCallbackTaskEntity task = buildTask(notifyType, body, time, tid);
|
||||
taskMapper.insert(task);
|
||||
if (properties.isCallbackQueueEnabled()) {
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE,
|
||||
String.valueOf(task.getId()));
|
||||
enqueueTaskId(task.getId(), task.getNotifyType());
|
||||
}
|
||||
log.info("|上游酒店通知|已入队|taskId={}|hotelCode={}|hotelId={}|status={}|", task.getId(),
|
||||
task.getHotelCode(), task.getHotelId(), task.getStatus());
|
||||
log.info("|上游酒店通知|已入队|taskId={}|type={}|priority={}|hotelCode={}|hotelId={}|status={}|",
|
||||
task.getId(), task.getNotifyType(),
|
||||
isHighPriority(task.getNotifyType()) ? "HIGH" : "NORMAL", task.getHotelCode(),
|
||||
task.getHotelId(), task.getStatus());
|
||||
});
|
||||
}
|
||||
catch (Exception e) {
|
||||
@@ -96,7 +97,9 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
|
||||
@Override
|
||||
public Page<SmHotelCallbackTaskEntity> pageTasks(Page<SmHotelCallbackTaskEntity> page,
|
||||
SmHotelCallbackTaskEntity query) {
|
||||
SmHotelCallbackTaskEntity query, String createTimeBegin, String createTimeEnd) {
|
||||
LocalDateTime begin = parseDateTime(createTimeBegin, true);
|
||||
LocalDateTime end = parseDateTime(createTimeEnd, false);
|
||||
return taskMapper.selectPage(page, Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
|
||||
.eq(StrUtil.isNotBlank(query.getNotifyType()), SmHotelCallbackTaskEntity::getNotifyType,
|
||||
query.getNotifyType())
|
||||
@@ -105,6 +108,8 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
query.getHotelCode())
|
||||
.like(StrUtil.isNotBlank(query.getHotelName()), SmHotelCallbackTaskEntity::getHotelName,
|
||||
query.getHotelName())
|
||||
.ge(begin != null, SmHotelCallbackTaskEntity::getCreateTime, begin)
|
||||
.le(end != null, SmHotelCallbackTaskEntity::getCreateTime, end)
|
||||
.orderByDesc(SmHotelCallbackTaskEntity::getCreateTime));
|
||||
}
|
||||
|
||||
@@ -137,8 +142,87 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
task.setCostMs(null);
|
||||
task.setUpdateTime(LocalDateTime.now());
|
||||
taskMapper.updateById(task);
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(task.getId()));
|
||||
return R.ok(Boolean.TRUE, "已重新入队");
|
||||
enqueueTaskId(task.getId(), task.getNotifyType());
|
||||
return R.ok(Boolean.TRUE, isHighPriority(task.getNotifyType()) ? "已重新入队(高优先级)" : "已重新入队");
|
||||
}
|
||||
|
||||
@Override
|
||||
public R<Boolean> removeTask(Long id) {
|
||||
if (id == null) {
|
||||
return R.failed("任务ID不能为空");
|
||||
}
|
||||
SmHotelCallbackTaskEntity task = taskMapper.selectById(id);
|
||||
if (task == null) {
|
||||
return R.failed("任务不存在");
|
||||
}
|
||||
if (STATUS_RUNNING.equals(task.getStatus())) {
|
||||
return R.failed("任务处理中,请稍后再删");
|
||||
}
|
||||
taskMapper.deleteById(id);
|
||||
return R.ok(Boolean.TRUE, "已删除");
|
||||
}
|
||||
|
||||
@Override
|
||||
public R<Integer> removeByCreateTimeRange(String createTimeBegin, String createTimeEnd) {
|
||||
LocalDateTime begin = parseDateTime(createTimeBegin, true);
|
||||
LocalDateTime end = parseDateTime(createTimeEnd, false);
|
||||
if (begin == null && end == null) {
|
||||
return R.failed("请指定开始或结束时间");
|
||||
}
|
||||
if (begin != null && end != null && end.isBefore(begin)) {
|
||||
return R.failed("结束时间不能早于开始时间");
|
||||
}
|
||||
// 不删 RUNNING,避免删掉正在消费的任务导致状态错乱
|
||||
var qw = Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
|
||||
.ne(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING);
|
||||
if (begin != null) {
|
||||
qw.ge(SmHotelCallbackTaskEntity::getCreateTime, begin);
|
||||
}
|
||||
if (end != null) {
|
||||
qw.le(SmHotelCallbackTaskEntity::getCreateTime, end);
|
||||
}
|
||||
Long cnt = taskMapper.selectCount(qw);
|
||||
int n = taskMapper.delete(qw);
|
||||
log.info("|回调任务|按创建时间删除|begin={}|end={}|deleted={}|matched={}|", begin, end, n, cnt);
|
||||
return R.ok(n, "已删除 " + n + " 条");
|
||||
}
|
||||
|
||||
@Override
|
||||
public R<Integer> clearAllTasks() {
|
||||
Long cnt = taskMapper.selectCount(Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
|
||||
.ne(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING));
|
||||
int n = taskMapper.delete(Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
|
||||
.ne(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING));
|
||||
// 清空消费队列(含高优),避免旧 taskId 反复弹出
|
||||
Long qLen = redisTemplate.opsForList().size(ShangyouHotelApi.REDIS_CALLBACK_QUEUE);
|
||||
Long qHigh = redisTemplate.opsForList().size(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH);
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUE);
|
||||
redisTemplate.delete(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH);
|
||||
log.warn("|回调任务|清空全部|deleted={}|queueCleared≈{}/high≈{}|", n, qLen, qHigh);
|
||||
return R.ok(n, "已清空 " + n + " 条任务,并清理队列"
|
||||
+ (cnt != null && cnt > n ? "(处理中任务未删)" : ""));
|
||||
}
|
||||
|
||||
/** 支持 yyyy-MM-dd 或 yyyy-MM-dd HH:mm:ss;日期无时分秒时 begin=00:00:00、end=23:59:59 */
|
||||
private static LocalDateTime parseDateTime(String raw, boolean beginOfDay) {
|
||||
if (StrUtil.isBlank(raw)) {
|
||||
return null;
|
||||
}
|
||||
String s = raw.trim().replace('T', ' ');
|
||||
try {
|
||||
if (s.length() <= 10) {
|
||||
LocalDate d = LocalDate.parse(s.substring(0, 10), DateTimeFormatter.ISO_LOCAL_DATE);
|
||||
return beginOfDay ? d.atStartOfDay() : d.atTime(23, 59, 59);
|
||||
}
|
||||
if (s.length() == 16) {
|
||||
return LocalDateTime.parse(s, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm"));
|
||||
}
|
||||
return LocalDateTime.parse(s.substring(0, Math.min(19, s.length())),
|
||||
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
|
||||
}
|
||||
catch (Exception e) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -156,14 +240,13 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
|
||||
return TenantBroker.applyAs(tenantId, tid -> {
|
||||
int n = 0;
|
||||
// 1) PENDING
|
||||
// 1) PENDING:按创建时间 FIFO;ORDER_STATUS 进高优队列
|
||||
var pending = taskMapper.selectList(Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
|
||||
.eq(SmHotelCallbackTaskEntity::getStatus, STATUS_PENDING)
|
||||
.orderByAsc(SmHotelCallbackTaskEntity::getCreateTime)
|
||||
.last("limit 200"));
|
||||
// 按创建时间从早到晚 leftPush:最终右侧是最早的,rightPop 先消费最早任务
|
||||
for (SmHotelCallbackTaskEntity t : pending) {
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId()));
|
||||
enqueueTaskId(t.getId(), t.getNotifyType());
|
||||
n++;
|
||||
}
|
||||
// 2) 卡住的 RUNNING
|
||||
@@ -209,7 +292,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
t.setStatus(STATUS_PENDING);
|
||||
t.setResultMsg("超时 RUNNING 已重置为 PENDING 并重新入队");
|
||||
taskMapper.updateById(t);
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(t.getId()));
|
||||
enqueueTaskId(t.getId(), t.getNotifyType());
|
||||
n++;
|
||||
}
|
||||
if (n > 0) {
|
||||
@@ -342,9 +425,9 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
task.setCostMs(System.currentTimeMillis() - t0);
|
||||
task.setUpdateTime(LocalDateTime.now());
|
||||
taskMapper.updateById(task);
|
||||
// 队列约定:入队 leftPush、消费 rightPop(FIFO)。
|
||||
// 队列约定:入队 leftPush、消费 rightPop(FIFO);ORDER_STATUS 进高优队列仍 FIFO。
|
||||
// 失败重试必须 leftPush 到队尾;rightPush 会插到 rightPop 端,等于立刻再跑同一条,饿死后面的新任务。
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, String.valueOf(task.getId()));
|
||||
enqueueTaskId(task.getId(), task.getNotifyType());
|
||||
return;
|
||||
}
|
||||
finish(task, STATUS_FAILED, msg, t0);
|
||||
@@ -387,7 +470,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
task.setReqJson(body == null ? null : JSONUtil.toJsonStr(body));
|
||||
task.setStatus(STATUS_PENDING);
|
||||
task.setAttemptCount(0);
|
||||
task.setResultMsg("ORDER_STATUS".equalsIgnoreCase(notifyType) ? "已入队,等待异步更新订单状态"
|
||||
task.setResultMsg(isHighPriority(notifyType) ? "已入队(高优先级),等待异步更新订单状态"
|
||||
: "已入队,等待异步拉取");
|
||||
task.setCreateBy("shangyou-callback");
|
||||
task.setCreateTime(LocalDateTime.now());
|
||||
@@ -396,6 +479,21 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
|
||||
return task;
|
||||
}
|
||||
|
||||
/** 订单状态回调最高优先:进高优队列,Worker 始终先于房价/房量消费 */
|
||||
private void enqueueTaskId(Long taskId, String notifyType) {
|
||||
String id = String.valueOf(taskId);
|
||||
if (isHighPriority(notifyType)) {
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, id);
|
||||
}
|
||||
else {
|
||||
redisTemplate.opsForList().leftPush(ShangyouHotelApi.REDIS_CALLBACK_QUEUE, id);
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isHighPriority(String notifyType) {
|
||||
return "ORDER_STATUS".equalsIgnoreCase(notifyType);
|
||||
}
|
||||
|
||||
private static String str(Map<String, Object> body, String key) {
|
||||
if (body == null || body.get(key) == null) {
|
||||
return null;
|
||||
|
||||
+17
@@ -221,6 +221,23 @@ public class SmHotelMidServiceImpl implements SmHotelMidService {
|
||||
return R.ok(data);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public R<Map<String, Object>> pullProductsFromUpstream(Long hotelId) {
|
||||
// 产品来自 getRoomPrice 的 roomRateList;与「拉取房价」同接口,侧重看产品落库数
|
||||
R<Map<String, Object>> r = pullPriceFromUpstream(hotelId);
|
||||
if (r == null) {
|
||||
return R.failed("拉取产品返回空");
|
||||
}
|
||||
if (r.getCode() != 0) {
|
||||
return r;
|
||||
}
|
||||
Map<String, Object> data = r.getData() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(r.getData());
|
||||
log.info("|酒店拉取产品|结束|hotelCode={}|product={}|daily={}|", data.get("hotelCode"), data.get("productCount"),
|
||||
data.get("dailyCount"));
|
||||
return R.ok(data);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public R<Map<String, Object>> pullPriceRange(Long hotelId, LocalDate startInclusive, LocalDate endExclusive) {
|
||||
|
||||
+56
-4
@@ -54,6 +54,9 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
|
||||
private static final String STATUS_FAILED = "FAILED";
|
||||
|
||||
/** 同步日志超过该时间无 progress 心跳,视为死任务 */
|
||||
private static final int STALE_RUNNING_MINUTES = 30;
|
||||
|
||||
private static final ExecutorService EXECUTOR = Executors.newFixedThreadPool(2, r -> {
|
||||
Thread t = new Thread(r, "sm-hotel-sync");
|
||||
t.setDaemon(true);
|
||||
@@ -74,11 +77,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
|
||||
@Override
|
||||
public R<SmHotelSyncLogEntity> startListSyncAsync(boolean autoImport, String triggerType) {
|
||||
abortStaleRunningSyncLogs();
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getSyncType, TYPE_LIST)
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
return R.failed("已有列表同步任务进行中,请稍候查看进度");
|
||||
return R.failed("已有列表同步任务进行中,请稍候查看进度(若长时间卡住请在同步日志点「终止」)");
|
||||
}
|
||||
SmHotelSyncLogEntity logEntity = newLog(TYPE_LIST, triggerType);
|
||||
logEntity.setStage("QUEUED");
|
||||
@@ -102,11 +106,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
if (!properties.isPriceRollDailyEnabled() && !"MANUAL".equalsIgnoreCase(triggerType)) {
|
||||
return R.failed("滚价未启用");
|
||||
}
|
||||
abortStaleRunningSyncLogs();
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getSyncType, TYPE_ROLL)
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
return R.failed("已有滚价任务进行中");
|
||||
return R.failed("已有滚价任务进行中(若长时间卡住请在同步日志点「终止」)");
|
||||
}
|
||||
SmHotelSyncLogEntity logEntity = newLog(TYPE_ROLL, triggerType);
|
||||
logEntity.setStage("QUEUED");
|
||||
@@ -125,11 +130,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
|
||||
@Override
|
||||
public R<SmHotelSyncLogEntity> startDetailPullAsync(Integer limit, String triggerType) {
|
||||
abortStaleRunningSyncLogs();
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getSyncType, TYPE_DETAIL)
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
return R.failed("已有全量详情拉取任务进行中,请打开「同步日志」查看进度");
|
||||
return R.failed("已有全量详情拉取任务进行中,请打开「同步日志」查看进度(若长时间卡住请点「终止」)");
|
||||
}
|
||||
String quotaErr = checkFullPriceSyncQuota();
|
||||
if (quotaErr != null) {
|
||||
@@ -176,11 +182,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
if (CollUtil.isEmpty(codes)) {
|
||||
return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因)");
|
||||
}
|
||||
abortStaleRunningSyncLogs();
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getSyncType, syncType)
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
return R.failed("已有同类型任务进行中,请稍后再试");
|
||||
return R.failed("已有同类型任务进行中,请稍后再试(若卡住请先终止)");
|
||||
}
|
||||
|
||||
SmHotelSyncLogEntity logEntity = newLog(syncType, "RETRY");
|
||||
@@ -212,6 +219,51 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
return R.ok(e);
|
||||
}
|
||||
|
||||
@Override
|
||||
public R<Boolean> abortSyncLog(Long id) {
|
||||
if (id == null) {
|
||||
return R.failed("日志ID不能为空");
|
||||
}
|
||||
SmHotelSyncLogEntity e = syncLogMapper.selectById(id);
|
||||
if (e == null) {
|
||||
return R.failed("同步记录不存在");
|
||||
}
|
||||
if (!STATUS_RUNNING.equals(e.getStatus())) {
|
||||
return R.failed("仅能终止「进行中」的任务,当前状态=" + e.getStatus());
|
||||
}
|
||||
failLog(id, "人工终止死任务(线程已中断或进程重启后残留)",
|
||||
e.getCreateTime() == null ? 0L
|
||||
: Math.max(0L, java.time.Duration.between(e.getCreateTime(), LocalDateTime.now()).toMillis()));
|
||||
log.warn("|酒店同步|人工终止|logId={}|type={}|stage={}|pct={}|", id, e.getSyncType(), e.getStage(),
|
||||
e.getProgressPct());
|
||||
return R.ok(Boolean.TRUE, "已终止并标记失败,可重新发起同步");
|
||||
}
|
||||
|
||||
@Override
|
||||
public int abortStaleRunningSyncLogs() {
|
||||
LocalDateTime staleBefore = LocalDateTime.now().minusMinutes(STALE_RUNNING_MINUTES);
|
||||
var list = syncLogMapper.selectList(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)
|
||||
.and(w -> w.lt(SmHotelSyncLogEntity::getUpdateTime, staleBefore)
|
||||
.or()
|
||||
.isNull(SmHotelSyncLogEntity::getUpdateTime)
|
||||
.lt(SmHotelSyncLogEntity::getCreateTime, staleBefore))
|
||||
.last("limit 50"));
|
||||
int n = 0;
|
||||
for (SmHotelSyncLogEntity e : list) {
|
||||
failLog(e.getId(),
|
||||
"超时无进度(>" + STALE_RUNNING_MINUTES + " 分钟),自动标记失败以免挡后续任务",
|
||||
e.getCreateTime() == null ? 0L
|
||||
: Math.max(0L,
|
||||
java.time.Duration.between(e.getCreateTime(), LocalDateTime.now()).toMillis()));
|
||||
n++;
|
||||
}
|
||||
if (n > 0) {
|
||||
log.warn("|酒店同步|自动清理卡死 RUNNING|count={}|staleMinutes={}|", n, STALE_RUNNING_MINUTES);
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Page<SmHotelSyncLogEntity> pageSyncLogs(Page<SmHotelSyncLogEntity> page,
|
||||
SmHotelSyncLogEntity query) {
|
||||
|
||||
Reference in new issue
Block a user