scene-question-1

5.24-场景题

订单到期关闭如何实现

典型回答

在电商、支付等系统中，一般都是先创建订单（支付单），再给用户一定的时间进行支付，如果没有按时支付的话，就需要把之前的订单（支付单）取消掉。这种类似的场景很多，比如到期自动收货、超时自动退款、下单后自动发送短信等等都是类似的业务问题。

订单到期关闭的实现有很多种方式，分别有：

被动关闭（不推荐）
定时任务（推荐，适合时间精确度要求不高的场景）
DelayQueue（不推荐，基于内存，无法持久化）
时间轮（不推荐，基于内存，无法持久化）
kafka（MQ方案不推荐，大量无效调度）
RocketMQ延迟消息（MQ方案不推荐，大量无效调度）
RabbitMQ死信队列（MQ方案不推荐，大量无效调度）
RabbitMQ插件（MQ方案不推荐，大量无效调度）
Redis过期监听（不推荐，容易丢消息）
Redis的ZSet（不推荐，可能会重复消费）
Redisson（推荐，可以用）

实现复杂度（包含用到的框架的依赖及部署）

Redisson > RabbitMQ插件 > RabbitMQ死信队列 > RocketMQ延迟消费 $\approx$ Redis的zset > Redis过期监听 $\approx$ kafka时间轮 > 定时任务 > Netty的时间轮 > JDK自带的DelayQueue > 被动关闭

不同场景中也适合不同方案:

自己玩：被动关闭
单体应用，业务量不大：Netty的时间轮、JDK自带的DelayQueue、定时任务
分布式应用，业务量不大：Redis过期监听、RabbitMQ死信队列、Redis的zset、定时任务
分布式应用、业务量大、并发高:Redisson、RabbitMQ插件、kafka时间轮、Rocket延迟任务、定时任务
业务量特别大：定时任务

总体考虑到成本、方案完整性、以及方案的复杂度，还有用到的第三方框架的流行度，优先考虑定时任务、Redisson+Redis、RabbitMQ插件、RocketMQ延迟消息等方案

但是，如果考虑到订单到期关闭的业务特点，如果在订单量特别大的时候，MQ其实并不适合：

为什么不建议使用MQ实现订单到期关闭? //todo

扩展知识

I、被动关闭

简单来说，就是订单创建好之后。我们系统上不做主动关单，什么时候用户来访问这个订单了，再去判断时间是不是超过了过期时间，如果过了时间就进行关单操作，然后再提示用户。

这种做法是最简单的，基本不需要开发定时关闭的功能，但是他的缺点也很明显，那就是如果用户一直不来查看这个订单，那么就会有很多脏数据冗余在数据库中一直无法被关单。

还有一个缺点，那就是需要在用户的查询过程中进行写的操作，一般写操作都会比读操作更耗时，而且有失败的可能，一旦关单失败了，就会导致系统处理起来比较复杂。

II、定时任务

具体实现细节就是我们通过一些调度平台来实现定时执行任务，任务就是去扫描所有到期的订单，然后执行关单动作。

这个方案的有点也是比较简单，实现起来很容易，基于Timer、ScheduledThreadPoolExecutor、或者像xxl-job这类调度框架都能实现，但是有以下几个问题：

时间不精确：一般定时任务基于固定的频率、按照时间定时执行的，那么就可能会发生很多订单已经到了超时时间，但是定时任务的调度时间还没到，那么就会导致这些订单的实际关闭时间要比应该关闭的时间晚一些。
无法处理大量订单：定时任务的方式是会把本来比较分散的关闭时间集中到任务调度的那一段时间，如果订单量比较大的话，那么就会导致任务执行时间很长，整个任务的时间越长，订单被扫描到时间可能就很晚，那么就会导致关闭时间更晚。
对数据库造成压力：定时任务集中扫表，这会使得数据库IO在短时间内被大量占用和消耗，如果没有做好隔离，并且业务量比较大的话，就可能会影响到线上的正常业务。
分库分表问题：订单系统，一旦订单量大就可能会考虑分库分表，在分库分表中进行全表扫描，这是一个极不推荐的方案。

