文章目录

• 前言
• 一、重试场景分析
• 一、如何实现重试
• • 1. 扫表
  • 2. 基于中间件自身特性
  • 3. 基于框架
  • 4. 根据公司业务特性自己实现的重试
• 二、重试组件封装
• • 1. 需求分析
  • 2. 模块设计
  • • 2.1 持久化模块
    • • 1. 表定义
      • 2. 持久化接口定义
      • 3. 持久化配置类
    • 2.2 重试模块
    • • 1.启动
      • 2.重试
  • 3. 业务端使用
  • • 1. 引入依赖
    • 2. 新增配置
    • 3. 使用
• 总结

前言

如何封装一套服务自身业务开箱即用的重试组件？是个值得思考的问题！

在开发支付系统过程中，我们经常会遇到这样的业务场景：调用下游系统、回调上游系统，由于网络原因或者当时对方系统不可用导致调用失败，那么调用失败就失败了么？当然肯定不是，一般都要有重试机制。这种重试机制实现有很多方式，但是万万不可依赖其他系统的重试机制去重试你要重试调用的系统，这个原因下面分析。本篇文章就重试场景给出一个个人觉得还不错的解决方案，也是作者所在用的解决方案，如有更好的解决方案欢迎交流。

一、重试场景分析

在支付系统中我们经常会将一些非核心业务流程做成异步的，在核心主流程中往MQ写入一条相对应的待处理消息，写入成功即认为业务处理成功了，所以我们要证在消费端最大程度的保证处理成功。
在结果通知中也有失败重试策略，我们对接支付渠道如支付宝：如果不返回指定成功的报文信息其将在25小时以内完成8次通知（通知的间隔频率一般是4m,10m,10m,1h,2h,6h,15）。
这里我们分析个场景，流程很简单，如下：
支付渠道通知我们的支付系统，支付系统通知商户系统，之间为同步调用，渠道调用过来，支付系统变更订单状态，变更后调用商户系统，如果调用商户系统失败了，那么支付系统给渠道返回失败，然后过一段时间后渠道发起重试，再次调用支付系统，支付系统再调用商户系统。借助渠道的通知重试策略来完成自身的重试通知。谁要是这么设计，原地刨个坑活埋了他吧，不要觉得没有人用这种方式，事实就是真的有公司这么用。结果可想而知，不出问题只能说明做的系统没交易量，一旦有交易量，支付系统会被商户系统给拖垮掉，原因自行分析。

本篇文章呢我们以支付结果通知为例作为场景展开分析，做一个面对这种场景的统一解决方案，同时是没有使用充值VIP的RabbitMQ作为消息中间件。

既然没钱充值VIP购买其强大的重试功能，只能自己开发了。

一、如何实现重试

1. 扫表

实现重试的方式有很多种，有基于扫描表的，如下：

前置通知失败后，即落入重试表，待定时任务触发扫描表重新发起调用，这种处理方案是很多公司在用的。这种方案虽然不会像上面有拖垮系统的风险，但是问题还是很多的，如定时任务多久触发一次？有些交易对实时性要求比较高，如果第一次因为网络原因导致的失败，紧接着重试一般就能成功了，那么就把定时任务设定1s一次的频率？这种方式不再详细分析了…有点设计能力的人都不会采用这种方式吧。

2. 基于中间件自身特性

RocketMQ中间件本身已经支持重试，下文直接截图了：

3. 基于框架

针对RabbitMQ中间件spring提供的retry：

server:port:8080
spring:rabbitmq:host: xxx.xxx.xxx.xxxport: 5672username: xxxxpassword: xxxpublisher-confirm-type: correlatedlistener:simple:acknowledge-mode: manualretry:enabled: truemax-attempts: 5initial-interval: 5000max-interval: 10000

4. 根据公司业务特性自己实现的重试

如上是自己基于“指数退避策略进行延迟重试”封装的一套重试组件，也是本篇要介绍的方案。

二、重试组件封装

1. 需求分析

如何封装一套服务自身业务开箱即用的重试组件？是个值得思考的问题，但是Spring-boot已经给出了答案。我们在使用Springboot开发项目时候想要集成RabbitMQ只需要加入依赖，然后配置yml就可以使用了，一旦满足约定好的条件，Springboot则帮我们激活所需要的Bean，那么我们是不是也可以参考其思想自己也装配重试所需的Bean。

   org.springframework.bootspring-boot-starter-amqp2.4.1

