如何实现异步通知的重试机制

528次阅读  |  发布于1年以前

工作中经常要和第三方做对接，比如支付、电子合同等系统。操作成功之后，第三方会发送异步的通知，返回最终的处理结果，使用异步而不是使用同步通知，是为了加快系统响应速度，防止线程阻塞。任务处理完成后通过异步的通知，发送给对应的服务端。之前对接微信支付，完成支付后，微信发送一个异步通知给服务端，服务端根据支付通知修改状态，通知规则看到以下的一段话。

其中有段话：

重新发送通知，直到成功为止（在通知一直不成功的情况下，微信总共会发起多次通知，通知频率为15s/15s/30s/3m/10m/20m/30m/30m/30m/60m/3h/3h/3h/6h/6h - 总计 24h4m）

微信为何要这么设计

微信结果通知本质就是发送一个网络请求到不同的服务器上，既然是一个网络请求，就可能因为各种原因导致请求超时或者失败，比如：

以上原因都会导致支付结果通知接收失败，也就无法通知给用户。为了解决上述的问题，就需要引入「重试机制」，当请求无法应答时，就需要重试几次，保证请求能确认发送。

异步通知的重试机制

从微信支付通知可以引申到所有的异步通知，或者和第三方对接时。如果要确保通知能被成功的接收，就需要考虑请求失败的情况，大部分都是需要使用重试机制。而重试机制是隔段时间不是固定的，是越来越大的，这是考虑到重试时，由于网络故障或者服务器故障重启设备需要花一段时间，而「间隔时间越来越长就可以更大的保证请求可以被成功接收」。

重复请求，接口需要考虑重复请求的情况，要设计成一个幂等性接口，多次请求和请求一次的效果是一致的。

重试机制的实现

重试机制就是一个定时器，隔一段时间执行一次，没有预期的效果就再重复执行一次。

实现的难点就在于，间隔的时间是不一致的，如果时间的间隔是固定的话，就可以使用定时任务。

方案一：定时任务（不可行）

使用定时器，每隔一段时间执行一次任务。在 SpringBoot 启动类添加 @EnableScheduling 注解，然后在执行的方法添加 @Scheduled 注解。

@Scheduled(fixedDelay = 1000*2)
public void test2() {
    Date date = new Date();
    System.out.println("tesk2 " + date);
}

以上表示每隔 2 秒执行一次。间隔时间都是固定的，这个不符合预期，因为要求的时间间隔是依次增加的。

❝如果是间隔时间是固定的，那定时任务就符合条件吗？ ❞

如果是只有一条任务在执行，执行不成功，存放在 Redis 中，然后定时执行任务，如果任务执行成功，就去掉任务。但是定时器还是会定时执行。

如果执行的任务很多的话，前面的任务要等待后续的任务执行，那延迟就很严重了，就需要使用到多线程，开启多个线程，在《阿里Java开发手册》有一条:

❝线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。 ❞

定时任务有以下几个缺点不满足：

方案二：线程池 + 定时任务 （不可行）

既然使用单线程会产生延迟，就使用线程池来降低延迟，因为发起请求属于 IO 密集型，所以线程数设置成 CPU 个数的两倍，在 SpringBoot 自定义一个线程池：

@Configuration
public class ThreadPoolConfig {

    // 线程存活时间
    private static int keepAliveTime = 10;

    // 调用线程运行多余任务
    RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();

    @Bean("customerTaskExecutor")
    public TaskExecutor taskExecutor() {
        // 核心线程数
        int cores = Runtime.getRuntime().availableProcessors()*2;
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(cores);
        executor.setMaxPoolSize(cores);
        executor.setKeepAliveSeconds(keepAliveTime);
        executor.setRejectedExecutionHandler(handler);
        executor.setThreadNamePrefix("Custom-");  // 线程名前缀
        executor.initialize();
        return executor;
    }
}

其中核心线程数和最大线程数设置成一致，拒绝策略使用调用线程运行多余的任务，确保每个任务都能执行。然后添加一个异步方法.

public interface AsyncService {

    void executeAsync();
}

@Service
@Slf4j
public class AsyncServiceImpl implements AsyncService {

    @Override
    @Async("customerTaskExecutor")
    public void executeAsync() {
        log.info("【开始执行任务】");
        // 延迟几秒
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("【结束执行任务】");

    }
}

使用 sleep 方法延迟，模拟请求，使用压测工具，发起 100 次请求，控制台输出如下：

2023-10-31 18:00:32.792  INFO 53009 --- [       Custom-1] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.811  INFO 53009 --- [       Custom-2] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.813  INFO 53009 --- [       Custom-3] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.814  INFO 53009 --- [       Custom-4] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.816  INFO 53009 --- [       Custom-5] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.817  INFO 53009 --- [       Custom-6] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.819  INFO 53009 --- [       Custom-7] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.820  INFO 53009 --- [       Custom-8] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.821  INFO 53009 --- [       Custom-9] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.823  INFO 53009 --- [      Custom-10] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.824  INFO 53009 --- [      Custom-11] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:32.825  INFO 53009 --- [      Custom-12] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】
2023-10-31 18:00:33.296  INFO 53009 --- [       Custom-1] com.jeremy.threadpool.AsyncServiceImpl   : 【结束执行任务】
2023-10-31 18:00:33.296  INFO 53009 --- [       Custom-1] com.jeremy.threadpool.AsyncServiceImpl   : 【开始执行任务】