解决方案如下：

定时任务扫表的问题解决方案

所以，定时任务的方案，适合对时间精确度要求不高、并且业务量不是很大的场景中。如果对时间精度要求比较高，这种方案不适用。（但是一般来说，订单的到期关闭这种业务，对时间精确度要求并不高，所以定时任务也是使用的最广泛的一种方案）

III、JDK自带的DelayQueue

有这样一种方案，他不需要借助任何外部的资源，直接基于应用自身就能实现，那就是基于JDK自带的DelayQueue来实现。

DelayQueue是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。

基于延迟队列，是可以实现订单的延迟关闭的，首先，在用户创建订单的时候，把订单加入到DelayQueue中，然后，还需要一个常驻任务不断的从队列中取出那些到了超时时间的订单，然后在把他们进行关单，之后再从队列中删除掉。

这个方案需要有一个线程，不断的从队列中取出需要关单的订单。一般在这个线程中需要加一个while(true)循环，这样才能确保任务不断的执行并且能够及时的取出超时订单。
使用DelayQueue实现超时关单的方案，实现起来简单，不须要依赖第三方的框架和类库，JDK原生就支持了。

当然这个方案也不是没有缺点的，首先，基于DelayQueue的话，需要把订单放进去，那如果订单量太大的话，可能会导致OOM的问题；另外，DelayQueue是基于JVM内存的，一旦机器重启了，里面的数据就都没有了。虽然我们可以配合数据库的持久化一起使用。而且现在很多应用都是集群部署的，那么集群中多个实例上的多个DelayQueue如何配合是一个很大的问题。

所以，基于JDK的DelayQueue方案只适合在单机场景、并且数据量不大的场景中使用，如果涉及到分布式场景，那还是不建议使用。

IV、Netty的时间轮

还有一种方式，和上面我们提到的JDK自带的DelayQueue类似的方式，那就是基于时间轮实现。

为什么要有时间轮呢？主要是因为DelayQueue插入和删除操作的平均时间复杂度——O(nlog(n))，虽然已经挺好的了，但是时间轮的方案可以将插入和删除操作的时间复杂度都降为O(1)。

时间轮可以理解为一种环形结构，像钟表一样被分为多个 slot。每个 slot 代表一个时间段，每个 slot 中可以存放多个任务，使用的是链表结构保存该时间段到期的所有任务。时间轮通过一个时针随着时间一个个 slot 转动，并执行 slot 中的所有到期任务。

基于Netty的HashedWheelTimer可以帮助我们快速的实现一个时间轮，这种方式和DelayQueue类似，缺点都是基于内存、集群扩展麻烦、内存有限制等等。

但是他相比DelayQueue的话，效率更高一些，任务触发的延迟更低。代码实现上面也更加精简。

所以，基于Netty的时间轮方案比基于JDK的DelayQueue效率更高，实现起来更简单，但是同样的，只适合在单机场景、并且数据量不大的场景中使用，如果涉及到分布式场景，那还是不建议使用。

V、Kafka的时间轮

既然基于Netty的时间轮存在一些问题，那么有没有其他的时间轮的实现呢？

还真有的，那就是Kafka的时间轮，Kafka内部有很多延时性的操作，如延时生产，延时拉取，延时数据删除等，这些延时功能由内部的延时操作管理器来做专门的处理，其底层是采用时间轮实现的。

而且，为了解决有一些时间跨度大的延时任务，Kafka 还引入了层级时间轮，能更好控制时间粒度，可以应对更加复杂的定时任务处理场景；

Kafka 中的时间轮的实现是 TimingWheel 类，位于 kafka.utils.timer 包中。基于Kafka的时间轮同样可以得到O(1)时间复杂度，性能上还是不错的。