决定了怎么做，然后分析业务系统特性，自己做的支付系统业务特性是：一个系统会有多个队列的消费者，并且每个队列消息处理失败后的重试次数、间隔时间也各不相同，并且达到最大失败重试次数后要入通知重试表，供后期业务系统恢复后再次发起重试。最终要的是，使用系统只需要简单配置下就可以实现上面需求，就像spring提供的retry机制一样，简单配置下就行了，不需要你知道底层原理。

2. 模块设计

从我们的架构图中可以看到，其主要分为两个模块，重试模块、持久化模块，我们逐个分析这俩模块的设计实现，首先从简单的开始，持久化模块。

2.1 持久化模块

首先没得说需要建表，需要使用starter提供的自动持久化功能就要创建starter持久化所需要的表：

1. 表定义

/*** @author Kkk* @Description: 异常通知恢复表*/
@Entity
@Table(name = "notify_recover")
public class NotifyRecover implements Serializable {/**id*/@Id@Column(name="id",insertable = false)private Long id;/** 唯一标识键 */@Column(name="unique_key")private String uniqueKey ;/** 场景码 */@Column(name="scene_code")private String sceneCode ;/** 调用方系统 */@Column(name="system_id")private String systemId;/** 通知内容 */@Column(name="notify_content")private String notifyContent ;/** 通知方式:http mq */@Column(name="notify_type")private int notifyType ;/** 交换器*/@Column(name="exchange")private String exchange ;/** 异步通知路由键 */@Column(name="notify_key")private String notifyKey ;/** 通知次数 */@Column(name="notify_num")private int notifyNum ;/** 通知状态 */@Column(name="notify_status")private String notifyStatus ;/** 备注 */@Column(name="remark")private String remark ;/** 扩展字段 */@Column(name="extend")private String extend ;/** 创建时间 */@Column(name="create_time",insertable = false)private Date createTime ;/** 修改时间 */@Column(name="update_time",insertable = false)private Date updateTime ;@Column(name="bucket")private String bucket ;// ... ...
}

2. 持久化接口定义

然后入表接口肯定也是需要的：

/*** @author Kkk* @Description: 发送失败处理*/
public interface NotifyRecoverHandler {/*** 处理重发失败入重试表* @param t*/public void handlerSendFail(T t);
}

3. 持久化配置类

创建持久化配置类：

/*** @author Kkk* @Description: 持久化配置类*/
@Configuration
@ConditionalOnProperty(prefix = "spring.rabbitmq.retry",value = "recover",havingValue = "true",matchIfMissing = false)
public class JdbcHelperMqConfiguration {@Bean(name = "jdbcSelectProvider")public JdbcSelectProvider jdbcSelectProviderBean() {return new JdbcSelectProvider();}@Bean(name = "jdbcInsertProvider")public JdbcInsertProvider jdbcInsertProviderBean() {return new JdbcInsertProvider();}@Bean(name = "jdbcUpdateProvider")public JdbcUpdateProvider jdbcUpdateProviderBean() {return new JdbcUpdateProvider();}@Bean(name = "jdbcHelper")public JdbcHelper jdbcHelperBean(@Qualifier("jdbcSelectProvider")JdbcSelectProvider jdbcSelectProvider,@Qualifier("jdbcInsertProvider")JdbcInsertProvider jdbcInsertProvider,@Qualifier("jdbcUpdateProvider")JdbcUpdateProvider jdbcUpdateProvider) {return new JdbcHelperImpl(jdbcSelectProvider,jdbcInsertProvider,jdbcUpdateProvider);}@Bean(name = "notifyRecoverHandler")@ConditionalOnMissingBean(value = NotifyRecoverHandler.class)public NotifyRecoverHandler notifyRecoverHandlerBean(@Qualifier("jdbcHelper")JdbcHelper jdbcHelper) {return new DefaultNotifyRecoverHandlerImpl(jdbcHelper);}
}

