1.酒店 2.酒店消息推送
This commit is contained in:
1 parent
e658118d85
commit
cb4ef187f1
14 files changed
+449
-101
No files matched your search
+1
-1
@@ -64,7 +64,7 @@ public class SmHotelOrderEntity extends Model<SmHotelOrderEntity> {
|
||||
|
||||
private String guestJson;
|
||||
|
||||
@Schema(description = "CREATED/PENDING/UPSTREAM_OK/CHECKED_IN/CANCELLED/UPSTREAM_FAIL/PENDING_CONFIRM/NEED_MANUAL")
|
||||
@Schema(description = "CREATED/PENDING/UPSTREAM_OK/CHECKED_IN/CANCELLED/UPSTREAM_FAIL/BOOK_FAIL/PENDING_CONFIRM/NEED_MANUAL")
|
||||
private String orderStatus;
|
||||
|
||||
@Schema(description = "上游 bookStatus:P/R/E/X")
|
||||
|
||||
+63
-6
@@ -167,12 +167,15 @@ public class SmHotelHuidingOrderServiceImpl implements SmHotelHuidingOrderServic
|
||||
.eq(SmHotelOrderEntity::getChannelNo, channelOrderNo)
|
||||
.last("limit 1"));
|
||||
if (exist != null) {
|
||||
outer.put("msg", "订单已存在(幂等)");
|
||||
outer.put("data", buildCreateData(exist));
|
||||
return outer;
|
||||
if (!"BOOK_FAIL".equals(exist.getOrderStatus())) {
|
||||
outer.put("msg", "订单已存在(幂等)");
|
||||
outer.put("data", buildCreateData(exist));
|
||||
return outer;
|
||||
}
|
||||
}
|
||||
MappedOrder mapped = mapCreateFields(partnerCode, data);
|
||||
if (mapped.error != null) {
|
||||
saveBookFailOrder(exist, partnerCode, channelOrderNo, mapped, data, mapped.error);
|
||||
return fail200(mapped.error);
|
||||
}
|
||||
Map<String, Object> checkBody = new LinkedHashMap<>();
|
||||
@@ -191,13 +194,16 @@ public class SmHotelHuidingOrderServiceImpl implements SmHotelHuidingOrderServic
|
||||
&& Boolean.TRUE.equals(check.getData().get("bookable"));
|
||||
if (!bookable) {
|
||||
Object reasons = check == null || check.getData() == null ? "预检失败" : check.getData().get("reasons");
|
||||
return fail200("预检不可订:" + reasons);
|
||||
String failMsg = "预检不可订:" + reasons;
|
||||
saveBookFailOrder(exist, partnerCode, channelOrderNo, mapped, data, failMsg);
|
||||
return fail200(failMsg);
|
||||
}
|
||||
|
||||
SmHotelOrderEntity order = new SmHotelOrderEntity();
|
||||
SmHotelOrderEntity order = exist != null ? exist : new SmHotelOrderEntity();
|
||||
order.setPartnerCode(partnerCode);
|
||||
order.setChannelNo(channelOrderNo);
|
||||
order.setHotelId(mapped.hotel.getId());
|
||||
order.setDeptId(mapped.hotel.getDeptId());
|
||||
order.setHotelCode(mapped.hotelCode);
|
||||
order.setGroupCode(mapped.hotel.getGroupCode());
|
||||
order.setRoomTypeCode(mapped.roomTypeCode);
|
||||
@@ -215,9 +221,15 @@ public class SmHotelHuidingOrderServiceImpl implements SmHotelHuidingOrderServic
|
||||
order.setCompensateStatus("NONE");
|
||||
order.setInventoryStatus("NONE");
|
||||
order.setCompensateRound(0);
|
||||
order.setFailMsg(null);
|
||||
order.setReqJson(data.toString());
|
||||
order.setHuanqiPayload(data.toString());
|
||||
orderMapper.insert(order);
|
||||
if (order.getId() == null) {
|
||||
orderMapper.insert(order);
|
||||
}
|
||||
else {
|
||||
orderMapper.updateById(order);
|
||||
}
|
||||
|
||||
// 本地酒店:占库存后 PENDING_CONFIRM,返回待确认,人工确认后再 operationOrder=1
|
||||
// 上游酒店:同步订上游,成功当场确认
|
||||
@@ -480,11 +492,56 @@ public class SmHotelHuidingOrderServiceImpl implements SmHotelHuidingOrderServic
|
||||
}
|
||||
return StrUtil.isNotBlank(order.getPlatformNo())
|
||||
&& !"UPSTREAM_FAIL".equals(order.getOrderStatus())
|
||||
&& !"BOOK_FAIL".equals(order.getOrderStatus())
|
||||
&& !"NEED_MANUAL".equals(order.getOrderStatus())
|
||||
&& !"CANCELLED".equals(order.getOrderStatus())
|
||||
&& !"PENDING_CANCEL".equals(order.getOrderStatus());
|
||||
}
|
||||
|
||||
/** 入站预检/映射失败落 BOOK_FAIL,对慧订仍返回失败 */
|
||||
private void saveBookFailOrder(SmHotelOrderEntity exist, String partnerCode, String channelOrderNo,
|
||||
MappedOrder mapped, JSONObject data, String failMsg) {
|
||||
try {
|
||||
SmHotelOrderEntity order = exist != null ? exist : new SmHotelOrderEntity();
|
||||
order.setPartnerCode(partnerCode);
|
||||
order.setChannelNo(channelOrderNo);
|
||||
order.setOrderStatus("BOOK_FAIL");
|
||||
order.setCompensateStatus("NONE");
|
||||
order.setInventoryStatus("NONE");
|
||||
order.setCompensateRound(0);
|
||||
order.setMarkupPercent(BigDecimal.ZERO);
|
||||
order.setFailMsg(StrUtil.maxLength(failMsg, 500));
|
||||
order.setReqJson(data == null ? null : data.toString());
|
||||
order.setHuanqiPayload(data == null ? null : data.toString());
|
||||
if (mapped != null && mapped.hotel != null) {
|
||||
order.setHotelId(mapped.hotel.getId());
|
||||
order.setDeptId(mapped.hotel.getDeptId());
|
||||
order.setHotelCode(mapped.hotelCode);
|
||||
order.setGroupCode(mapped.hotel.getGroupCode());
|
||||
order.setRoomTypeCode(mapped.roomTypeCode);
|
||||
order.setProductCode(mapped.productCode);
|
||||
order.setPriceCode(mapped.product == null ? null : mapped.product.getPriceCode());
|
||||
order.setCheckInDate(mapped.checkIn);
|
||||
order.setCheckOutDate(mapped.checkOut);
|
||||
order.setCount(mapped.quantity <= 0 ? 1 : mapped.quantity);
|
||||
order.setContactName(mapped.contactName);
|
||||
order.setContactPhone(mapped.contactPhone);
|
||||
order.setGuestJson(mapped.guestJson);
|
||||
order.setSellAmount(mapped.orderAmount);
|
||||
}
|
||||
if (order.getId() == null) {
|
||||
orderMapper.insert(order);
|
||||
}
|
||||
else {
|
||||
orderMapper.updateById(order);
|
||||
}
|
||||
log.info("|慧订订单|预订失败已落单|channelNo={}|localId={}|msg={}|", channelOrderNo, order.getId(), failMsg);
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.warn("|慧订订单|预订失败落单异常|channelNo={}|{}|", channelOrderNo, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isPendingConfirm(SmHotelOrderEntity order) {
|
||||
return order != null && "PENDING_CONFIRM".equals(order.getOrderStatus());
|
||||
}
|
||||
|
||||
+53
@@ -21,6 +21,7 @@ import com.bsd.pjcloud.supermarket.mapper.SmHotelMgrPushConfigMapper;
|
||||
import com.bsd.pjcloud.supermarket.mapper.SmHotelMgrPushLogMapper;
|
||||
import com.bsd.pjcloud.supermarket.mapper.SmHotelOrderMapper;
|
||||
import com.bsd.pjcloud.supermarket.service.SmHotelMgrPushService;
|
||||
import com.bsd.pjcloud.thirdparty.api.feign.RemoteThirdPushDeliveryService;
|
||||
import com.bsd.pjcloud.thirdparty.api.feign.RemoterThirdAppMessageService;
|
||||
import com.bsd.pjcloud.thirdparty.dto.appmessage.HotelManagerPushEventDto;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
@@ -79,6 +80,9 @@ public class SmHotelMgrPushServiceImpl implements SmHotelMgrPushService {
|
||||
|
||||
private final RemoterThirdAppMessageService remoteAppMessageService;
|
||||
|
||||
/** 推送入箱后立刻踢一轮投递扫描,减少对 Quartz 的依赖 */
|
||||
private final RemoteThirdPushDeliveryService remotePushDeliveryService;
|
||||
|
||||
@Override
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public SmHotelMgrPushConfigEntity getOrCreateConfig() {
|
||||
@@ -181,11 +185,23 @@ public class SmHotelMgrPushServiceImpl implements SmHotelMgrPushService {
|
||||
if (orderId == null) {
|
||||
return PushResult.fail("订单为空");
|
||||
}
|
||||
// 同一 JVM 内串行化同订单自动推送,避免补偿/双路径并发各写一条 6/6
|
||||
synchronized (("HOTEL_MGR_PUSH_ORDER:" + orderId).intern()) {
|
||||
return doPushNewOrderLocked(orderId, force);
|
||||
}
|
||||
}
|
||||
|
||||
private PushResult doPushNewOrderLocked(Long orderId, boolean force) {
|
||||
try {
|
||||
SmHotelMgrPushConfigEntity cfg = getOrCreateConfig();
|
||||
if (!force && !ENABLED.equals(cfg.getNewOrderEnabled())) {
|
||||
return PushResult.skip("新订单推送已关闭");
|
||||
}
|
||||
// 自动推送:同一订单只真正推一次;补偿/补上游会反复触发 schedule,但手机端 eventId 幂等本就只会响一次
|
||||
if (!force && alreadyAutoPushedNewOrder(orderId)) {
|
||||
log.info("|酒店管理推送|新订单已推送过,跳过|orderId={}|", orderId);
|
||||
return PushResult.skip("该订单已自动推送过");
|
||||
}
|
||||
SmHotelOrderEntity order = orderMapper.selectById(orderId);
|
||||
if (order == null) {
|
||||
return PushResult.fail("订单不存在");
|
||||
@@ -315,7 +331,11 @@ public class SmHotelMgrPushServiceImpl implements SmHotelMgrPushService {
|
||||
requestJson);
|
||||
return force ? PushResult.fail(tip) : PushResult.skip(tip);
|
||||
}
|
||||
kickPushDeliveryScan();
|
||||
String msg = "已推送成功 " + ok + "/" + recipientList.size() + " 人" + detail;
|
||||
if (fail > 0 && !failHints.isEmpty()) {
|
||||
msg = msg + ";部分失败:" + String.join(";", failHints);
|
||||
}
|
||||
savePushLog(SmHotelMgrPushLogEntity.TYPE_NEW_ORDER, trigger, String.valueOf(orderId),
|
||||
order.getHotelId(), hotelName, logDepts, recipientList, ok, fail, title, content, msg,
|
||||
requestJson);
|
||||
@@ -426,7 +446,11 @@ public class SmHotelMgrPushServiceImpl implements SmHotelMgrPushService {
|
||||
logDepts, leaderList, 0, fail, title, content, tip, requestJson);
|
||||
return force ? PushResult.fail(tip) : PushResult.skip(tip);
|
||||
}
|
||||
kickPushDeliveryScan();
|
||||
String msg = "日历提醒成功 " + ok + "/" + leaders.size() + " 人(负责人=" + leaderNames + ")";
|
||||
if (fail > 0 && !failHints.isEmpty()) {
|
||||
msg = msg + ";部分失败:" + String.join(";", failHints);
|
||||
}
|
||||
savePushLog(SmHotelMgrPushLogEntity.TYPE_CALENDAR, trigger, day + "-" + hour, null, null,
|
||||
logDepts, leaderList, ok, fail, title, content, msg, requestJson);
|
||||
return PushResult.success(ok, leaders.size(), msg);
|
||||
@@ -514,6 +538,20 @@ public class SmHotelMgrPushServiceImpl implements SmHotelMgrPushService {
|
||||
return item;
|
||||
}
|
||||
|
||||
/** 是否已有该订单的自动新订单成功/部分成功记录(补偿重试不应再刷手机与推送日志) */
|
||||
private boolean alreadyAutoPushedNewOrder(Long orderId) {
|
||||
if (orderId == null) {
|
||||
return false;
|
||||
}
|
||||
Long cnt = pushLogMapper.selectCount(Wrappers.<SmHotelMgrPushLogEntity>lambdaQuery()
|
||||
.eq(SmHotelMgrPushLogEntity::getPushType, SmHotelMgrPushLogEntity.TYPE_NEW_ORDER)
|
||||
.eq(SmHotelMgrPushLogEntity::getTriggerType, SmHotelMgrPushLogEntity.TRIGGER_AUTO)
|
||||
.eq(SmHotelMgrPushLogEntity::getBizId, String.valueOf(orderId))
|
||||
.in(SmHotelMgrPushLogEntity::getStatus, SmHotelMgrPushLogEntity.STATUS_OK,
|
||||
SmHotelMgrPushLogEntity.STATUS_PARTIAL));
|
||||
return cnt != null && cnt > 0;
|
||||
}
|
||||
|
||||
private void savePushLog(String pushType, String triggerType, String bizId, Long hotelId, String hotelName,
|
||||
List<Long> deptIds, List<Long> userIds, int success, int fail, String title, String content,
|
||||
String resultMsg, String requestJson) {
|
||||
@@ -633,6 +671,21 @@ public class SmHotelMgrPushServiceImpl implements SmHotelMgrPushService {
|
||||
}
|
||||
}
|
||||
|
||||
/** 入箱后立刻扫一轮极光投递,避免只靠 Quartz(任务停了会一直不响) */
|
||||
private void kickPushDeliveryScan() {
|
||||
if (remotePushDeliveryService == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
R<Integer> resp = remotePushDeliveryService.processDueDeliveries();
|
||||
log.info("|酒店管理推送|踢投递扫描|claimed={}|ok={}|",
|
||||
resp == null ? null : resp.getData(), resp != null && resp.isOk());
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.warn("|酒店管理推送|踢投递扫描失败|{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 本部门负责人(不含下级):upms listDeptLeader,岗位编码 TEAM_LEADER。
|
||||
* Feign 反序列化可能把 id 解成 Integer,这里统一转 Long。
|
||||
|
||||
+18
-18
@@ -252,7 +252,7 @@ public class SmHotelMidServiceImpl implements SmHotelMidService {
|
||||
hotelPullTx.executeWithoutResult(status -> {
|
||||
hotelMapper.updateById(hotel);
|
||||
counts[0] = upsertRoomsFromComplex(hotel, complex);
|
||||
ensureRoomsBatch(hotel, roomTypesFromPrice);
|
||||
ensureRoomsBatch(hotel, roomTypesFromPrice);
|
||||
counts[1] = persistProducts(hotel, productMap);
|
||||
counts[2] = persistDailies(hotel, dailyDrafts, start, endExclusive);
|
||||
});
|
||||
@@ -2916,8 +2916,8 @@ public class SmHotelMidServiceImpl implements SmHotelMidService {
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
private int persistDailiesOnce(SmHotelEntity hotel, List<DailyDraft> drafts, LocalDate fromInclusive,
|
||||
LocalDate toExclusive) {
|
||||
@@ -2969,15 +2969,15 @@ public class SmHotelMidServiceImpl implements SmHotelMidService {
|
||||
|
||||
Map<String, DailyDraft> unique = new LinkedHashMap<>();
|
||||
if (CollUtil.isNotEmpty(drafts)) {
|
||||
for (DailyDraft d : drafts) {
|
||||
for (DailyDraft d : drafts) {
|
||||
if (d == null || d.stayDate == null || StrUtil.isBlank(d.productCode)) {
|
||||
continue;
|
||||
}
|
||||
if (d.stayDate.isBefore(winFrom) || !d.stayDate.isBefore(winToEx)) {
|
||||
continue;
|
||||
}
|
||||
unique.put(d.productCode + "|" + d.stayDate, d);
|
||||
}
|
||||
unique.put(d.productCode + "|" + d.stayDate, d);
|
||||
}
|
||||
}
|
||||
|
||||
List<SmHotelDailyEntity> existRows = dailyMapper.selectList(Wrappers.<SmHotelDailyEntity>lambdaQuery()
|
||||
@@ -3005,19 +3005,19 @@ public class SmHotelMidServiceImpl implements SmHotelMidService {
|
||||
String key = d.productCode + "|" + d.stayDate;
|
||||
SmHotelDailyEntity exist = existMap.remove(key);
|
||||
if (exist == null) {
|
||||
SmHotelDailyEntity daily = new SmHotelDailyEntity();
|
||||
daily.setHotelId(hotel.getId());
|
||||
daily.setProductId(productId);
|
||||
daily.setRoomTypeCode(d.roomTypeCode);
|
||||
daily.setProductCode(d.productCode);
|
||||
daily.setStayDate(d.stayDate);
|
||||
daily.setCostPrice(d.costPrice);
|
||||
daily.setSellPrice(d.sellPrice);
|
||||
daily.setStock(d.stock);
|
||||
SmHotelDailyEntity daily = new SmHotelDailyEntity();
|
||||
daily.setHotelId(hotel.getId());
|
||||
daily.setProductId(productId);
|
||||
daily.setRoomTypeCode(d.roomTypeCode);
|
||||
daily.setProductCode(d.productCode);
|
||||
daily.setStayDate(d.stayDate);
|
||||
daily.setCostPrice(d.costPrice);
|
||||
daily.setSellPrice(d.sellPrice);
|
||||
daily.setStock(d.stock);
|
||||
daily.setSold(0);
|
||||
daily.setDailyStatus(d.dailyStatus);
|
||||
fillNewMeta(daily, tenantId, now);
|
||||
toInsert.add(daily);
|
||||
daily.setDailyStatus(d.dailyStatus);
|
||||
fillNewMeta(daily, tenantId, now);
|
||||
toInsert.add(daily);
|
||||
continue;
|
||||
}
|
||||
// 有则改:保留 sold,避免冲掉本地占房
|
||||
|
||||
+88
-4
@@ -112,11 +112,16 @@ public class SmHotelShengnuoOrderServiceImpl implements SmHotelShengnuoOrderServ
|
||||
if (exist != null) {
|
||||
if (isBookConfirmed(exist)) {
|
||||
pushOrderChange(partnerCode, orderId, ST_ACCEPTED);
|
||||
return ok("订单已存在(幂等)", exist.getId());
|
||||
}
|
||||
// 预订失败单允许同渠道单号重新预检;其它中间态仍幂等
|
||||
if (!"BOOK_FAIL".equals(exist.getOrderStatus())) {
|
||||
return ok("订单已存在(幂等)", exist.getId());
|
||||
}
|
||||
return ok("订单已存在(幂等)", exist.getId());
|
||||
}
|
||||
Mapped m = mapCreate(partnerCode, body);
|
||||
if (m.error != null) {
|
||||
saveBookFailOrder(exist, partnerCode, orderId, m, body, m.error);
|
||||
return fail(m.error);
|
||||
}
|
||||
Map<String, Object> checkBody = new LinkedHashMap<>();
|
||||
@@ -137,7 +142,9 @@ public class SmHotelShengnuoOrderServiceImpl implements SmHotelShengnuoOrderServ
|
||||
boolean priceSoft = checkData != null && Boolean.TRUE.equals(checkData.get("priceMismatchOnly"));
|
||||
if (!bookable && !priceSoft) {
|
||||
Object reasons = checkData == null ? "预检失败" : checkData.get("reasons");
|
||||
return fail("预检不可订:" + reasons);
|
||||
String failMsg = "预检不可订:" + reasons;
|
||||
saveBookFailOrder(exist, partnerCode, orderId, m, body, failMsg);
|
||||
return fail(failMsg);
|
||||
}
|
||||
String priceWarn = null;
|
||||
if (priceSoft) {
|
||||
@@ -147,10 +154,11 @@ public class SmHotelShengnuoOrderServiceImpl implements SmHotelShengnuoOrderServ
|
||||
log.warn("|{}|价差软通过|orderId={}|{}|", TAG, orderId, priceWarn);
|
||||
}
|
||||
|
||||
SmHotelOrderEntity order = new SmHotelOrderEntity();
|
||||
SmHotelOrderEntity order = exist != null ? exist : new SmHotelOrderEntity();
|
||||
order.setPartnerCode(partnerCode);
|
||||
order.setChannelNo(orderId);
|
||||
order.setHotelId(m.hotel.getId());
|
||||
order.setDeptId(m.hotel.getDeptId());
|
||||
order.setHotelCode(m.hotelCode);
|
||||
order.setGroupCode(m.hotel.getGroupCode());
|
||||
order.setRoomTypeCode(m.roomTypeCode);
|
||||
@@ -168,12 +176,18 @@ public class SmHotelShengnuoOrderServiceImpl implements SmHotelShengnuoOrderServ
|
||||
order.setCompensateStatus("NONE");
|
||||
order.setInventoryStatus("NONE");
|
||||
order.setCompensateRound(0);
|
||||
order.setFailMsg(null);
|
||||
order.setReqJson(body.toString());
|
||||
order.setHuanqiPayload(body.toString());
|
||||
if (priceWarn != null) {
|
||||
order.setFailMsg(StrUtil.maxLength(priceWarn, 500));
|
||||
}
|
||||
orderMapper.insert(order);
|
||||
if (order.getId() == null) {
|
||||
orderMapper.insert(order);
|
||||
}
|
||||
else {
|
||||
orderMapper.updateById(order);
|
||||
}
|
||||
|
||||
R<Boolean> book = hotelOrderService.bookUpstreamForExisting(order.getId());
|
||||
SmHotelOrderEntity latest = orderMapper.selectById(order.getId());
|
||||
@@ -300,6 +314,9 @@ public class SmHotelShengnuoOrderServiceImpl implements SmHotelShengnuoOrderServ
|
||||
m.error = "未找到本地酒店(请先推送资源建立映射,或核对外部酒店ID/sHotelId)nsHotelId=" + nsHotelId;
|
||||
return m;
|
||||
}
|
||||
// 后续匹配失败时仍带上酒店,便于预订失败单挂到对应门店
|
||||
m.hotel = hotel;
|
||||
m.hotelCode = hotel.getHotelCode();
|
||||
String roomType = idMappingCache.roomTypeByNs(map, nsRoomId);
|
||||
if (StrUtil.isBlank(roomType)) {
|
||||
roomType = matchLocalRoomType(hotel.getId(), nsRoomId, StrUtil.trim(str(body.get("room_type"))), map);
|
||||
@@ -600,11 +617,78 @@ public class SmHotelShengnuoOrderServiceImpl implements SmHotelShengnuoOrderServ
|
||||
}
|
||||
return StrUtil.isNotBlank(order.getPlatformNo())
|
||||
&& !"UPSTREAM_FAIL".equals(order.getOrderStatus())
|
||||
&& !"BOOK_FAIL".equals(order.getOrderStatus())
|
||||
&& !"NEED_MANUAL".equals(order.getOrderStatus())
|
||||
&& !"CANCELLED".equals(order.getOrderStatus())
|
||||
&& !"PENDING_CANCEL".equals(order.getOrderStatus());
|
||||
}
|
||||
|
||||
/**
|
||||
* 入站预检/映射失败也落酒店订单(BOOK_FAIL),便于运营在订单列表实时看到进单与原因;对下游仍返回失败。
|
||||
*/
|
||||
private Long saveBookFailOrder(SmHotelOrderEntity exist, String partnerCode, String orderId, Mapped m,
|
||||
JSONObject body, String failMsg) {
|
||||
try {
|
||||
SmHotelOrderEntity order = exist != null ? exist : new SmHotelOrderEntity();
|
||||
order.setPartnerCode(partnerCode);
|
||||
order.setChannelNo(orderId);
|
||||
order.setOrderStatus("BOOK_FAIL");
|
||||
order.setCompensateStatus("NONE");
|
||||
order.setInventoryStatus("NONE");
|
||||
order.setCompensateRound(0);
|
||||
order.setMarkupPercent(BigDecimal.ZERO);
|
||||
order.setFailMsg(StrUtil.maxLength(failMsg, 500));
|
||||
order.setReqJson(body == null ? null : body.toString());
|
||||
order.setHuanqiPayload(body == null ? null : body.toString());
|
||||
if (m != null && m.hotel != null) {
|
||||
order.setHotelId(m.hotel.getId());
|
||||
order.setDeptId(m.hotel.getDeptId());
|
||||
order.setHotelCode(m.hotelCode);
|
||||
order.setGroupCode(m.hotel.getGroupCode());
|
||||
order.setRoomTypeCode(m.roomTypeCode);
|
||||
order.setProductCode(m.productCode);
|
||||
order.setPriceCode(m.product == null ? null : m.product.getPriceCode());
|
||||
order.setCheckInDate(m.checkIn);
|
||||
order.setCheckOutDate(m.checkOut);
|
||||
order.setCount(m.quantity <= 0 ? 1 : m.quantity);
|
||||
order.setContactName(m.contactName);
|
||||
order.setContactPhone(m.contactPhone);
|
||||
order.setGuestJson(m.guestJson);
|
||||
order.setSellAmount(m.orderAmount);
|
||||
}
|
||||
else if (body != null) {
|
||||
long nsHotelId = toLong(body.get("hotel_id"));
|
||||
if (nsHotelId > 0) {
|
||||
order.setHotelCode(String.valueOf(nsHotelId));
|
||||
}
|
||||
order.setCheckInDate(parseDay(str(body.get("checkIn"))));
|
||||
order.setCheckOutDate(parseDay(str(body.get("checkOut"))));
|
||||
order.setCount(Math.max(1, (int) toLong(body.get("room_number"))));
|
||||
order.setContactName(StrUtil.blankToDefault(StrUtil.trim(str(body.get("check_person_name"))), "旅客"));
|
||||
order.setContactPhone(StrUtil.trim(str(body.get("check_person_phone"))));
|
||||
String priceStr = StrUtil.blankToDefault(str(body.get("total_prices")), str(body.get("pay_total_prices")));
|
||||
try {
|
||||
order.setSellAmount(new BigDecimal(StrUtil.blankToDefault(priceStr, "0")));
|
||||
}
|
||||
catch (Exception ignored) {
|
||||
order.setSellAmount(BigDecimal.ZERO);
|
||||
}
|
||||
}
|
||||
if (order.getId() == null) {
|
||||
orderMapper.insert(order);
|
||||
}
|
||||
else {
|
||||
orderMapper.updateById(order);
|
||||
}
|
||||
log.info("|{}|预订失败已落单|orderId={}|localId={}|msg={}|", TAG, orderId, order.getId(), failMsg);
|
||||
return order.getId();
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.warn("|{}|预订失败落单异常|orderId={}|{}|", TAG, orderId, e.getMessage());
|
||||
return exist == null ? null : exist.getId();
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isPendingConfirm(SmHotelOrderEntity order) {
|
||||
return order != null && "PENDING_CONFIRM".equals(order.getOrderStatus());
|
||||
}
|
||||
|
||||
+21
-21
@@ -564,10 +564,10 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
}
|
||||
}
|
||||
else {
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getSyncType, syncType)
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>lambdaQuery()
|
||||
.eq(SmHotelSyncLogEntity::getSyncType, syncType)
|
||||
.eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING));
|
||||
if (running != null && running > 0) {
|
||||
return R.failed("已有同类型任务进行中,请稍后再试(若卡住请先暂停/终止)");
|
||||
}
|
||||
}
|
||||
@@ -1160,7 +1160,7 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
+ remain + "(每批 " + batchSize + ")");
|
||||
R<Map<String, Object>> imp = importDiffNewHotels(batchSize, dataSource);
|
||||
int n = 0;
|
||||
if (imp != null && imp.getData() != null && imp.getData().get("insertCount") != null) {
|
||||
if (imp != null && imp.getData() != null && imp.getData().get("insertCount") != null) {
|
||||
n = Integer.parseInt(String.valueOf(imp.getData().get("insertCount")));
|
||||
}
|
||||
if (n <= 0) {
|
||||
@@ -2028,12 +2028,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
processed.incrementAndGet();
|
||||
}
|
||||
int doneCnt = processed.get();
|
||||
if (intervalMs > 0) {
|
||||
try {
|
||||
Thread.sleep(intervalMs);
|
||||
}
|
||||
catch (InterruptedException ie) {
|
||||
Thread.currentThread().interrupt();
|
||||
if (intervalMs > 0) {
|
||||
try {
|
||||
Thread.sleep(intervalMs);
|
||||
}
|
||||
catch (InterruptedException ie) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
if (doneCnt % 20 == 0 || doneCnt >= target) {
|
||||
@@ -2882,10 +2882,10 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
continue;
|
||||
}
|
||||
codes.add(code);
|
||||
if (codes.size() >= batch) {
|
||||
updated += updateStatusByCodes(codes, "OFFLINE");
|
||||
codes.clear();
|
||||
}
|
||||
if (codes.size() >= batch) {
|
||||
updated += updateStatusByCodes(codes, "OFFLINE");
|
||||
codes.clear();
|
||||
}
|
||||
}
|
||||
if (!codes.isEmpty()) {
|
||||
updated += updateStatusByCodes(codes, "OFFLINE");
|
||||
@@ -2968,12 +2968,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
|
||||
int to = (int) Math.min(skip + size, ordered.size());
|
||||
for (int i = from; i < to; i++) {
|
||||
String hotelId = ordered.get(i);
|
||||
Map<String, Object> row = new LinkedHashMap<>(4);
|
||||
row.put("hotelId", hotelId);
|
||||
row.put("hotelName", upstreamClient.hotelNameOf(hotelId));
|
||||
row.put("type", "offline".equalsIgnoreCase(type) ? "offline" : "new");
|
||||
records.add(row);
|
||||
}
|
||||
Map<String, Object> row = new LinkedHashMap<>(4);
|
||||
row.put("hotelId", hotelId);
|
||||
row.put("hotelName", upstreamClient.hotelNameOf(hotelId));
|
||||
row.put("type", "offline".equalsIgnoreCase(type) ? "offline" : "new");
|
||||
records.add(row);
|
||||
}
|
||||
}
|
||||
}
|
||||
Map<String, Object> result = new LinkedHashMap<>();
|
||||
|
||||
+3
-1
@@ -6,6 +6,7 @@ import com.bsd.pjcloud.common.swagger.annotation.EnableOpenApi;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
|
||||
/**
|
||||
* @author archetype auto
|
||||
@@ -16,7 +17,8 @@ import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
|
||||
@EnablePjcloudFeignClients
|
||||
@EnableDiscoveryClient
|
||||
@EnablePjcloudResourceServer
|
||||
// 周期调度已统一迁移到 daemon-quartz(sys_job 注册 thirdPushTask / 支付任务租约表),集群下避免各实例重复触发
|
||||
// daemon-quartz thirdPushTask 仍为主调度;本地 @Scheduled 仅作系统通知 5 秒兜底(CAS 领取,可与 Quartz 并存)
|
||||
@EnableScheduling
|
||||
@SpringBootApplication
|
||||
public class ThirdPartyApp {
|
||||
public static void main(String[] args) {
|
||||
|
||||
+2
-2
@@ -109,8 +109,8 @@ public class AppMessageConstants {
|
||||
/** 单个投递任务允许的最大供应商调用次数。 */
|
||||
public static final int MAX_DELIVERY_ATTEMPTS = 8;
|
||||
|
||||
/** 处理中超过该分钟数视为任务失联并允许恢复。 */
|
||||
public static final int PROCESSING_TIMEOUT_MINUTES = 10;
|
||||
/** 处理中超过该分钟数视为任务失联并允许恢复(原 10 分钟过长,系统通知会像「卡死」)。 */
|
||||
public static final int PROCESSING_TIMEOUT_MINUTES = 2;
|
||||
|
||||
/** 一次投递扫描领取的最大任务数,控制数据库和供应商压力。 */
|
||||
public static final int DELIVERY_SCAN_LIMIT = 100;
|
||||
|
||||
+8
-2
@@ -1,5 +1,6 @@
|
||||
package com.bsd.pjcloud.thirdparty.controller.appmessage;
|
||||
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
||||
import com.bsd.pjcloud.common.api.encrypt.annotation.crypto.ApiCryptoAes;
|
||||
import com.bsd.pjcloud.common.core.util.R;
|
||||
@@ -98,14 +99,19 @@ public class ThirdAppMessageController {
|
||||
|
||||
/**
|
||||
* 内部幂等接收酒店管理推送(不进入商户策略/关系订阅矩阵;独立于有房通知)。
|
||||
* <p>收件箱写入成功但系统通知被跳过(无设备/用户关推送)时返回失败文案,避免上游误记为「手机已送达」。
|
||||
*
|
||||
* @param dto 超市酒店管理推送事件快照
|
||||
* @return true表示消息中心已完整接收
|
||||
* @return true 表示入箱且系统通知已进入投递;失败时 msg 说明跳过原因
|
||||
*/
|
||||
@Inner
|
||||
@PostMapping("/inner/hotel-manager-push-event")
|
||||
public R<Boolean> receiveHotelManagerPushEvent(@Valid @RequestBody HotelManagerPushEventDto dto) {
|
||||
return R.ok(TenantBroker.applyAs(dto.getTenantId(), tenantId -> hotelManagerNotifyService.receive(dto)));
|
||||
String skipTip = TenantBroker.applyAs(dto.getTenantId(), tenantId -> hotelManagerNotifyService.receive(dto));
|
||||
if (StrUtil.isNotBlank(skipTip)) {
|
||||
return R.failed(skipTip);
|
||||
}
|
||||
return R.ok(Boolean.TRUE);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
package com.bsd.pjcloud.thirdparty.job;
|
||||
|
||||
import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* App 消息系统通知本地兜底扫描。
|
||||
* <p>正式调度仍以 daemon-quartz thirdPushTask 为准;本地短周期扫描保证 Quartz 停摆/积压时
|
||||
* 系统通知不会拖到数分钟。领取靠 status CAS,多实例重复触发安全。
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class ThirdAppMessagePushJob {
|
||||
|
||||
private final ThirdPushDeliveryService deliveryService;
|
||||
|
||||
/** 上轮结束后再隔 5 秒扫一轮 PENDING/RETRY */
|
||||
@Scheduled(fixedDelay = 5000L, initialDelay = 8000L)
|
||||
public void processDueDeliveries() {
|
||||
try {
|
||||
int claimed = deliveryService.processDueDeliveries();
|
||||
if (claimed > 0) {
|
||||
log.info("|App消息Push|本地兜底扫描完成|claimed={}|", claimed);
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("|App消息Push|本地兜底扫描异常|{}", e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -11,7 +11,7 @@ public interface ThirdHotelManagerNotifyService {
|
||||
* 幂等接收酒店管理推送事件。
|
||||
*
|
||||
* @param dto 超市模块提交的事件快照
|
||||
* @return true 表示已入箱或此前已幂等完成
|
||||
* @return null 表示入箱且系统通知已进入待投递/已幂等;非空为跳过系统通知的中文原因(收件箱通常已写入)
|
||||
*/
|
||||
boolean receive(HotelManagerPushEventDto dto);
|
||||
String receive(HotelManagerPushEventDto dto);
|
||||
}
|
||||
+10
-1
@@ -18,8 +18,9 @@ public interface ThirdPushDeliveryService {
|
||||
*
|
||||
* @param message 已持久化的收件箱消息
|
||||
* @param skipReason 稳定跳过原因,允许为空
|
||||
* @return 最终跳过原因;null 表示已进入待投递(或已有投递记录)
|
||||
*/
|
||||
void createDelivery(ThirdAppMessageEntity message, String skipReason);
|
||||
String createDelivery(ThirdAppMessageEntity message, String skipReason);
|
||||
|
||||
/**
|
||||
* 扫描并处理所有租户到期的投递任务,供轻量定时任务调用。
|
||||
@@ -28,6 +29,14 @@ public interface ThirdPushDeliveryService {
|
||||
*/
|
||||
int processDueDeliveries();
|
||||
|
||||
/**
|
||||
* 事务提交后立即尝试投递指定收件箱消息(PENDING),不坐等 Quartz 扫描。
|
||||
*
|
||||
* @param messageId 收件箱消息ID
|
||||
* @return 是否成功领取并处理
|
||||
*/
|
||||
boolean processPendingByMessageId(Long messageId);
|
||||
|
||||
/**
|
||||
* 分页查询指定商户的业务消息投递记录。
|
||||
*
|
||||
|
||||
+52
-5
@@ -10,15 +10,20 @@ import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdAppMessageSettingServi
|
||||
import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdHotelManagerNotifyService;
|
||||
import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.support.TransactionSynchronization;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
/**
|
||||
* 酒店管理推送:仿有房通知旁路,独立 eventCode,避免进入商户策略矩阵。
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class ThirdHotelManagerNotifyServiceImpl implements ThirdHotelManagerNotifyService {
|
||||
@@ -35,7 +40,7 @@ public class ThirdHotelManagerNotifyServiceImpl implements ThirdHotelManagerNoti
|
||||
|
||||
@Override
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public boolean receive(HotelManagerPushEventDto dto) {
|
||||
public String receive(HotelManagerPushEventDto dto) {
|
||||
if (dto == null || StrUtil.isBlank(dto.getEventId()) || dto.getRecipientUserId() == null) {
|
||||
throw new IllegalArgumentException("酒店管理推送参数不完整");
|
||||
}
|
||||
@@ -44,7 +49,7 @@ public class ThirdHotelManagerNotifyServiceImpl implements ThirdHotelManagerNoti
|
||||
throw new IllegalArgumentException("不支持的酒店管理推送事件编码: " + eventCode);
|
||||
}
|
||||
if (ownerMessageExists(dto.getEventId(), dto.getRecipientUserId())) {
|
||||
return true;
|
||||
return null;
|
||||
}
|
||||
String title = trimToMax(dto.getTitle(), AppMessageConstants.MESSAGE_TITLE_MAX_LENGTH);
|
||||
String content = trimToMax(dto.getContent(), AppMessageConstants.MESSAGE_CONTENT_MAX_LENGTH);
|
||||
@@ -74,13 +79,45 @@ public class ThirdHotelManagerNotifyServiceImpl implements ThirdHotelManagerNoti
|
||||
}
|
||||
}
|
||||
catch (DuplicateKeyException conflict) {
|
||||
return true;
|
||||
return null;
|
||||
}
|
||||
String skipReason = settingService.isUserPushEnabled(dto.getRecipientUserId())
|
||||
? null
|
||||
: AppMessageConstants.SKIP_USER_DISABLED;
|
||||
deliveryService.createDelivery(message, skipReason);
|
||||
return true;
|
||||
String resolvedSkip = deliveryService.createDelivery(message, skipReason);
|
||||
if (resolvedSkip != null) {
|
||||
log.warn("|酒店管理推送|系统通知跳过|userId={}|eventId={}|reason={}|",
|
||||
dto.getRecipientUserId(), dto.getEventId(), resolvedSkip);
|
||||
return explainSkip(resolvedSkip);
|
||||
}
|
||||
scheduleImmediateDelivery(message.getId());
|
||||
return null;
|
||||
}
|
||||
|
||||
/** 事务提交后异步立刻投递,避免 PENDING 未可见,也不阻塞 Feign 返回 */
|
||||
private void scheduleImmediateDelivery(Long messageId) {
|
||||
if (messageId == null) {
|
||||
return;
|
||||
}
|
||||
Runnable task = () -> CompletableFuture.runAsync(() -> {
|
||||
try {
|
||||
boolean ok = deliveryService.processPendingByMessageId(messageId);
|
||||
log.info("|酒店管理推送|立即投递|messageId={}|claimed={}|", messageId, ok);
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.warn("|酒店管理推送|立即投递异常|messageId={}|{}", messageId, e.getMessage());
|
||||
}
|
||||
});
|
||||
if (TransactionSynchronizationManager.isSynchronizationActive()) {
|
||||
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
|
||||
@Override
|
||||
public void afterCommit() {
|
||||
task.run();
|
||||
}
|
||||
});
|
||||
return;
|
||||
}
|
||||
task.run();
|
||||
}
|
||||
|
||||
private boolean ownerMessageExists(String eventId, Long userId) {
|
||||
@@ -90,6 +127,16 @@ public class ThirdHotelManagerNotifyServiceImpl implements ThirdHotelManagerNoti
|
||||
return count != null && count > 0;
|
||||
}
|
||||
|
||||
private static String explainSkip(String code) {
|
||||
if (AppMessageConstants.SKIP_NO_DEVICE.equals(code)) {
|
||||
return "收件箱已写入,但该用户未绑定推送设备,无法发系统通知(请用手机打开 App 并登录以绑定极光设备)";
|
||||
}
|
||||
if (AppMessageConstants.SKIP_USER_DISABLED.equals(code)) {
|
||||
return "收件箱已写入,但该用户关闭了消息推送开关";
|
||||
}
|
||||
return "收件箱已写入,系统通知跳过:" + code;
|
||||
}
|
||||
|
||||
private static String trimToMax(String raw, int max) {
|
||||
String text = StrUtil.nullToEmpty(raw).trim();
|
||||
if (text.length() <= max) {
|
||||
|
||||
+94
-38
@@ -96,12 +96,15 @@ public class ThirdPushDeliveryServiceImpl implements ThirdPushDeliveryService {
|
||||
*
|
||||
* @param message 已持久化的收件箱消息
|
||||
* @param skipReason 业务策略跳过原因,允许为空
|
||||
* @return 最终跳过原因;null 表示待投递或已有投递记录
|
||||
*/
|
||||
@Override
|
||||
public void createDelivery(ThirdAppMessageEntity message, String skipReason) {
|
||||
if (deliveryMapper.selectCount(Wrappers.<ThirdPushDeliveryEntity>lambdaQuery()
|
||||
.eq(ThirdPushDeliveryEntity::getMessageId, message.getId())) > 0) {
|
||||
return;
|
||||
public String createDelivery(ThirdAppMessageEntity message, String skipReason) {
|
||||
ThirdPushDeliveryEntity existing = deliveryMapper.selectOne(Wrappers.<ThirdPushDeliveryEntity>lambdaQuery()
|
||||
.eq(ThirdPushDeliveryEntity::getMessageId, message.getId())
|
||||
.last("limit 1"));
|
||||
if (existing != null) {
|
||||
return existing.getSkipReason();
|
||||
}
|
||||
String resolvedSkipReason = skipReason;
|
||||
if (resolvedSkipReason == null && !deviceBindingService.hasActiveBinding(message.getRecipientUserId())) {
|
||||
@@ -118,6 +121,33 @@ public class ThirdPushDeliveryServiceImpl implements ThirdPushDeliveryService {
|
||||
delivery.setSkipReason(resolvedSkipReason);
|
||||
delivery.setAttemptCount(0);
|
||||
deliveryMapper.insert(delivery);
|
||||
return resolvedSkipReason;
|
||||
}
|
||||
|
||||
/**
|
||||
* 立即处理指定消息的 PENDING 投递(事务提交后调用)。
|
||||
*/
|
||||
@Override
|
||||
public boolean processPendingByMessageId(Long messageId) {
|
||||
if (messageId == null) {
|
||||
return false;
|
||||
}
|
||||
ThirdPushDeliveryEntity candidate = deliveryMapper.selectOne(Wrappers.<ThirdPushDeliveryEntity>lambdaQuery()
|
||||
.eq(ThirdPushDeliveryEntity::getMessageId, messageId)
|
||||
.eq(ThirdPushDeliveryEntity::getStatus, AppMessageConstants.DELIVERY_PENDING)
|
||||
.last("limit 1"));
|
||||
if (candidate == null || candidate.getTenantId() == null) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
Boolean claimed = TenantBroker.applyAs(candidate.getTenantId(),
|
||||
tenantId -> processOne(candidate.getId()));
|
||||
return Boolean.TRUE.equals(claimed);
|
||||
}
|
||||
catch (Exception exception) {
|
||||
log.error("立即投递失败,messageId:{},deliveryId:{}", messageId, candidate.getId(), exception);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -146,8 +176,9 @@ public class ThirdPushDeliveryServiceImpl implements ThirdPushDeliveryService {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
Boolean claimed = TenantBroker.applyAs(candidate.getTenantId(), tenantId ->
|
||||
transactionTemplate.execute(status -> processOne(candidate.getId())));
|
||||
// 极光调用不能包在长事务里:否则领取后进程抖动会卡在 PROCESSING 直到超时(原先 10 分钟)
|
||||
Boolean claimed = TenantBroker.applyAs(candidate.getTenantId(),
|
||||
tenantId -> processOne(candidate.getId()));
|
||||
if (Boolean.TRUE.equals(claimed)) {
|
||||
claimedCount++;
|
||||
}
|
||||
@@ -361,35 +392,20 @@ public class ThirdPushDeliveryServiceImpl implements ThirdPushDeliveryService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 领取并处理单条投递任务,条件更新保证多实例只有一个节点获得处理权。
|
||||
* 领取并处理单条投递任务:短事务领取 → 事务外调极光 → 短事务落终态。
|
||||
* 条件更新保证多实例只有一个节点获得处理权。
|
||||
*
|
||||
* @param deliveryId 投递任务ID
|
||||
* @return 是否成功领取任务
|
||||
*/
|
||||
private boolean processOne(Long deliveryId) {
|
||||
ThirdPushDeliveryEntity current = deliveryMapper.selectById(deliveryId);
|
||||
if (current == null || current.getAttemptCount() >= AppMessageConstants.MAX_DELIVERY_ATTEMPTS) {
|
||||
return false;
|
||||
}
|
||||
ThirdPushDeliveryEntity claim = new ThirdPushDeliveryEntity();
|
||||
claim.setStatus(AppMessageConstants.DELIVERY_PROCESSING);
|
||||
claim.setNextRetryTime(null);
|
||||
claim.setUpdateTime(LocalDateTime.now());
|
||||
var claimWrapper = Wrappers.<ThirdPushDeliveryEntity>lambdaUpdate()
|
||||
.eq(ThirdPushDeliveryEntity::getId, deliveryId)
|
||||
.eq(ThirdPushDeliveryEntity::getStatus, current.getStatus());
|
||||
if (AppMessageConstants.DELIVERY_PROCESSING.equals(current.getStatus())) {
|
||||
// 超时任务仍保持PROCESSING状态,必须附加旧更新时间条件,防止多个实例同时把同一任务领取成功。
|
||||
claimWrapper.le(ThirdPushDeliveryEntity::getUpdateTime,
|
||||
LocalDateTime.now().minusMinutes(AppMessageConstants.PROCESSING_TIMEOUT_MINUTES));
|
||||
}
|
||||
int claimed = deliveryMapper.update(claim, claimWrapper);
|
||||
if (claimed == 0) {
|
||||
ThirdPushDeliveryEntity current = transactionTemplate.execute(status -> tryClaim(deliveryId));
|
||||
if (current == null) {
|
||||
return false;
|
||||
}
|
||||
ThirdAppMessageEntity message = messageMapper.selectById(current.getMessageId());
|
||||
if (message == null) {
|
||||
markTerminalFailure(current, "MESSAGE_NOT_FOUND", 1);
|
||||
transactionTemplate.executeWithoutResult(status -> markTerminalFailure(current, "MESSAGE_NOT_FOUND", 1));
|
||||
return true;
|
||||
}
|
||||
PushProviderRequest request = new PushProviderRequest();
|
||||
@@ -400,7 +416,43 @@ public class ThirdPushDeliveryServiceImpl implements ThirdPushDeliveryService {
|
||||
"messageId", String.valueOf(message.getId()),
|
||||
"path", AppMessageConstants.MESSAGE_DETAIL_PATH));
|
||||
int remainingAttempts = AppMessageConstants.MAX_DELIVERY_ATTEMPTS - current.getAttemptCount();
|
||||
PushDispatchResult dispatchResult = providerDispatcher.dispatch(request, remainingAttempts);
|
||||
PushDispatchResult dispatchResult;
|
||||
try {
|
||||
dispatchResult = providerDispatcher.dispatch(request, remainingAttempts);
|
||||
}
|
||||
catch (Exception exception) {
|
||||
log.error("Push供应商调用异常,deliveryId:{}", deliveryId, exception);
|
||||
transactionTemplate.executeWithoutResult(status -> applyDispatchFailure(current, null,
|
||||
StrUtil.sub(exception.getMessage(), 0, 500), 1));
|
||||
return true;
|
||||
}
|
||||
transactionTemplate.executeWithoutResult(status -> applyDispatchResult(current, dispatchResult));
|
||||
return true;
|
||||
}
|
||||
|
||||
/** @return 领取成功时的投递快照;未领到返回 null */
|
||||
private ThirdPushDeliveryEntity tryClaim(Long deliveryId) {
|
||||
ThirdPushDeliveryEntity current = deliveryMapper.selectById(deliveryId);
|
||||
if (current == null || current.getAttemptCount() >= AppMessageConstants.MAX_DELIVERY_ATTEMPTS) {
|
||||
return null;
|
||||
}
|
||||
ThirdPushDeliveryEntity claim = new ThirdPushDeliveryEntity();
|
||||
claim.setStatus(AppMessageConstants.DELIVERY_PROCESSING);
|
||||
claim.setNextRetryTime(null);
|
||||
claim.setUpdateTime(LocalDateTime.now());
|
||||
var claimWrapper = Wrappers.<ThirdPushDeliveryEntity>lambdaUpdate()
|
||||
.eq(ThirdPushDeliveryEntity::getId, deliveryId)
|
||||
.eq(ThirdPushDeliveryEntity::getStatus, current.getStatus());
|
||||
if (AppMessageConstants.DELIVERY_PROCESSING.equals(current.getStatus())) {
|
||||
// 超时任务仍保持 PROCESSING,必须附加旧更新时间条件,防止多实例同时领取。
|
||||
claimWrapper.le(ThirdPushDeliveryEntity::getUpdateTime,
|
||||
LocalDateTime.now().minusMinutes(AppMessageConstants.PROCESSING_TIMEOUT_MINUTES));
|
||||
}
|
||||
int claimed = deliveryMapper.update(claim, claimWrapper);
|
||||
return claimed == 0 ? null : current;
|
||||
}
|
||||
|
||||
private void applyDispatchResult(ThirdPushDeliveryEntity current, PushDispatchResult dispatchResult) {
|
||||
int consumedAttempts = persistAttempts(current, dispatchResult);
|
||||
int newAttemptCount = current.getAttemptCount() + Math.max(consumedAttempts, 1);
|
||||
if (dispatchResult.isAccepted()) {
|
||||
@@ -414,28 +466,32 @@ public class ThirdPushDeliveryServiceImpl implements ThirdPushDeliveryService {
|
||||
deliveryMapper.update(success, Wrappers.<ThirdPushDeliveryEntity>lambdaUpdate()
|
||||
.eq(ThirdPushDeliveryEntity::getId, current.getId())
|
||||
.eq(ThirdPushDeliveryEntity::getStatus, AppMessageConstants.DELIVERY_PROCESSING));
|
||||
return true;
|
||||
return;
|
||||
}
|
||||
PushDispatchAttempt lastAttempt = dispatchResult.getAttempts().isEmpty()
|
||||
? null : dispatchResult.getAttempts().get(dispatchResult.getAttempts().size() - 1);
|
||||
String error = lastAttempt == null ? "NO_PROVIDER_AVAILABLE"
|
||||
: StrUtil.blankToDefault(lastAttempt.getResult().getErrorMessage(), lastAttempt.getResult().getErrorCode());
|
||||
applyDispatchFailure(current, lastAttempt, error, newAttemptCount);
|
||||
}
|
||||
|
||||
private void applyDispatchFailure(ThirdPushDeliveryEntity current, PushDispatchAttempt lastAttempt,
|
||||
String error, int newAttemptCount) {
|
||||
if (lastAttempt != null && "INVALID_DEVICE".equals(lastAttempt.getResult().getErrorCode())) {
|
||||
deviceBindingService.disableInvalidBindings(current.getRecipientUserId(), lastAttempt.getProviderCode());
|
||||
}
|
||||
if (newAttemptCount >= AppMessageConstants.MAX_DELIVERY_ATTEMPTS) {
|
||||
markTerminalFailure(current, error, newAttemptCount);
|
||||
} else {
|
||||
ThirdPushDeliveryEntity retry = new ThirdPushDeliveryEntity();
|
||||
retry.setStatus(AppMessageConstants.DELIVERY_RETRY);
|
||||
retry.setAttemptCount(newAttemptCount);
|
||||
retry.setNextRetryTime(LocalDateTime.now().plusMinutes(retryMinutes(newAttemptCount)));
|
||||
retry.setLastError(StrUtil.sub(error, 0, 2000));
|
||||
deliveryMapper.update(retry, Wrappers.<ThirdPushDeliveryEntity>lambdaUpdate()
|
||||
.eq(ThirdPushDeliveryEntity::getId, current.getId())
|
||||
.eq(ThirdPushDeliveryEntity::getStatus, AppMessageConstants.DELIVERY_PROCESSING));
|
||||
return;
|
||||
}
|
||||
return true;
|
||||
ThirdPushDeliveryEntity retry = new ThirdPushDeliveryEntity();
|
||||
retry.setStatus(AppMessageConstants.DELIVERY_RETRY);
|
||||
retry.setAttemptCount(newAttemptCount);
|
||||
retry.setNextRetryTime(LocalDateTime.now().plusMinutes(retryMinutes(newAttemptCount)));
|
||||
retry.setLastError(StrUtil.sub(error, 0, 2000));
|
||||
deliveryMapper.update(retry, Wrappers.<ThirdPushDeliveryEntity>lambdaUpdate()
|
||||
.eq(ThirdPushDeliveryEntity::getId, current.getId())
|
||||
.eq(ThirdPushDeliveryEntity::getStatus, AppMessageConstants.DELIVERY_PROCESSING));
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in new issue
Block a user