基于Kafka的时间轮的实现方式，在实现方式上有点复杂，需要依赖kafka，但是他的稳定性和性能都要更高一些，而且适合用在分布式场景中。

VI、RocketMQ延迟消息

相比于Kafka来说，RocketMQ中有一个强大的功能，那就是支持延迟消息

延迟消息，当消息写入Broker后，不会立刻被消费者消费，需要等待指定的时长后才可被消费处理的消息，称为延迟消息

有了延迟消息，我们就可以在订单创建好之后，发送一个延迟消息，比如20分钟取消订单，那就发一个延迟20分钟的延迟消息，然后在20分钟之后，消息就会被消费者消费，消费者在接收到消息之后，去关单就行了。

但是，RocketMQ的延迟消息并不是支持任意时长的延迟的，它只支持：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h这几个时长。（商业版支持任意时长）

可以看到，有了RocketMQ延迟消息之后，我们处理上就简单很多，只需要发消息，和接收消息就行了，系统之间完全解耦了。但是因为延迟消息的时长受到了限制，所以并不是很灵活。

如果我们的业务上，关单时长刚好和RocketMQ延迟消息支持的时长匹配的话，那么是可以基于RocketMQ延迟消息来实现的。否则，这种方式并不是最佳的。（但是在RocketMQ 5.0中新增了基于时间轮实现的定时消息，可以解决这个问题）

VII、RabbitMQ死信队列

延迟消息不仅在RocketMQ中支持，其实在RabbitMQ中也是可以实现的，只不过其底层是基于死信队列实现的。

当RabbitMQ中的一条正常的消息，因为过了存活时间（TTL过期）、队列长度超限、被消费者拒绝等原因无法被消费时，就会变成Dead Message，即死信。

当一个消息变成死信之后，他就能被重新发送到死信队列中（其实是交换机-exchange）。
那么基于这样的机制，就可以实现延迟消息了。那就是我们给一个消息设定TTL，但是并不消费这个消息，等他过期，过期后就会进入到死信队列，然后我们再监听死信队列的消息消费就行了。
而且，RabbitMQ中的这个TTL是可以设置任意时长的，这就解决了RocketMQ的不灵活的问题。

但是，死信队列的实现方式存在一个问题，那就是可能造成队头阻塞，如果死信队列中的队头的消息一直无法消费成功，那么就会阻塞整个队列，这时候即使排在他后面的消息过期需要处理了，那么也会被一直阻塞。

基于RabbitMQ的死信队列，可以实现延迟消息，非常灵活的实现定时关单，并且借助RabbitMQ的集群扩展性，可以实现高可用，以及处理大并发量。他的缺点第一是可能存在消息阻塞的问题，还有就是方案比较复杂，不仅要依赖RabbitMQ，而且还需要声明很多队列(exchange)出来，增加系统的复杂度。

VIII、RabbitMQ插件

其实，基于RabbitMQ的话，可以不用死信队列也能实现延迟消息，那就是基于rabbitmq_delayed_message_exchange插件，这种方案能够解决通过死信队列实现延迟消息出现的消息阻塞问题。但是该插件从RabbitMQ的3.6.12开始支持的，所以对版本有要求。

这个插件是官方出的，可以放心使用，安装并启用这个插件之后，就可以创建x-delayed-message类型的队列了。

前面我们提到的基于死信队列的方式，是消息先会投递到一个正常队列，在TTL过期后进入死信队列。但是基于插件的这种方式，消息并不会立即进入队列，而是先把他们保存在一个基于Erlang开发的Mnesia数据库中，然后通过一个定时器去查询需要被投递的消息，再把他们投递到x-delayed-message队列中。

基于RabbitMQ插件的方式可以实现延迟消息，并且不存在消息阻塞的问题，但是因为是基于插件的，而这个插件支持的最大延长时间是(2^32)-1 毫秒，大约49天，超过这个时间就会被立即消费。但是他基于RabbitMQ实现，所以在可用性、性能方便都很不错