此配置类的激活条件时，配置了失败是否需要入重试表配置。同时也可以不使用starter提供的入表策略，如果业务系统有自己的重试表那么就可以将失败的消息入到自定义的表中，此处预留的扩展点。
jdbcSelectProvider、jdbcInsertProvider、jdbcUpdateProvider这个三个类为查询、新增、更新对应的处理类，为底层的JDBC操作。

/*** @author Kkk* @Description: select提供类*/
public class JdbcSelectProvider {private static final Logger logger = LoggerFactory.getLogger(JdbcSelectProvider.class);@Resourceprivate DataSource dataSource;public JdbcSelectProvider() {}public List select(String sql, Class outputClass) {return this.selectExecute(sql,outputClass);}private List selectExecute(String sql, Class outputClass,Object... params) {Connection connection = null;PreparedStatement pst = null;ResultSet res = null;List ts =null;try {connection = DataSourceUtils.getConnection(this.dataSource);pst = connection.prepareStatement(sql);for(int i = 0; i < params.length; ++i) {pst.setObject(i + 1, params[i]);}res = pst.executeQuery();ts = mapRersultSetToObject(res, outputClass);} catch (SQLException var7) {var7.printStackTrace();}finally {try {connection.close();pst.close();} catch (SQLException throwables) {throwables.printStackTrace();}}return ts;}@SuppressWarnings("unchecked")public List mapRersultSetToObject(ResultSet rs, Class outputClass) {List outputList = null;try {if (rs != null) {if (outputClass.isAnnotationPresent(Entity.class)) {ResultSetMetaData rsmd = rs.getMetaData();Field[] fields = outputClass.getDeclaredFields();while (rs.next()) {T bean = (T) outputClass.newInstance();for (int _iterator = 0; _iterator < rsmd.getColumnCount(); _iterator++) {String columnName = rsmd.getColumnName(_iterator + 1);Object columnValue = rs.getObject(_iterator + 1);for (Field field : fields) {if (field.isAnnotationPresent(Column.class)) {Column column = field.getAnnotation(Column.class);if (column.name().equalsIgnoreCase(columnName) && columnValue != null) {BeanUtils.setProperty(bean, field.getName(), columnValue);break;}}}}if (outputList == null) {outputList = new ArrayList();}outputList.add(bean);}} else {logger.error("查询结果集映射失败，映射类需要@Entity注解");}} else {return null;}} catch (Exception e) {logger.error("查询结果集映射失败",e);}return outputList;}
}

jdbcHelper对如上几个Provider进行了统一包装处理：

/*** @author Kkk* @Description:*/
public class JdbcHelperImpl implements JdbcHelper {private Logger logger = LoggerFactory.getLogger(JdbcHelperImpl.class);String s="'";private JdbcSelectProvider jdbcSelectProvider;private JdbcInsertProvider jdbcInsertProvider;private JdbcUpdateProvider jdbcUpdateProvider;ResultSetMapper resultSetMapper = new ResultSetMapper();public JdbcHelperImpl(JdbcSelectProvider jdbcSelectProvider, JdbcInsertProvider jdbcInsertProvider,JdbcUpdateProvider jdbcUpdateProvider) {this.jdbcSelectProvider = jdbcSelectProvider;this.jdbcInsertProvider = jdbcInsertProvider;this.jdbcUpdateProvider = jdbcUpdateProvider;}public List  selectData(String uniqueKey,String sceneCode){StringBuilder stringBuilder = new StringBuilder("SELECT * FROM notify_recover WHERE unique_key='");stringBuilder.append(uniqueKey);stringBuilder.append(s);stringBuilder.append(" AND scene_code='");stringBuilder.append(sceneCode);stringBuilder.append(s);String sql = stringBuilder.toString();List pojoList = this.jdbcSelectProvider.select(sql, NotifyRecover.class);if(null==pojoList || pojoList.size()==0 ){logger.info("根据uniqueKey（{}）,sceneCode({})查询结果为空!",uniqueKey,sceneCode);return null;}return pojoList;}@Overridepublic void insertData(NotifyRecover notifyRecover) {jdbcInsertProvider.insert(notifyRecover);}@Overridepublic int updateData(NotifyRecover notifyRecover) {StringBuilder stringBuilder = new StringBuilder("UPDATE notify_recover SET notify_status='");stringBuilder.append(notifyRecover.getNotifyStatus());stringBuilder.append("', notify_num=");stringBuilder.append(notifyRecover.getNotifyNum());stringBuilder.append(" WHERE unique_key='");stringBuilder.append(notifyRecover.getUniqueKey());stringBuilder.append(s);stringBuilder.append(" AND scene_code='");stringBuilder.append(notifyRecover.getSceneCode());stringBuilder.append(s);String sql = stringBuilder.toString();int resultSet = this.jdbcUpdateProvider.update(sql);return resultSet;}
}

