|
@@ -2,6 +2,7 @@ package com.payment.platform.module.payment.har.service;
|
|
|
|
|
|
|
|
import cn.hutool.core.util.StrUtil;
|
|
import cn.hutool.core.util.StrUtil;
|
|
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
|
|
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
|
|
|
import com.payment.platform.common.utils.SnowflakeIdGenerator;
|
|
import com.payment.platform.common.utils.SnowflakeIdGenerator;
|
|
|
import com.payment.platform.module.payment.apikey.entity.TenantApiKeyEntity;
|
|
import com.payment.platform.module.payment.apikey.entity.TenantApiKeyEntity;
|
|
|
import com.payment.platform.module.payment.apikey.mapper.TenantApiKeyMapper;
|
|
import com.payment.platform.module.payment.apikey.mapper.TenantApiKeyMapper;
|
|
@@ -26,7 +27,9 @@ import org.springframework.util.LinkedMultiValueMap;
|
|
|
import org.springframework.util.MultiValueMap;
|
|
import org.springframework.util.MultiValueMap;
|
|
|
import org.springframework.web.client.RestTemplate;
|
|
import org.springframework.web.client.RestTemplate;
|
|
|
|
|
|
|
|
|
|
+import java.time.OffsetDateTime;
|
|
|
import java.util.LinkedHashMap;
|
|
import java.util.LinkedHashMap;
|
|
|
|
|
+import java.util.List;
|
|
|
import java.util.Map;
|
|
import java.util.Map;
|
|
|
import java.util.concurrent.CompletableFuture;
|
|
import java.util.concurrent.CompletableFuture;
|
|
|
import java.util.concurrent.ExecutorService;
|
|
import java.util.concurrent.ExecutorService;
|
|
@@ -42,24 +45,24 @@ import java.util.concurrent.TimeUnit;
|
|
|
* 回调地址解析也沿用 OpenapiService.resolveReturnUrl 的优先级:
|
|
* 回调地址解析也沿用 OpenapiService.resolveReturnUrl 的优先级:
|
|
|
* sys_tenant_api_key.return_url → open_conf.return_url。
|
|
* sys_tenant_api_key.return_url → open_conf.return_url。
|
|
|
*
|
|
*
|
|
|
- * 两点刻意的设计:
|
|
|
|
|
|
|
+ * 三点刻意的设计:
|
|
|
*
|
|
*
|
|
|
* 1. 用 CompletableFuture + 自有线程池而非 @Async —— 项目没有开 @EnableAsync,
|
|
* 1. 用 CompletableFuture + 自有线程池而非 @Async —— 项目没有开 @EnableAsync,
|
|
|
* @Async 会静默地同步执行,把上游通知响应时间拖到 30 秒以上,直接超出上游
|
|
* @Async 会静默地同步执行,把上游通知响应时间拖到 30 秒以上,直接超出上游
|
|
|
* 25 秒的重试窗口。为了一个回调去开全局 @EnableAsync 又会改变其他地方的语义。
|
|
* 25 秒的重试窗口。为了一个回调去开全局 @EnableAsync 又会改变其他地方的语义。
|
|
|
*
|
|
*
|
|
|
- * 2. 回调失败只记日志并落 notify_status,不抛异常 —— 上游通知要求我们尽快返回
|
|
|
|
|
- * success,不能被下游回调拖住。
|
|
|
|
|
|
|
+ * 2. 推送**单次**、失败只记录不抛 —— 上游通知要求我们尽快返回 success,不能被
|
|
|
|
|
+ * 下游回调拖住;线程内退避重试(旧实现睡 1s、2s 共 3 次)会占满回调池。
|
|
|
|
|
+ * 重试交给调度器 {@link #retryFailed}:有次数上限、不占回调线程。
|
|
|
|
|
+ *
|
|
|
|
|
+ * 3. 缺地址与推送失败**统一记 FAILED**(不再静默跳过)—— 重试扫描的入口条件
|
|
|
|
|
+ * 就是 notify_status='FAILED',只打日志的话,商户事后补配地址也永远等不到
|
|
|
|
|
+ * 这一笔(2026-10-09 补齐)。
|
|
|
*/
|
|
*/
|
|
|
@Slf4j
|
|
@Slf4j
|
|
|
@Service
|
|
@Service
|
|
|
public class HarCallbackService {
|
|
public class HarCallbackService {
|
|
|
|
|
|
|
|
- private static final int MAX_ATTEMPTS = 3;
|
|
|
|
|
-
|
|
|
|
|
- /** 退避秒数,与 OpenapiService.CALLBACK_BACKOFF_SECONDS 保持一致 */
|
|
|
|
|
- private static final int[] BACKOFF_SECONDS = {1, 2};
|
|
|
|
|
-
|
|
|
|
|
/** 回调报文的序列化器。显式注册 JavaTimeModule,保证 OffsetDateTime 输出 ISO 字符串 */
|
|
/** 回调报文的序列化器。显式注册 JavaTimeModule,保证 OffsetDateTime 输出 ISO 字符串 */
|
|
|
private static final ObjectMapper MAPPER = new ObjectMapper()
|
|
private static final ObjectMapper MAPPER = new ObjectMapper()
|
|
|
.registerModule(new JavaTimeModule())
|
|
.registerModule(new JavaTimeModule())
|
|
@@ -71,6 +74,8 @@ public class HarCallbackService {
|
|
|
private final RestTemplate restTemplate;
|
|
private final RestTemplate restTemplate;
|
|
|
/** 订单信号发布(订单实时推送)—— 回调状态(notify_status)落库后发。 */
|
|
/** 订单信号发布(订单实时推送)—— 回调状态(notify_status)落库后发。 */
|
|
|
private final OrderEventPublisher publisher;
|
|
private final OrderEventPublisher publisher;
|
|
|
|
|
+ /** 回调尝试次数上限(含首次)—— 重试封顶对齐乐收赢 downstreamNotifyMaxRetries */
|
|
|
|
|
+ private final HarProperties harProperties;
|
|
|
private final ExecutorService callbackPool;
|
|
private final ExecutorService callbackPool;
|
|
|
|
|
|
|
|
public HarCallbackService(TenantApiKeyMapper apiKeyMapper,
|
|
public HarCallbackService(TenantApiKeyMapper apiKeyMapper,
|
|
@@ -83,6 +88,7 @@ public class HarCallbackService {
|
|
|
this.confMapper = confMapper;
|
|
this.confMapper = confMapper;
|
|
|
this.orderMapper = orderMapper;
|
|
this.orderMapper = orderMapper;
|
|
|
this.restTemplate = restTemplate;
|
|
this.restTemplate = restTemplate;
|
|
|
|
|
+ this.harProperties = harProperties;
|
|
|
this.publisher = publisher;
|
|
this.publisher = publisher;
|
|
|
int poolSize = Math.max(1, harProperties.getCallbackPoolSize());
|
|
int poolSize = Math.max(1, harProperties.getCallbackPoolSize());
|
|
|
this.callbackPool = Executors.newFixedThreadPool(poolSize, r -> {
|
|
this.callbackPool = Executors.newFixedThreadPool(poolSize, r -> {
|
|
@@ -106,7 +112,9 @@ public class HarCallbackService {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 异步回调企业商户。方法立即返回,不阻塞上游通知的响应。
|
|
|
|
|
|
|
+ * 回调企业商户。有地址走异步推送(方法立即返回,不阻塞上游通知的响应);
|
|
|
|
|
+ * 没地址**记 FAILED 落库**(同步写,不占回调线程),让 {@link #retryFailed}
|
|
|
|
|
+ * 将来还能捞起它 —— 商户补配地址后照样能收到这一笔。
|
|
|
*/
|
|
*/
|
|
|
public void notifyEnterprise(HarOrderEntity order) {
|
|
public void notifyEnterprise(HarOrderEntity order) {
|
|
|
if (order == null) {
|
|
if (order == null) {
|
|
@@ -114,13 +122,14 @@ public class HarCallbackService {
|
|
|
}
|
|
}
|
|
|
String returnUrl = resolveReturnUrl(order.getTenantId(), order.getApiKey());
|
|
String returnUrl = resolveReturnUrl(order.getTenantId(), order.getApiKey());
|
|
|
if (StrUtil.isBlank(returnUrl)) {
|
|
if (StrUtil.isBlank(returnUrl)) {
|
|
|
- log.info("[汇安融回调] 租户未配置回调地址,跳过: tenant_id={}, order_no={}",
|
|
|
|
|
|
|
+ log.warn("[汇安融回调] 租户未配置回调地址,记 FAILED 待重试: tenant_id={}, order_no={}",
|
|
|
order.getTenantId(), order.getOrderNo());
|
|
order.getTenantId(), order.getOrderNo());
|
|
|
|
|
+ writeNotifyResult(order, false);
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
String orderNo = order.getOrderNo();
|
|
String orderNo = order.getOrderNo();
|
|
|
- CompletableFuture.runAsync(() -> doNotify(order, returnUrl, orderNo), callbackPool)
|
|
|
|
|
|
|
+ CompletableFuture.runAsync(() -> writeNotifyResult(order, pushOnce(returnUrl, order)), callbackPool)
|
|
|
.exceptionally(ex -> {
|
|
.exceptionally(ex -> {
|
|
|
log.error("[汇安融回调] 异步任务异常: order_no={}, error={}",
|
|
log.error("[汇安融回调] 异步任务异常: order_no={}, error={}",
|
|
|
orderNo, ex.getMessage(), ex);
|
|
orderNo, ex.getMessage(), ex);
|
|
@@ -128,36 +137,37 @@ public class HarCallbackService {
|
|
|
});
|
|
});
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- private void doNotify(HarOrderEntity order, String returnUrl, String orderNo) {
|
|
|
|
|
- String notifyId = "n" + SnowflakeIdGenerator.nextIdStr();
|
|
|
|
|
- long timestamp = System.currentTimeMillis();
|
|
|
|
|
- String content = buildContent(order);
|
|
|
|
|
-
|
|
|
|
|
- boolean ok = post(returnUrl, notifyId, timestamp, content);
|
|
|
|
|
- writeNotifyStatus(orderNo, ok);
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 回写回调结果。
|
|
|
|
|
|
|
+ * 回写一次回调尝试的结果。
|
|
|
*
|
|
*
|
|
|
- * 重新查一次再更新,而不是直接 update 传入的实体 —— 传入的实体可能已被
|
|
|
|
|
- * 通知线程后续的状态更新覆盖,用乐观锁直接更新会因 version 过期而失败。
|
|
|
|
|
|
|
+ * 【为什么是 UpdateWrapper 而不是「先查再 updateById」】
|
|
|
|
|
+ * ① 乐观锁竞态:本方法可能由回调线程在 {@code syncFromUpstream} 事务提交**前**
|
|
|
|
|
+ * 执行(轮询腿触发回调),先查拿到的是提交前的旧 version —— 事务提交后
|
|
|
|
|
+ * updateById 的 {@code WHERE version=?} 被新版本挡掉,写库**静默丢**(0 行)。
|
|
|
|
|
+ * 免版本的 SET 更新直接排队等行锁,提交后落在新行上。
|
|
|
|
|
+ * ② notify_count 自增必须落在 SQL 侧:并发重试下 read-modify-write 会丢计数,
|
|
|
|
|
+ * 计数丢了次数上限就形同虚设。
|
|
|
|
|
+ * 代价是 updated_time 也要显式写({@code update(null, wrapper)} 不经自动填充,
|
|
|
|
|
+ * 同 {@code HarOrderService#doRefund} 先例)。
|
|
|
*/
|
|
*/
|
|
|
- private void writeNotifyStatus(String orderNo, boolean ok) {
|
|
|
|
|
|
|
+ private void writeNotifyResult(HarOrderEntity order, boolean ok) {
|
|
|
try {
|
|
try {
|
|
|
- HarOrderEntity latest = orderMapper.selectOne(new LambdaQueryWrapper<HarOrderEntity>()
|
|
|
|
|
- .eq(HarOrderEntity::getOrderNo, orderNo)
|
|
|
|
|
- .last("LIMIT 1"));
|
|
|
|
|
- if (latest == null) {
|
|
|
|
|
|
|
+ OffsetDateTime now = OffsetDateTime.now();
|
|
|
|
|
+ UpdateWrapper<HarOrderEntity> w = new UpdateWrapper<>();
|
|
|
|
|
+ w.set("notify_status", ok ? HarNotifyStatus.SUCCESS : HarNotifyStatus.FAILED)
|
|
|
|
|
+ .setSql("notify_count = COALESCE(notify_count, 0) + 1")
|
|
|
|
|
+ .set("last_notify_time", now)
|
|
|
|
|
+ .set("updated_time", now)
|
|
|
|
|
+ .eq("order_no", order.getOrderNo());
|
|
|
|
|
+ if (orderMapper.update(null, w) == 0) {
|
|
|
|
|
+ log.warn("[汇安融回调] 状态写入未命中订单: order_no={}", order.getOrderNo());
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
|
- latest.setNotifyStatus(ok ? HarNotifyStatus.SUCCESS : HarNotifyStatus.FAILED);
|
|
|
|
|
- orderMapper.updateById(latest);
|
|
|
|
|
// 回调状态也是页面上的一格(HAR 详情弹窗),成功失败都要发。
|
|
// 回调状态也是页面上的一格(HAR 详情弹窗),成功失败都要发。
|
|
|
- // 本方法 private + 跑在回调线程池里(Task 7 唯一没被单测钉住的写点,Task 14 E2E 覆盖)。
|
|
|
|
|
- publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, latest.getTenantId());
|
|
|
|
|
|
|
+ publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, order.getTenantId());
|
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
|
- log.warn("[汇安融回调] 回写 notify_status 失败: order_no={}, error={}", orderNo, e.getMessage());
|
|
|
|
|
|
|
+ log.warn("[汇安融回调] 回写 notify_status 失败: order_no={}, error={}",
|
|
|
|
|
+ order.getOrderNo(), e.getMessage());
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -212,44 +222,64 @@ public class HarCallbackService {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- private boolean post(String url, String notifyId, long timestamp, String content) {
|
|
|
|
|
- for (int attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
|
|
|
|
|
- try {
|
|
|
|
|
- MultiValueMap<String, Object> formData = new LinkedMultiValueMap<>();
|
|
|
|
|
- formData.add("notify_id", notifyId);
|
|
|
|
|
- formData.add("timestamp", String.valueOf(timestamp));
|
|
|
|
|
- formData.add("content", content);
|
|
|
|
|
-
|
|
|
|
|
- HttpHeaders headers = new HttpHeaders();
|
|
|
|
|
- headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 单次推送 —— 线程内不再退避重试(重试见 {@link #retryFailed})。
|
|
|
|
|
+ *
|
|
|
|
|
+ * @return true = 商户侧返回 2xx;任何非 2xx / 异常都算一次失败尝试,记 FAILED 等调度器再试
|
|
|
|
|
+ */
|
|
|
|
|
+ private boolean pushOnce(String url, HarOrderEntity order) {
|
|
|
|
|
+ String notifyId = "n" + SnowflakeIdGenerator.nextIdStr();
|
|
|
|
|
+ MultiValueMap<String, Object> formData = new LinkedMultiValueMap<>();
|
|
|
|
|
+ formData.add("notify_id", notifyId);
|
|
|
|
|
+ formData.add("timestamp", String.valueOf(System.currentTimeMillis()));
|
|
|
|
|
+ formData.add("content", buildContent(order));
|
|
|
|
|
|
|
|
- ResponseEntity<String> response = restTemplate.postForEntity(
|
|
|
|
|
- url, new HttpEntity<>(formData, headers), String.class);
|
|
|
|
|
|
|
+ HttpHeaders headers = new HttpHeaders();
|
|
|
|
|
+ headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
|
|
|
|
|
+ try {
|
|
|
|
|
+ ResponseEntity<String> response = restTemplate.postForEntity(
|
|
|
|
|
+ url, new HttpEntity<>(formData, headers), String.class);
|
|
|
|
|
+ if (response.getStatusCode().is2xxSuccessful()) {
|
|
|
|
|
+ log.info("[汇安融回调] 成功: notify_id={}, url={}", notifyId, url);
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ log.warn("[汇安融回调] 返回非 2xx,记 FAILED 待调度器重试: notify_id={}, status={}, url={}",
|
|
|
|
|
+ notifyId, response.getStatusCode(), url);
|
|
|
|
|
+ return false;
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ log.warn("[汇安融回调] 推送异常,记 FAILED 待调度器重试: notify_id={}, url={}, error={}",
|
|
|
|
|
+ notifyId, url, e.getMessage());
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
- if (response.getStatusCode().is2xxSuccessful()) {
|
|
|
|
|
- log.info("[汇安融回调] 成功: notify_id={}, url={}", notifyId, url);
|
|
|
|
|
- return true;
|
|
|
|
|
- }
|
|
|
|
|
- // 4xx 是对方的问题,重试没有意义
|
|
|
|
|
- if (response.getStatusCode().is4xxClientError()) {
|
|
|
|
|
- log.warn("[汇安融回调] 返回 {},不再重试: notify_id={}",
|
|
|
|
|
- response.getStatusCode(), notifyId);
|
|
|
|
|
- return false;
|
|
|
|
|
- }
|
|
|
|
|
- log.warn("[汇安融回调] 第 {} 次返回 {}", attempt + 1, response.getStatusCode());
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 扫出推送失败且未达次数上限的订单再推一遍 —— 供轮询调度器周期调用。
|
|
|
|
|
+ *
|
|
|
|
|
+ * 对齐乐收赢 LsyCallbackService#retryFailed:同一查询形状(FAILED +
|
|
|
|
|
+ * count < 上限 + 按 last_notify_time 升序饿最久的先推 + LIMIT 50)。
|
|
|
|
|
+ * 扫描异常只记不抛:调用方是调度线程,上面还挂着轮询主任务。
|
|
|
|
|
+ */
|
|
|
|
|
+ public void retryFailed() {
|
|
|
|
|
+ List<HarOrderEntity> orders;
|
|
|
|
|
+ try {
|
|
|
|
|
+ orders = orderMapper.selectList(new LambdaQueryWrapper<HarOrderEntity>()
|
|
|
|
|
+ .eq(HarOrderEntity::getNotifyStatus, HarNotifyStatus.FAILED)
|
|
|
|
|
+ .lt(HarOrderEntity::getNotifyCount, harProperties.getCallbackMaxRetries())
|
|
|
|
|
+ .orderByAsc(HarOrderEntity::getLastNotifyTime)
|
|
|
|
|
+ .last("LIMIT 50"));
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ log.error("[汇安融回调] 扫描待重试订单失败: {}", e.getMessage(), e);
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ for (HarOrderEntity order : orders) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ notifyEnterprise(order);
|
|
|
} catch (Exception e) {
|
|
} catch (Exception e) {
|
|
|
- log.warn("[汇安融回调] 第 {} 次失败: {}", attempt + 1, e.getMessage());
|
|
|
|
|
- }
|
|
|
|
|
- if (attempt < MAX_ATTEMPTS - 1) {
|
|
|
|
|
- try {
|
|
|
|
|
- Thread.sleep(BACKOFF_SECONDS[attempt] * 1000L);
|
|
|
|
|
- } catch (InterruptedException ie) {
|
|
|
|
|
- Thread.currentThread().interrupt();
|
|
|
|
|
- return false;
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ // notifyEnterprise 各段自身都有保护,这里是防将来改动引入的抛出拖垮整批
|
|
|
|
|
+ log.error("[汇安融回调] 重试推送异常: order_no={}, error={}",
|
|
|
|
|
+ order.getOrderNo(), e.getMessage());
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
- log.error("[汇安融回调] {} 次尝试后仍失败: url={}", MAX_ATTEMPTS, url);
|
|
|
|
|
- return false;
|
|
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|