IX、Redis过期监听

在 redis.conf 中，加入一条配置notify-keyspace-events Ex开启过期监听，然后再代码中实现一个KeyExpirationEventMessageListener，就可以监听key的过期消息了。

这样就可以在接收到过期消息的时候，进行订单的关单操作。

这个方案不建议大家使用，是因为Redis官网上明确的说过，Redis并不保证Key在过期的时候就能被立即删除，更不保证这个消息能被立即发出。所以，消息延迟是必然存在的，随着数据量越大延迟越长，延迟个几分钟都是常事儿。

而且，在Redis 5.0之前，这个消息是通过PUB/SUB模式发出的，他不会做持久化，至于你有没有接到，有没有消费成功，他不管。也就是说，如果发消息的时候，你的客户端挂了，之后再恢复的话，这个消息你就彻底丢失了。（在Redis 5.0之后，因为引入了Stream，是可以用来做延迟消息队列的。）

X、Redis的zset

我们可以借助Redis中的有序集合——zset来实现这个功能。

zset是一个有序集合，每一个元素(member)都关联了一个 score，可以通过 score 排序来取集合中的值。

我们将订单超时时间的时间戳（下单时间+超时时长）与订单号分别设置为 score 和 member。这样redis会对zset按照score延时时间进行排序。然后我们再开启redis扫描任务，获取”当前时间 > score”的延时任务，扫描到之后取出订单号，然后查询到订单进行关单操作即可。

使用redis zset来实现订单关闭的功能的优点是可以借助redis的持久化、高可用机制。避免数据丢失。但是这个方案也有缺点，那就是在高并发场景中，有可能有多个消费者同时获取到同一个订单号，一般采用加分布式锁解决，但是这样做也会降低吞吐型。

但是，在大多数业务场景下，如果幂等性做得好的，多个消费者取到同一个订单号也无妨。

XI、Redisson + Redis

Redisson时一个在Redis的基础上实现的框架，它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。

Redisson中定义了分布式延迟队列RDelayedQueue，这是一种基于我们前面介绍过的zset结构实现的延时队列，它允许以指定的延迟时长将元素放到目标队列中。

其实就是在zset的基础上增加了一个基于内存的延迟队列。当我们要添加一个数据到延迟队列的时候，redisson会把数据+超时时间放到zset中，并且起一个延时任务，当任务到期的时候，再去zset中把数据取出来，返回给客户端使用。

如何基于Redisson实现一个延迟队列

基于Redisson的实现方式，是可以解决基于zset方案中的并发重复问题的，而且还能实现方式也比较简单，稳定性、性能都比较高。

为什么不建议使用MQ实现订单到期关闭？

资源占用与成本：如果系统中存在大量订单，为每个订单都创建一个延迟消息可能会导致消息队列中积压大量的消息，这不仅增加了消息队列的资源消耗，也可能导致增加成本，尤其是在使用云服务提供商的消息队列服务时。
延迟消息的限制：首先并不是所有的消息队列服务都支持延迟消息，及时有一些支持，也可能对消息的延迟时间有限制。例如，某些服务可能限制延迟时间的最大值，这可能无法满足所有订单的到期关闭需求。
可靠性问题：虽然消息队列一般来说可靠性较高，但是也没办法做到100%不丢消息，所以在极端情况下，会有丢消息的风险。
精确性问题：消息的投递延迟是比较常见的一种情况，这可能会导致订单关闭操作不够精确。
大量无效消息：使用MQ实现订单到期关闭就要把订单放到MQ中，但是大部分订单会提前取消或者完成支付，这就会导致很多无效的消息。
扩展性问题：随着系统规模的扩大，依赖于消息队列的延迟消息来处理订单到期可能遇到扩展性问题。系统可能需要更复杂的消息队列管理策略和更高效的资源利用策略来应对不断增长的订单量。

定时任务扫表的问题解决方案

数据量多，扫表慢