最后一部分持久化接口默认实现，如果业务方想使用持久化进制，并没有实现持久化接口则采用默认实现：

    @Bean(name = "notifyRecoverHandler")@ConditionalOnMissingBean(value = NotifyRecoverHandler.class)public NotifyRecoverHandler notifyRecoverHandlerBean(@Qualifier("jdbcHelper")JdbcHelper jdbcHelper) {return new DefaultNotifyRecoverHandlerImpl(jdbcHelper);}

持久化默认实现：

/*** @author Kkk* @Description: 持久化默认实现*/
public class DefaultNotifyRecoverHandlerImpl implements NotifyRecoverHandler {private Logger logger = LoggerFactory.getLogger(DefaultNotifyRecoverHandlerImpl.class);BasicThreadFactory factory = new BasicThreadFactory.Builder().namingPattern("recover-execute-thread-%d").uncaughtExceptionHandler(new NotifyRecoverThreadUncaughtExceptionHandler()).build();private ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(4,factory);private JdbcHelper jdbcHelperImpl;public DefaultNotifyRecoverHandlerImpl(JdbcHelper jdbcHelperImpl) {this.jdbcHelperImpl = jdbcHelperImpl;}@Overridepublic void handlerSendFail(NotifyRecover notifyRecover) {executor.execute(new Runnable() {@Overridepublic void run() {//采用异步持久化}});}
}

到这里就完成了持久化工作了，但是还有一个很重要的问题，怎么将此类注册为Spring中的Bean呢？方式多种，最简单的是使用@Import标签，在重试的主配置类上引入此配置类。

@Import(JdbcHelperMqConfiguration.class)
public class RabbitMqRetrySendConfigurationMultiply {
}

2.2 重试模块

下面分析重试模块，首先重试模块我们是基于RabbitMQ死信队列来做的，关于死信、死信队列的概念这里不做解释了，
重试最主要分为启动时、运行时两部分。

1.启动

根据配置自动生成死信队列并通过对应的交换器与原队列进行路由绑定，大概流程见很久之前写的一篇博客[商户交易结果通知设计]，当时只是针对支付系统通知功能做的，并没有做什么组件化，后期发现实际
项目中很多场景都需要这种重试机制，所以为了避免重复代码的编写，后期就简单的封装了下作为一个延迟重试组件以供在项目中开发作为一个组件直接引入依赖使用就行了。
要做的是如何将原来的代码片段封装到starter并装配到Spring中。

@Configuration
@EnableConfigurationProperties({RabbitMqRetryMultiplyProperties.class, SystenEnvProperties.class})
@ConditionalOnProperty(prefix = "spring.rabbitmq",value = "isRetry",havingValue = "true")
@ConditionalOnClass({ AmqpAdmin.class, RabbitTemplate.class })
@Import(JdbcHelperMqConfiguration.class)
public class RabbitMqRetrySendConfigurationMultiply {@Autowiredprivate RabbitMqRetryMultiplyProperties rabbitMqRetryMultiplyProperties;@Autowiredprivate SystenEnvProperties systenEnvProperties;@Bean(name = "rabbitMqService")public RabbitMqService rabbitMqServiceBean() {return new RabbitMqServiceImpl();}@Bean(initMethod = "start", destroyMethod = "stop")public PscCommonRetryQueueManager pscCommonRetryQueueManager(@Qualifier("rabbitMqService")RabbitMqService rabbitMqService,@Autowired(required = false) @Qualifier("notifyRecoverHandler")NotifyRecoverHandler notifyRecoverHandler) {return PscCommonRetryQueueManager.builder().configs(rabbitMqRetryMultiplyProperties.getConfigs()).retryCountFlag(SystemConstant.RETRY_COUNT_FLAG).rabbitMqService(rabbitMqService).notifyRecoverHandler(notifyRecoverHandler).applicationName(systenEnvProperties.getName()).build();}
}

即满足如下两个条件即会构建PscCommonRetryQueueManager这个Bean。

@ConditionalOnProperty
@ConditionalOnClass

初始化时候会调用其start方法，在看之前先看下配置类，需要用户配置什么东西。

/*** @author Kkk* @Description: 重试配置类*/
@Data
public class ConfigEntity implements Serializable {//重试次数private Integer retry_count=5;//重试队列名private String retry_queue_name_prefix;//死信消息失效时间计算方式：指数方式 exponentialprivate String message_expiration_type="exponential";//x-dead-letter-exchangeprivate String x_dead_letter_exchange;//x-dead-letter-exchangeprivate String x_dead_letter_routing_key;//延迟时间因子：10s。具体延迟时间计算方式：2^count*10spublic Integer delay_milliseconds=10000;//项目需要消费的队列名称public String consumer_queue_name;//消息丢失处理策略public String notify_recover_handler;
}

接下来看其start方法做了什么，首先看下类继承关系

在接口中定义方法。

/*** @author Kkk* @Description: 重试管理接口*/
public interface RetryQueueManager {/*** 启动*/void start();/*** 停止*/void stop();/*** 发送延迟消息 -可捕获异常入重试表*/boolean sendRetryMessage(Message message);/***发送消息 -可捕获异常入重试表*/boolean sendMessage(String exchange, String routingKey, String jsonString,String uniqueKey,String sceneCode);/*** 发送消息 -可捕获异常入重试表*/boolean sendMessage(String exchange, String routingKey, String jsonString);/*** 发送延迟消息-发送网络异常可以放入重试表*/boolean sendRetryMessage(Message message,String uniqueKey,String sceneCode);
}

抽象层抽取了写公共参数，具体实现由子类实现。

/*** @author Kkk* @Description: 抽象层*/
public abstract class AbstractRetryQueueManager implements RetryQueueManager {private Logger logger = LoggerFactory.getLogger(AbstractRetryQueueManager.class);// 重试处理protected NotifyRecoverHandler notifyRecoverHandler;// 消息处理protected RabbitMqService rabbitMqService;//消息重试次数标识 埋点到消息头中的字段public String retryCountFlag;//应用名称public String applicationName;//重试配置相关信息public List  retryQueueConfigs;@Datapublic static final class RetryQueueConfigs {//重试次数public Integer retryCount=10;//重试队列名public String retryQueueNamePrefix;//死信消息失效时间计算方式：指数方式 exponentialpublic String messageExpirationType="exponential";//x-dead-letter-exchangepublic String xDeadLetterExchange="topic";//x-dead-letter-routing-keypublic String xDeadLetterRoutingKey;//延迟时间因子：10s。具体延迟时间计算方式：2^count*10spublic Integer delayMilliseconds;//项目需要消费的队列名称public String consumerQueueName;}@Overridepublic void start() {logger.info("开始创建重试队列！");createRetryQueue();logger.info("创建重试队列完成！");}/*** 应用启动构建重试队列*/protected abstract void createRetryQueue();@Overridepublic void stop() {}// ... ...
}

在子类实现抽象层方法createRetryQueue()，生成死信交换器和队列并绑定，接着根据配置生成指定个说的死信队列，默认按照指数类型（延迟时间因子：10s。具体延迟时间计算方式：2^count*10s），然后将这些队列绑定到上面生成的交换器上，由于这些生成的死信队列没有消费者，所以消息过期后会再被路由到原队列中，即可又被正常消费处理，以此来达到延迟的效果，原理比较简单。

@Override
protected void createRetryQueue() {for (RetryQueueConfigs config:retryQueueConfigs) {TopicExchange topicExchange = ExchangeBuilder.topicExchange(config.getXDeadLetterExchange()).build();rabbitAdmin.declareExchange(topicExchange);Queue queue1 = QueueBuilder.durable(config.getConsumerQueueName()).build();rabbitAdmin.declareQueue(queue1);Binding binding = BindingBuilder.bind(queue1).to(topicExchange).with(config.getXDeadLetterRoutingKey());rabbitAdmin.declareBinding(binding);if(ExpirationTypeEnum.EXPONENTIAL.getCode().equals(config.getMessageExpirationType())){logger.info("申明“指数型”重试队列开始...");for (int i = 0; i < config.getRetryCount(); i++) {String queueName = null;try {Map args = new HashMap();//指定当成为死信时，重定向到args.put("x-dead-letter-exchange", config.getXDeadLetterExchange());args.put("x-dead-letter-routing-key", config.getXDeadLetterRoutingKey());String expiration = String.valueOf(Double.valueOf(Math.pow(2, i)).intValue()*config.getDelayMilliseconds());queueName = config.getRetryQueueNamePrefix() + "." + expiration;//声明重试队列，将参数带入Queue queue = QueueBuilder.durable(queueName).withArguments(args).build();rabbitAdmin.declareQueue(queue);logger.info("申明“指数型”重试队列成功[queueName:{}]", queueName);}catch (Throwable e){logger.error("申明“指数型”重试队列失败[i:{}, queueName:{}, e.message：{}]，异常:", i, queueName, e.getMessage(), e);}}logger.info("申明“指数型”重试队列结束...");}}
}

2.重试

判断重试次数，消费端获取到消息后，根据消息头埋点可以获到重试次数，重试次数超过最大次数则入重试表，待后期分析处理。

/*** 判断是否超过重试次数*/
public RetryEntity isOutOfRetryCount(Message message){int messageRetryCount = getMessageRetryCount(message);RetryQueueConfigs config = getRetryConfigByOriQueue(message);boolean result=messageRetryCount>(null==config?0:config.getRetryCount())?false:true;if(!result){logger.info("超过最大重试次数,入重试表!");//... ...}return new RetryEntity(result,messageRetryCount);
}/*** 获取重试次数*/
public int getMessageRetryCount(Message message){//初始为0int count = 0;Map headers = message.getMessageProperties().getHeaders();if(headers.containsKey(retryCountFlag)){count = NumberUtils.toInt((String) message.getMessageProperties().getHeaders().get(retryCountFlag), 0);}return count;
}

关于重试即消费端处理失败后进行重新投递，根据重试次数计算要投递的队列名称。

@Override
public boolean sendRetryMessage(Message message) {boolean result=true;try {//从消息题中获取到消息来源--队列名称，然后根据队列名称获取到配置中心此队列配置的相关信息RetryQueueConfigs retryConfigByOriQueue = getRetryConfigByOriQueue(message);//从消息头中获取到重试次数int retryCount = getMessageRetryCount(message);//根据配置中心配置的死信消息失效时间计算方式（默认指数方式），和重试次数计算出死信队列名称后缀String expiration = getRetryMessageExpiration(retryCount,retryConfigByOriQueue);logger.info("消息重发开始[expiration:{}, retryCount:{}]", expiration, retryCount);//获取死信队列名称String queueName = getRetryQueueName(expiration,retryConfigByOriQueue);logger.info("消息重发获取重试队列[expiration:{}, retryCount:{}, queueName:{}]", expiration, retryCount, queueName);//发送消息rabbitMqService.sendRetry("", queueName, message, expiration, retryCount,retryCountFlag);logger.info("消息重发结束[expiration:{}, retryCount:{}]", expiration, retryCount);} catch (Exception e) {logger.info("({})发送重试消息失败！", JSON.toJSONString(message),e);result=false;}return result;
}

3. 业务端使用

1. 引入依赖

  com.epaydelay-component-spring-boot-stater1.0.0-SNAPSHOT

2. 新增配置

3. 使用

总结

本篇简单的介绍了下在工作中，将RabbitMQ进行简单封装作为延时组件使用，在使用时只需要简单的进行配置下就可以达到延时效果，降低了重复代码的编写，大大缩短了项目开发周期，由于工期紧张封装的starter还是比较粗糙的，还有好多地方需要斟酌打磨。

本篇也只是提供一种思想吧，在工作中可以借鉴下，避免重复劳动，将业务功能组件化，以后不管在什么项目中只要有相同业务场景就可以引入现有组件快速完成业务功能开发。

拙技蒙斧正，不胜雀跃。