北京个人制作网站有哪些,大宗商品交易平台是什么,网站建设项目环境影响评价目录,江东seo做关键词优化一、背景
微服务设计中#xff0c;跨服务的调用#xff0c;由于网络或程序故障等各种原因#xff0c;经常会出现调用失败而需要重试。另外#xff0c;在异步操作中#xff0c;我们提供接口让外部服务回调。回调过程中#xff0c;也可能出现故障。
这就要求我们主动向外…一、背景
微服务设计中跨服务的调用由于网络或程序故障等各种原因经常会出现调用失败而需要重试。另外在异步操作中我们提供接口让外部服务回调。回调过程中也可能出现故障。
这就要求我们主动向外部服务发起查询以获取外部服务的操作结果。
见下图 这种异步操作流程比较常见于支付系统其思路值得借鉴。
本文主要是想介绍在没有订单号、支付流水号等业务唯一性的场景下使用一张任务表实现上面的异步操作达到最终一致性。
二、定时任务 1、定时查询 查询下一次重试时间早于当前时间的记录每次拉取N条。定时机制可以依赖xxl-job这样的分布式定时任务。 notifyTasksRepository.findTop50ByNextTimeBeforeOrderByNextTimeAsc(DateUtil.date());2、指数级重试 public void retry() {this.retryTimes 1;this.nextTime calcNextRetryTime();this.lastTime new Date();}private Date calcNextRetryTime() {final ZoneId zone ZoneId.systemDefault();final LocalDateTime lastTime LocalDateTime.ofInstant(Instant.now(), zone);final LocalDateTime nextTime lastTime.plus(exp(this.retryTimes), ChronoUnit.MILLIS);final Instant instant nextTime.atZone(zone).toInstant();return Date.from(instant);}private long exp(int retryCount) {long waitTime ((long) Math.pow(2, retryCount) * 1000L);return waitTime;}3、请求流水号
既然是微服务之间的异步操作流程就需要设计一个类似于订单号能够贯穿整个流程。它就是请求流水号要求是不能重复。发起请求、回调通知和主动查询都是需要传递该字段。
4、notifyParams参数
这是一个json格式的string字符串当外部服务是post请求的时候请求报文就保存在该字段。 而如果仅仅是保存http接口的请求入参的话在实际不同的业务还是不够的。 所以需要你把业务中需要透传的字段也都保存在该字段里。
比如http接口需要requestNo和classroomId两个字段就可以了。而你的业务系统里还需要字段课程编号和讲次编号那么请一并把它们都保存进去。 待外部服务回调的时候根据requestNo请求流水号查询任务需要反解析该字段取出需要的课程编号和讲次编号。 如此一来很好地体现了任务表的抽象性又满足了不同业务的具体性。 当然了前提是任务的查询不会把它们作查询条件。
JsonIgnoreProperties(ignoreUnknown true)
Data
Builder
NoArgsConstructor
AllArgsConstructor
public class ClassroomCopyRequest {/*** 原课堂ID(http请求必填参数)*/private String classroomId;/*** 请求流水号(http请求必填参数)*/private String requestNo;/*** 讲次编号透传参数*/private String lectureNo;/*** 课程编号透传参数*/private String courseNo;/*** 任务编号透传参数*/private String taskCode;
}三、代码设计 从上面的流程图也可以看出业务逻辑处理是有两个方向一是业务方的回调二是定时任务的主动查询。
二者最终都需要发送一个异步事件让异步线程池去处理。
所以无论从哪个链路获取到的返回结果都应该是一样的实际上也如此。 示例 复制课堂返回一个新的课堂ID。由于我们在不同的业务流程中都会有复制课堂的需求所以需要用任务编号加以区分。而外部服务是不会关注你的任务更遑论任务编号了。
在异步设计中尽量抽象系统之间的交互术语而不能具化。
进一步说复制课堂只需要知道原课堂ID即可不要去让它去关心是什么业务下的复制。 这就是请求流水号requestNo的巧妙设计之处。
ApiModel
Data
Builder
NoArgsConstructor
AllArgsConstructor
public class ClassroomCopyResultNotify {ApiModelProperty(notes 源课堂ID)private String sourceClassroomId;ApiModelProperty(notes 新课堂ID, required true)NotEmpty(message 新课堂ID必填)private String newClassroomId;ApiModelProperty(notes 请求流水号, required true)NotEmpty(message 请求流水号必填)private String requestNo;1、回调通知接口 有了上面的回调报文下面的接口设计就非常简单了。 ApiOperation(value 异步通知课堂的复制结果)PostMapping(value {/api/v1/classroom/copy/callback})public ResponseEntity? copyResultNotify(Validated RequestBody ClassroomCopyResultNotify notify) {return ResponseEntity.ok(classroomAppService.copyResultNotify(notify));}2、服务层ClassroomAppService.java
final NotifyTasks notifyTasks notifyTasksRepository.findByRequestNoAndTaskCode(request.getRequestNo(),Constants.TaskCode.CLASSROOM_COPY_RESULT);if (null notifyTasks) {if (log.isWarnEnabled()) {log.warn(请求流水号{}不存在, request.getRequestNo());}
}
//发送异步事件3、事件订阅者 SubscribeAllowConcurrentEventsTransactional(rollbackFor Throwable.class)public void onNotifySuccessEvent(NotifySuccessEvent event) {if (null event || null event.getTaskCode()) {return;}if (log.isInfoEnabled()) {log.info(订阅异步通知结果事件详情{}, JsonUtils.toJsonString(event));}switch (event.getTaskCode()) {case Constants.TaskCode.CLASSROOM_COPY_RESULT:// 解析任务的入参final ClassroomCopyRequest copyRequest JSON.parseObject(event.getRequestJson(), ClassroomCopyRequest.class);// 解析回调报文final ClassroomCopyResultNotify copyResponse JSON.parseObject(event.getResponseJson(), ClassroomCopyResultNotify.class);// 调用业务逻辑处理break;default:break;}}4、业务逻辑处理
经过上一步对参数和回调的解析接下来就是不同业务的处理了。
处理完成后删除任务即可。
// 业务处理
...
...
...
// 处理成功删除定时任务
notifyTasksRepository.deleteByRequestNoAndTaskCode(requestNo, Constants.TaskCode.CLASSROOM_COPY_RESULT);5、发起对外部服务的请求
ClassroomCopyRequest request ClassroomCopyRequest.builder().requestNo(requestNo).classroomId(sourceClassroomId).lectureNo(lectureNo).courseNo(courseNo).taskCode(taskCode).build();// notify方法发起Http请求如果失败则会像下面的语句一样保存至定时任务表
notifyTasksService.notify(taskCode, requestNo, url, gson.toJson(request));// 保存至定时任务表
notifyTasksService.schedule(Constants.TaskCode.CLASSROOM_COPY_RESULT, requestNo,queryUrl, gson.toJson(request));schedule()方法的定义见下
public void schedule(String taskCode, String requestNo, String notifyUrl, String notifyParams)这里的设计思路是发起外部服务的请求同时保存主动查询的任务到库里。外部服务处理完成回调我们的时候需要把主动查询的任务届时删除。防止不必要的查询主动查询复制结果的任务必须在发起请求的时候同时就保存到库表里。这是因为回调的报文中只有请求流水号requestNo在处理回调的时候需反查出本任务记录。
四、任务的增删查改
这一小节我们试着总结下对任务的操作。
1、新增
向外部服务发起请求的时候保存主动查询的任务(任务内容见notifyParams)
2、删除
业务逻辑处理完成后删除主动查询的任务。否则定时任务会一直轮询该任务浪费性能。
3、查询
分布式定时任务cron表达式规则下每次查询N条待处理的任务列表。
4、修改
主动查询结果的任务如果处理出现失败的时候则需要更新下一次重试时间支持指数级的退避重试。
五、总结
本文试着通过一个任务表向你表述如何实现重试补偿以实现跨服务间的最终一致性。