随着需要扫描的表中的数据量越来越大，通过定时任务扫表的方式会越来越慢，那么想要解决这个问题，首先可以考虑加索引。

我们可以在state字段上增加一个索引，虽然这个字段的区分度不高，但是一般来说，这张表中，SUCCESS的数据量占90%，而INIT的数据量只占10%，而我们扫表的时候只关心INIT即可，所以增加索引后，扫表的效率是可以大大提升的。

其次，可以考虑多线程并发扫表，这里可以考虑采用线程池，在任务中开多个线程并发的从数据库中扫描数据进行处理。

但是这样做，会带来一个问题，那就是多个线程之间如何做好隔离，如何确保不会出现并发导致同一条记录被多个线程执行多次呢？

首先最基本的保障，扫表之后的处理逻辑要做好幂等控制，一旦出现了重复的情况，下游也能因为做了幂等而不会重复处理。

除此以外，在扫表的时候，可以通过分段的思想进行数据隔离。举个例子：

Long minId = messageService.getMinInitId();

for(int i = 1; i <= threadPool.size(); i++) {
  Long maxId = minId + segmentSize()*i;
  
  List<Message> messages = messageService.scanInitMessages(minId,maxId);
  
  procss(messages);
  minId = maxId + 1;
}

像上面的例子中，假设有10个线程，那么第一个线程就扫描Id处于0-1000的数据，第二个线程扫描1001-2000的数据，第三个线程扫描2001-3000的数据。这样以此类推，线程之间通过分段的方式就做好了隔离，可以避免同一个数据被多个线程扫描到。

这个做法，有个小问题，那就是INIT的数据的ID可能不是连续的，那么就需要考虑其他的分段方式，比如在事件表中增加一个业务ID，然后根据这个biz_id做分片也可以。

比如：

for (int i = 1; i <= threadPool.size(); i++) {
	List<Message> messages = messageService.scanInitMessages(i);
	process(messages);
}

这样在SQL中：

1
2
3

SELECT * FROM RETRY_MESSAGE WHERE
STATE = "INIT"
AND BIZ_ID LIKE "${frontNumber}$";

那么，不同线程执行的SQL就不一样了，分别是：

SELECT * FROM RETRY_MESSAGE WHERE
STATE = "INIT"
AND BIZ_D LIKE "1%"

SELECT * FROM RETRY_MESSAGE WHERE
STATE = "INIT"
AND BIZ_D LIKE "2%"

SELECT * FROM RETRY_MESSAGE WHERE
STATE = "INIT"
AND BIZ_D LIKE "3%"

SELECT * FROM RETRY_MESSAGE WHERE
STATE = "INIT"
AND BIZ_D LIKE "4%"

集中式扫表会影响正常业务

如果业务量比较大的话，集中式的扫描数据库势必给数据库带来一定的压力，那么就会影响到正常的业务。

那么想要解决这个问题，首先可以考虑，不扫主库，而是扫描备库。之所以能这么做，是因为这个业务场景一般都是可以接受一定的数据延迟的，那么备库带来延迟就可以忽略，但是备库是没有业务操作的，所以对备库的扫描是不会对业务造成影响的。

当然，这里还要考虑一个问题，那就是备库扫描数据之后的执行，执行完该如何同步到主库，这里可以直接修改主库，主备库数据ID一致的，直接去修改主库的就行了。不建议直接在备库上修改。

但是不管怎么样，备库还是可以分担扫表的这个大量高峰请求的。

除了扫备库，还有一个方案，那就是做分库。把原来集中在同一个数据库的数据分散到不同的数据库中，这样用集群代替单库来整体对外提供服务，可以大大的提升吞吐量。

因为多个数据库的话，每个库提供的连接数就会多，并且多个实例的话，CPU、IO、LOAD这些指标也可以互相分担。

定时扫表存在延迟问题

定时任务都是集中式的定时执行的，那么就会存在延迟的问题。随着数据库越来越大，延迟会越来越长。