采用线程池执行的任务，多个线程同时执行任务，能有效的降低了任务的延迟性。定时任务间隔固定时间从数据库 Mysql 或者 Redis 获取需要请求的数据，同时执行请求。

这样就有几个问题：

除了定时器，还有什么组件可以解决上面问题，那就是使用消息中间件了。

方案三：消息中间件 + 线程池（可行）

使用线程池的方式开启多个线程运行。那针对固定时间间隔和只能同时执行的问题使用消息中间件就能很好的解决问题，消息中间件采用「生产+消费」模型实现消息的生产和消费，

延迟队列

本文使用消息中间件 RabbitMQ实现延迟队列，具体实现可以看我的另外一篇文章延迟队列实现订单超时自动取消,具体实现流程图试下如下。

请求发送失败之后，调用生产者发送消息，经过设定的时间间隔之后，发送给消费者，消费端再次发起请求，如果请求失败，再调用生产者发送消息，并设置好下一次的时间间隔，其中消费端发起任务使用线程池发起请求。

❞

下载 RabbitMQ 延迟消息的插件 delayed_message_exchange，

在Github官网找到对应的版本，我选择的是 3.8.17：

配置延迟队列:

@Configuration
public class XDelayedMessageConfig {

 /**
  * 延迟交换机
  */
 public static final String DELAYED_EXCHANGE = "exchange.delayed";

 /**
  * 重试队列
  */
 public static final String RETRY_QUEUE = "queue.retry";


 /**
  * 重试routing key
  */
 public static final String RETRY_ROUTING_KEY = "routingKey.bind.retry";

    @Bean
 public Queue retryQueue() {
  return new Queue(RETRY_QUEUE,true);
 }


 /**
  * 定义延迟交换机
  * 交换机的类型为 x-delayed-message
  * @return
  */
 @Bean
 public CustomExchange delayedExchange() {
  Map<String,Object> map = new HashMap<>();
  map.put("x-delayed-type","direct");
  return new CustomExchange(DELAYED_EXCHANGE,"x-delayed-message",true,false,map);
 }



 @Bean
 public Binding retryQueueBinding() {
  return BindingBuilder.bind(retryQueue()).to(delayedExchange()).with(RETRY_ROUTING_KEY).noargs();
 }



}

在发送端模拟重试机制，设置时间间隔 5、10、30 秒。

@Autowired
private RabbitTemplate rabbitTemplate;

private final int[] INTERVAL_ARRAY= {5,10,30};

@GetMapping("/retry")
public String retry(int index) {
    if (index >= 0 && index <= 2) {
        send(index +",延迟" + INTERVAL_ARRAY[index] + "s",INTERVAL_ARRAY[index]);
    }
    return "ok";
}


private void send(String message,Integer delayTime) {
    message = message + " " + DateUtil.dateFormat(new Date());
    System.out.println("【发送消息】" + message);
    rabbitTemplate.convertAndSend(XDelayedMessageConfig.DELAYED_EXCHANGE,XDelayedMessageConfig.RETRY_ROUTING_KEY,
            message, message1 -> {
                message1.getMessageProperties().setDelay(delayTime*1000);
                return message1;
            });
}

接收端：

@RabbitListener(queues = XDelayedMessageConfig.RETRY_QUEUE)
public void delayProcess(String msg, Channel channel, Message message) {
    System.out.println("【接收消息】" + msg + " 当前时间" + DateUtil.dateFormat(new Date()));
    try {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    } catch (IOException e) {
        e.printStackTrace();
    }
    int index = Integer.parseInt(msg.split(",")[0]);
    retry(++index);
}

控制台输出：

【发送消息】0,延迟5s 10:59:29
【接收消息】0,延迟5s 10:59:29 当前时间10:59:33
【发送消息】1,延迟10s 10:59:33
【接收消息】1,延迟10s 10:59:33 当前时间10:59:43
【发送消息】2,延迟30s 10:59:43
【接收消息】2,延迟30s 10:59:43 当前时间11:00:10

其中 0、1、2表示重试的次数。通过延迟消息的方式，重试发送信息。每个任务作为一个消息进行消费。和定时服务相比，有以下几个优点：

总结

在发送一些异步通知时候，需要考虑到通知可能接收失败的情况，比如：

此时无法正确的及时推送通知，无法保证通知的可靠性。这个时候就需要重试多次，而且间隔要依次增加，因为服务启动或者网络的卡顿在经过一段时间就恢复了。后续重试成功的概率就更高了。

Copyright© 2013-2020

All Rights Reserved 京ICP备2023019179号-8