V2EX = way to explore

V2EX 是一个关于分享和探索的地方

已注册用户请登录

这是一个创建于 50 天前的主题，其中的信息可能已经有所发展或是发生改变。

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

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

其中有段话：

重新发送通知，直到成功为止（在通知一直不成功的情况下，微信总共会发起多次通知，通知频率为 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 表示重试的次数。通过延迟消息的方式，重试发送信息。每个任务作为一个消息进行消费。和定时服务相比，有以下几个优点：

支持动态间隔
任务不是同时执行，降低服务器的压力。

总结

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

请求的服务器挂了。
网络发生了波动。
服务器响应异常，服务重启。

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

定时重试
- 定时重试首先不符合变化的间隔时间，间隔的时间是固定的，重试的任务都堆积在一起请求，这样也会给服务器造成很大的压力。而空闲的时候，服务器的利用率有比较低。
- 同时请求，只能一个一个同步执行任务，同时执行的任务越多，延迟就越严重。
定时任务 + 线程池
- 为了解决同时处理任务，添加了自定义的线程池，因为请求属于 IO 密集型，所以设置线程数为 CPU 核数的两倍。
- 多个任务执行，降低了延迟性。
- 无法满足动态间隔时间的问题，而且同时请求服务器压力大。
延迟队列 + 线程池
- 延迟时间请求可以使用到延迟队列，每个任务都作为一个消息。每次处理不成功，就发送消息到延迟队列中，到达时间间隔之后，再消费消息。如果请求失败再重复以上操作。
- 消费者处理消息，使用线程池处理，加快处理速度。也可以开启多台服务器分发处理任务，加快处理速度，降低任务的延迟性。

如果感觉写的不错的，欢迎关注我的公众号

目前尚无回复

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

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

微信为何要这么设计

异步通知的重试机制

重试机制的实现

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

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

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

延迟队列

总结

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