想要降低延迟，那就要抛弃定时任务的方案，可以考虑延迟消息，基于延迟消息来做定时执行。

用了延迟消息之后，还可以缓解数据库的压力。也能比定时扫表的性能要好，实时性也更高。

当然，引入另外一个中间件也需要考虑成本的。

同步转异步

就是同步先执行，失败了的话，再异步执行。如：

private static ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
    .setNameFormat("hollis-pool-%d").build();

private static ExecutorService pool = new ThreadPoolExecutor(5, 200,
    0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
@Transactional(rollbackFor = Exception.class)
public void pay(PayRequest payRequest){
	//在同一个事务中做本地业务操作和记录消息
	payService.doPay(payRequest);
    retryMessageService.init(payRequest);
	//同步执行一次外部调用
    try{
        Result result = outerService.doSth(payRequest);
        if(result.isSuccess()){
            retryMessageService.success(payRequest);
        }
    }catch(Exception e){
        // 捕获异常，失败依赖异步重试
    }
	
}

如上，在同步接口中，先尝试执行一次要可能会失败的任务，如果成功了，那就把事件推进到成功。如果失败了也无所谓，因为会有异步定时任务捞起来重试。

如何基于Redisson实现一个延迟队列

Redisson中定义了分布式延迟队列RDelayedQueue，这是一种基于我们前面介绍过的zset结构实现的延时队列，它允许以指定的延迟时长将元素放到目标队列中。

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>最新版</version> 
</dependency>

定义一个Redisson客户端

@Configuration
public class RedissonConfig {
    
    @Bean(destroyMethod="shutdown")
    public RedissonClient redisson() throws IOException {
        Config config = new Config();
		config.useSingleServer().setAddress("redis://127.0.0.1:6379");
		RedissonClient redisson = Redisson.create(config);
        return redisson;
    }
}

接下来，在想要使用延迟队列的地方做如下方式：

import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
@Component
public class RedissonOrderDelayQueue {
    @Autowired
    RedissonClient redisson;
    public void addTaskToDelayQueue(String orderId) {
      
        RBlockingDeque<String> blockingDeque = redisson.getBlockingDeque("orderQueue");
        RDelayedQueue<String> delayedQueue = redisson.getDelayedQueue(blockingDeque);
        System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "添加任务到延时队列里面");
        delayedQueue.offer(orderId, 3, TimeUnit.SECONDS);
        delayedQueue.offer(orderId, 6, TimeUnit.SECONDS);
        delayedQueue.offer(orderId, 9, TimeUnit.SECONDS);
    }
   public String getOrderFromDelayQueue() {
        RBlockingDeque<String> blockingDeque = redisson.getBlockingDeque("orderQueue");
        RDelayedQueue<String> delayedQueue = redisson.getDelayedQueue(blockingDeque);
        String orderId = blockingDeque.take();
        return orderId;
    }
}

使用offer方法将两条延迟消息添加到RDelayedQueue中，使用take方法从RQueue中获取消息，如果没有消息可用，该方法会阻塞等待，直到消息到达。

我们使用 RDelayedQueue 的 offer 方法将元素添加到延迟队列，并指定延迟的时间。当元素的延迟时间到达时，Redisson 会将元素从 RDelayedQueue 转移到关联的 RBlockingDeque 中。

使用 RBlockingDeque 的 take 方法从关联的 RBlockingDeque 中获取元素。这是一个阻塞操作，如果没有元素可用，它会等待直到有元素可用。

所以，为了从延迟队列中取出元素，使用 RBlockingDeque 的 take 方法，因为 Redisson 的 RDelayedQueue 实际上是通过转移元素到关联的 RBlockingDeque 来实现延迟队列的。

interview

#scene #mq #redis+redisson

scene-question-1

https://www.polarishaw.top/2024/05/24/scene-question-1/

作者

Polaris

发布于

2024年5月24日

许可协议

troubleshooting-1 上一篇

tea5.20-5.24 下一篇