关于支付回调的一些思考
前段时间在工作中又一次接触到了公司关于支付相关的业务领域，于是又一次重新回顾了工作中使用的支付框架。 之前写过的很多篇文章都是先介绍整体背景，再深入到每一点去做分析，今天打算换种方式，尝试从 why what how 三个点来进行切入。
why--**为什么要有一套支付框架**
企业规模变大的之后，可能会有多套支付渠道，例如支付宝支付，微信支付，银联支付等等，不同的支付渠道下边又会有多种支付入口，例如：网站二维码支付，手机小程序内支付，短信跳转到支付宝App支付等等。支付框架的引入就是为了统一这些各式各样的支付渠道，避免重复造轮子的情况发生。
what--**什么是支付框架？**
按照统一的一套支付流程规范，将公用逻辑统一处理，专门解决支付相关的程序代码。
其中包括：
- 支付密钥的统一
- 管理支付回调信息的统一配置
- 支付重试机制
- 支付流水记录
- 支付渠道管理
等等。
how--如何设计一款支付框架？
通过数据库表进行一些基本属性的配置，根据不同的支付配置id映射到对应的第三方对接程序，然后进行支付处理。好了大体的一个思路已经逐渐呈现，接下来便开始进入设计环节。常规的支付流程模式：
大部分的情况下，一次正常的支付流程是这种模式的。
至于第三方支付回调模块的处理，核心都是交给第三方进行管控，所以这一块对于研发人员来说大部分是处于一个黑盒的状态，我们更多的是关心在自己业务系统收到第三方支付回调之后应该做哪些操作。
所以包含了内部的业务系统调用流程之后，整个流程图基本如下所示：
而作为研发成员来说，我们更加会关注自己内部的业务系统回调模块，这个时候不妨可以将回调的步骤拆解为两步：
1.先支付第三方通知我们业务服务器
2.业务服务器自己进行内部的业务回调
这里面重点关注到第二步骤：
不妨将业务回调接口尝试通过配置化进行管理，这里给出了一种我自己的设计方案：首先设计一张支付配置表，这里以微信支付渠道作为案例讲解，这张表我们暂时称之为：t_wx_pay_config
CREATE TABLE `t_wx_pay_config` (
`id` int(6) NOT NULL AUTO_INCREMENT,
`pay_config_id` int(6) NOT NULL,
`appid` varchar(60) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT 'appid',
`secrity_key` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '支付密钥',
`third_notify_url` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '第三方回调地址',
`bussiness_notify_url` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '业务内部回调地址',
`des` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '备注说明',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
`creater_name` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin DEFAULT NULL COMMENT '创建者名字',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_pay_config_id` (`pay_config_id`) USING BTREE COMMENT '支付配置唯一id'
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
其中配置了第三方的支付回调接口地址和业务内部的接口回调地址信息。
如果要通过简单的代码逻辑来实现这种设计思路，大概设计为：
public String thirdNotify(HttpServletRequest req){
// 处理支付回调参数校验...
this.checkParam(req);
//处理自己业务系统内部的回调处理
this.notifyBusinessUrl(req);
return "success";
}具体配置如下方的一份测试数据所示：
如何处理回调部分的工作？直接在代码里面实现吗？
可以这么做，但是如果遇到了一些业务回调异常该如何处理呢？例如第三方回调业务A系统，业务A系统再去回调其他B系统的时候出现了错误，需要过一段时间B系统才能恢复正常，这个时候就需要A系统支持重试的功能。
这里我分享一种自己在工作中实际落地实现的技术方案供大家参考：
将第三方回调的信息和参数封装成一个对象，存入到一个队列中，接下来由专门的任务线程去消费这个队列，将队列中的任务提取出来。
任务的目的：携带参数请求指定的内部系统接口。
问题点：
如何指定请求的内部系统接口？
通过上边说的 t_wx_pay_config 表查询即可，在调用支付接口之前，通常都会允许发起方传入一些指定的参数信息，这个时候我们可以自定义传入一些相关的参数，一般都是传入订单OrderNo，Order相关的信息可以在发起支付确认订单的时候和一些相关支付配置做绑定（例如t_wx_pay_config表的pay_config_id），这样在回调的时候也可以根据订单编号找到对应的支付发起配置信息了。
如果直接在微信第三方回调的接口进行硬编码的同步调用，可能会出现调用超时，导致接口堵塞的情况，这种该如何处理？
可以尝试将回调的任务丢入一个队列中，然后专门由指定的线程去处理这些回调任务。
为了避免机器重发之后本地内存的丢失，所以队列的信息存储可以选择使用分布式缓存redis来实现。
如果回调失败如何处理？直接重新丢入队列进行回调重试吗？
一般出现回调失败的原因大部分都是因为系统存在一定的bug因素导致的，需要人工介入手工修复的机率较大，所以回调的间隔一般不会是连续性的立即回调，而是呈现一个阶段性的回调处理，例如分别间隔3秒，30秒，1min，3min，9min，30min，60min...的间隔进行回调尝试，如果经过多次回调都始终处于失败状态，则需要进行专门的记录。
回调队列的任务获取到之后如何判断是否需要执行任务？因为有可能是回调失败，重试的任务，并不需要那么快就进行回调重试。
这里可以利用到一个redis的zset数据结构。zset结构中基本存储信息和set结构类似，但是会给每个元素都有一个排分的score标示，根据排分的大小会在内存中进行排序操作。所以可以将时间戳作为排分的计数项，时间戳靠前的排在zset的前头，那么在进行回调任务处理的时候就会被优先读取出来。
从回调队列取出的任务，判断当前的时间戳是否小于或等于预期的执行时间，是的话就可以进行业务回调了。
回调队列里的元素什么时候进行删除？
回调队列读出首项任务，并且进行准备回调接口的过程中，如果机器发生了重启，那么这项回调任务可能就直接消失不见了，所以回调任务应该在回调成功之后再去进行删除。但是如果回调过程中如果网络通信较慢，可能会导致回调队列的消费效率较低，容易出现堵塞的情况发生。
所以这里我引入了一个就绪队列的设计
当回调队列中的任务的执行时间达到预期的时候，就将其放入到就绪队列中，这份转移的工作交给了专门的就绪线程去处理。并且一旦转移到就绪队列之后，便将其从回调队列移除。
就绪队列是采用了redis的queue队列进行存储，满足了先进先出的优先级设计需求。
如果支付环节中出现重试次数超过预期上线，该如何记录？
将这类信息专门记录到一张map中，并且通知到指定负责人进行人工介入即可。
大体的一些设计难点和可能遇到的问题罗列了出来，这里我给出一些关键点的实现代码供大家参考：
延迟任务基本模型：
package com.qiyu.tech.pay.queue.model;
import java.io.Serializable;
/**
* 延迟任务信息
*
* @Author idea
* @Date created in 8:41 上午 2021/4/1
*/
public class DelayTaskInfo implements Serializable {
private static final long serialVersionUID = -1003963035927679825L;
/**
* 唯一标识
*/
private long taskId;
/**
* 支付配置id
*/
private int payConfigId;
/**
* 任务类型
*/
private String topic;
/**
* 任务执行时间
*/
private Long executeTime;
/**
* 任务执行的超时时长限制
*/
private Long executeTimeout;
/**
* 任务的重试次数
*/
private int retryTimes;
/**
* 任务的消息内容
*/
private String message;
//getter setter省略
}每次进行回调任务的时候，不论是处于回调队列还是就绪队列，都需要在这张Map中做一次回调信息的记录，用于后续的一些基本参数核对。防止如果有人恶意修改队列内部的属性，可能会造成错误回调，所以多加一份map做校验。
/**
* 支付的每一次任务都会被记录到这张map中
*
* @Author idea
* @Date created in 9:08 上午 2021/4/2
*/
@Component
public class DelayTaskInfoManager {
@Resource
private RedisService redisService;
private static final String DELAY_TASK_INFO_MAP_KEY = "delay:task:info:map";
public boolean put(long taskId, DelayTaskInfo delayTaskInfo) {
return redisService.setMapItem(DELAY_TASK_INFO_MAP_KEY, String.valueOf(taskId), JSON.toJSONString(delayTaskInfo));
}
public DelayTaskInfo get(long taskId) {
String json = (String) redisService.getMapField(DELAY_TASK_INFO_MAP_KEY, String.valueOf(taskId), String.class);
DelayTaskInfo delayTaskInfo = JSON.parseObject(json, DelayTaskInfo.class);
return delayTaskInfo;
}
/**
* 完成支付任务
*/
public void finishPayNotifyJob(DelayTaskInfo delayTaskInfo) {
redisService.deleteMapItem(DELAY_TASK_INFO_MAP_KEY, String.valueOf(delayTaskInfo.getTaskId()));
}
}基于redis的队列封装与实现:
接口抽象
/**
* @Author idea
* @Date created in 8:50 上午 2021/3/30
*/
public interface RedisQueue<T> {
/**
* 弹出任务
*
* @return
*/
T pop();
/**
* 放入任务
*
* @param t
* @return
*/
boolean put(T t);
/**
* 删除队列中的一项内容
*
* @return
*/
boolean delete(T item);
}回调队列实现设计，本质上一条延迟队列
/**
* @Author idea
* @Date created in 9:01 上午 2021/3/30
*/
@Component("redisDelayQueue")
@Slf4j
public class RedisDelayQueue implements RedisQueue<DelayTaskSortItem> {
@Resource
private RedisService redisService;
private static String redisDelayQueueKey = "redis:delay:pay:notify";
private static final Object mutex = new Object();
@Override
public DelayTaskSortItem pop() {
synchronized (mutex) {
Set<Tuple> rangeSet = redisService.zRangeWithScores(redisDelayQueueKey, 0, 1);
if (CollectionUtils.isEmpty(rangeSet)) {
return null;
}
for (Tuple tuple : rangeSet) {
long currentTimeMillis = System.currentTimeMillis();
double expireTime = tuple.getScore();
if ((currentTimeMillis - expireTime) > 0) {
DelayTaskSortItem delayTaskInfo = JSON.parseObject(tuple.getElement(), DelayTaskSortItem.class);
log.info("[delayTaskInfo] info is : {}", delayTaskInfo);
return delayTaskInfo;
}
}
}
return null;
}
@Override
public boolean put(DelayTaskSortItem delayTaskSortItem) {
redisService.zAdd(redisDelayQueueKey, delayTaskSortItem.getExecuteTime(), JSON.toJSONString(delayTaskSortItem));
return true;
}
@Override
public boolean delete(DelayTaskSortItem delayTaskSortItem) {
redisService.zRem(redisDelayQueueKey, Arrays.asList(String.valueOf(JSON.toJSONString(delayTaskSortItem))));
return true;
}
}
就绪队列的设计
/**
* @Author idea
* @Date created in 8:46 上午 2021/4/2
*/
@Component("redisReadyQueue")
public class RedisReadyQueue implements RedisQueue<DelayTaskInfo> {
@Resource
private RedisService redisService;
private static final String REDIS_READY_QUEUE_KEY = "redis:ready:queue:pay";
@Override
public DelayTaskInfo pop() {
String item = redisService.rpop(REDIS_READY_QUEUE_KEY);
if(item != null) {
return JSON.parseObject(item,DelayTaskInfo.class);
}
return null;
}
@Override
public boolean put(DelayTaskInfo delayTaskInfo) {
Long id = -1L;
id = redisService.lpush(REDIS_READY_QUEUE_KEY, JSON.toJSONString(delayTaskInfo));
return id > 0;
}
@Override
public boolean delete(DelayTaskInfo delayTaskInfo) {
return true;
}
}微信第三方的接口回调，也就是表中配置的 third_notify_url 地址信息
@GetMapping(value = "wx-notify-url")
public String wxNotifyUrl(HttpServletRequest request){
log.info("get wx request success");
String orderNo = request.getParameter("orderNo");
String lockKey = "pay:notify:wx:lock" + orderNo;
try {
boolean lockStatus = iDistributionLock.tryLock(lockKey);
if(!lockStatus) {
throw new RuntimeException("try lock fail,key is " + lockKey);
}
OrderInfoPO orderInfoPO = orderInfoDao.getOrderInfoPO(orderNo);
if(orderInfoPO == null){
return "orderInfo not exist!";
}
DelayTaskInfo delayTaskInfo = new DelayTaskInfo();
delayTaskInfo.setPayConfigId(orderInfoPO.getPayConfigId());
delayTaskInfo.setTaskId(unionIdService.getUnionId());
delayTaskInfo.setExecuteTime(System.currentTimeMillis());
delayTaskInfo.setExecuteTimeout(System.currentTimeMillis() + DEFAULT_TIME_OUT);
delayTaskInfo.setMessage(JSON.toJSONString(orderInfoPO));
delayTaskInfo.setTopic(PAY_TOPIC);
delayTaskInfo.setRetryTimes(0);
DelayTaskSortItem delayTaskSortItem = new DelayTaskSortItem(delayTaskInfo.getTaskId(),PAY_TOPIC,delayTaskInfo.getExecuteTime());
//存放一些回调的信息到map中做记录
delayTaskInfoManager.put(delayTaskInfo.getTaskId(),delayTaskInfo);
//将回调任务存入到回调队列中
redisDelayQueue.put(delayTaskSortItem);
return "wx-notify-success";
}catch (Exception e){
log.error("[wx-notify-url] has unKnow error , e is {}",e);
}finally {
iDistributionLock.releaseLock(lockKey);
}
return "unKnow Error";
}
将回调队列中的元素转换到就绪队列的专门处理线程：
/**
* @Author idea
* @Date created in 8:47 上午 2021/4/1
*/
@Component
@Slf4j
public class RedisDelayQueueHandler implements Runnable {
@Resource(name = "redisDelayQueue")
private RedisQueue<DelayTaskSortItem> redisDelayQueue;
@Resource(name = "redisReadyQueue")
private RedisQueue<DelayTaskInfo> redisReadyQueue;
@Resource
private IDistributionLock distributionLock;
@Resource
private DelayTaskInfoManager delayTaskInfoManager;
@SneakyThrows
@Override
public void run() {
while (true) {
try {
DelayTaskSortItem delayTaskSortItem= redisDelayQueue.pop();
if (delayTaskSortItem == null) {
continue;
}
String lockKey = "pay:notify:dequeue:lock" + delayTaskSortItem.getTaskId();
try {
boolean lockStatus = distributionLock.tryLockTimeOut(lockKey, 3000);
if (!lockStatus) {
continue;
}
if(delayTaskSortItem.getExecuteTime() > System.currentTimeMillis()){
continue;
}
DelayTaskInfo delayTaskInfo = delayTaskInfoManager.get(delayTaskSortItem.getTaskId());
if(delayTaskInfo == null) {
log.error("队列任务有误！taskId is {}",delayTaskSortItem.getTaskId());
continue;
}
if(delayTaskInfo.getExecuteTime() > System.currentTimeMillis()) {
continue;
}
//推入任务到达就绪队列之前需要检查是否符合条件：执行任务时间，任务池和延迟队列任务是否一致，任务id是否一致
log.info("推入任务："+delayTaskInfo);
redisReadyQueue.put(delayTaskInfo);
redisDelayQueue.delete(delayTaskSortItem);
delayTaskInfoManager.finishPayNotifyJob(delayTaskInfo);
}catch (Exception e){
log.info("支付的延迟队列出现异常情况1：", e);
}finally {
distributionLock.releaseLock(lockKey);
}
} catch (Exception e) {
log.info("支付的延迟队列出现异常情况2：", e);
}
}
}
}
专门处理就绪队列中的线程：
/**
* 专门用于处理就绪队列内部的任务信息
*
* @Author idea
* @Date created in 11:35 上午 2021/4/3
*/
@Component
@Slf4j
public class RedisReadyQueueHandler implements Runnable {
@Resource
private RedisReadyQueue redisReadyQueue;
@Resource
private RedisDelayQueue redisDelayQueue;
@Resource
private DelayTaskInfoManager delayTaskInfoManager;
@Resource
private WXPayConfigDao wxPayConfigDao;
@Resource
private NotifyErrorTaskInfoManager notifyErrorTaskInfoManager;
@Resource
private NotifyRetryTaskInfoManager notifyRetryTaskInfoManager;
@SneakyThrows
@Override
public void run() {
while (true) {
DelayTaskInfo delayTaskInfo = redisReadyQueue.pop();
if (delayTaskInfo == null) {
continue;
}
log.info("弹出任务：delayTaskInfo is {}",delayTaskInfo);
ResponseEntity<String> responseEntity = null;
try {
WXPayConfigPO wxPayConfigPO = wxPayConfigDao.getByPayConfigId(delayTaskInfo.getPayConfigId());
responseEntity = this.doBusinessNotify(delayTaskInfo,wxPayConfigPO);
if (HttpStatus.OK.equals(responseEntity.getStatusCode()) && "success".equals(responseEntity.getBody())) {
log.info("支付回调成功！====== ");
//删除失败任务map中的记录
notifyErrorTaskInfoManager.delete(delayTaskInfo.getTaskId());
notifyRetryTaskInfoManager.delete(delayTaskInfo.getTaskId());
} else {
throw new RuntimeException("支付业务回调失败");
}
}catch (Exception e){
log.error("支付业务回调失败===== responseEntity is {}", responseEntity);
retryAgain(delayTaskInfo);
}
}
}
/**
* 业务内部系统回调处理
*
* @param delayTaskInfo
* @param wxPayConfigPO
* @return
*/
public ResponseEntity<String> doBusinessNotify(DelayTaskInfo delayTaskInfo,WXPayConfigPO wxPayConfigPO){
RestTemplate restTemplate = new RestTemplate();
HttpHeaders headers = new HttpHeaders();
headers.add("pay-token", createPayNotifySeq());
//业务回调参数
String message = delayTaskInfo.getMessage();
MultiValueMap<String, String> paramMap = new LinkedMultiValueMap<>();
paramMap.add("extInfo", message);
HttpEntity<MultiValueMap<String, String>> request = new HttpEntity<>(paramMap, headers);
log.info("wxPayConfigPO is {},paramMap is {},headers is {}", wxPayConfigPO, paramMap, headers);
ResponseEntity<String> responseEntity = restTemplate.postForEntity(wxPayConfigPO.getBussinessNotifyUrl(), request, String.class);
return responseEntity;
}
/**
* 创建业务方回调使用的seq序列号
*
* @return
*/
public String createPayNotifySeq(){
return UUID.randomUUID().toString();
}
/**
* 进入重试阶段
*
* @param delayTaskInfo
*/
public void retryAgain(DelayTaskInfo delayTaskInfo){
log.info("支付重试：delayTaskInfo is :{}",delayTaskInfo);
//进入支付重试环节
int retryTime = delayTaskInfo.getRetryTimes();
if(retryTime >= RETRY_TIMES.length) {
notifyRetryTaskInfoManager.delete(delayTaskInfo.getTaskId());
//超过重试还是异常的信息存在这张map中
notifyErrorTaskInfoManager.put(delayTaskInfo.getTaskId(),delayTaskInfo);
return;
}
delayTaskInfo.setExecuteTime(System.currentTimeMillis() + RETRY_TIMES[retryTime] * 1000 );
delayTaskInfo.setExecuteTimeout(delayTaskInfo.getExecuteTimeout() + DEFAULT_TIME_OUT);
delayTaskInfo.setRetryTimes(retryTime+1);
DelayTaskSortItem delayTaskSortItem = new DelayTaskSortItem(delayTaskInfo.getTaskId(),PAY_TOPIC,delayTaskInfo.getExecuteTime());
delayTaskInfoManager.put(delayTaskInfo.getTaskId(),delayTaskInfo);
redisDelayQueue.put(delayTaskSortItem);
//将重试的信息放入到一张map里面做管理
notifyRetryTaskInfoManager.put(delayTaskInfo.getTaskId(),delayTaskInfo);
}
}回调任务的失败管理器
/**
* 回调任务失败管理器
*
* @Author idea
* @Date created in 3:04 下午 2021/4/4
*/
@Component
public class NotifyErrorTaskInfoManager {
@Resource
private RedisService redisService;
private static final String NOTIFY_ERROR_TASK_INFO_MAP = "notify:error:task:info:map";
public boolean put(long taskId, DelayTaskInfo delayTaskInfo) {
return redisService.setMapItem(NOTIFY_ERROR_TASK_INFO_MAP, String.valueOf(taskId), JSON.toJSONString(delayTaskInfo));
}
public boolean delete(long taskId) {
return redisService.deleteMapItem(NOTIFY_ERROR_TASK_INFO_MAP, String.valueOf(taskId));
}
}
回调任务的重试管理器
/**
* @Author idea
* @Date created in 3:11 下午 2021/4/4
*/
@Component
public class NotifyRetryTaskInfoManager {
@Resource
private RedisService redisService;
private static final String NOTIFY_RETRY_TASK_INFO_MAP = "notify:retry:task:info:map";
public boolean put(long taskId, DelayTaskInfo delayTaskInfo) {
return redisService.setMapItem(NOTIFY_RETRY_TASK_INFO_MAP, String.valueOf(taskId), JSON.toJSONString(delayTaskInfo));
}
public boolean delete(long taskId) {
return redisService.deleteMapItem(NOTIFY_RETRY_TASK_INFO_MAP, String.valueOf(taskId));
}
}支付相关的常量配置
/**
* @Author idea
* @Date created in 10:35 上午 2021/4/4
*/
public class PayConfigConstants {
public static final String PAY_TOPIC = "微信支付默认回调处理";
/**
* 默认超时时间为3秒
*/
public static int DEFAULT_TIME_OUT = 3000;
/**
* 重试时间段
*/
public static int[] RETRY_TIMES = new int[]{3,9,60,180,360,540,1200,1800};
}
这里面我只是简单地将核心代码贴了出来，至于哪些简单的dao，redis封装工具比较简单，所以就没有做过多的展示。希望这些基本设计思路可以对大家有所帮助。
小结
关于这款支付回调组件的设计其实主要涉及的技术点是大家都比较熟悉的MySQL和Redis 以及一些熟悉的数据结构。
关于不同的第三方支付对接模块代码，这块我并没有体现，因为它主要是属于支付方通知第三方服务器的环节，根据选用不同的支付渠道选择不同的支付函数去触发进行设计即可。
例如：
涉及的几条队列包含有：
回调队列：存储第三方回调的支付任务以及处于回调重试的支付任务。
就绪队列：处理进过校验，确认需要立即进行回调处理的任务。
重试异常队列：回调过程中出现异常并且重试的超过一定次数的任务。
支付回调任务Map：记录处于回调处理过程中的任务数据。
最后贴出一张自己的测试redis截图：
