基于redis实现延迟队列

Redis实现延时队列

延时队列里装的主要是延时任务，用延时队列来维护延时任务的执行时间。

1、延时队列有哪些使用情景？

1、如果请求加锁没加成功

可以将这个请求扔到延时队列里，延后处理。

2、业务中有延时任务的需要

比如说，文章定时发布。

2、基于redis实现延时队列

2.1.使用zset实现

使用redis的zset来实现延时队列

使用zset的添加、查询、删除命令

ZADD命令：
ZADD key score member [score member...]
ZRANGEBYSCORE命令：
ZRANGEBYSCORE key min max
ZREM命令：
ZREM key member [member ...]

1）将延时任务的【到期执行时间作为zset的score】、【延时任务序列化为一个字符串作为zset的member】使用zadd命令装进zset中。
2）zset会为这些延时任务按照到期执行时间排序。
3）设置多线程轮询zset获取到期的任务 | 用@Schedule注解标注定时轮询
4）在zset中删除这些到期任务
5）将这些到期任务放进list中，在list中使用【blpop/brpop】消费任务。

优化1：

对于第3步来说，由于是多线程，所以同一个到期任务可能会被多个进程获取到，然后再使用 zrem 进行争抢。最终一定只有一个进程zrem成功。因此对于那些zrem没成功的进程相当于白取任务。所以我们可以进一步优化：【使用 lua scripting 让 zrangebyscore 和 zrem 一同挪到服务器端进行原子化操作】。

优化2：

使用redis管道操作：通过管道方式将获取到的到期任务push到list中。

优化：redis管道

1）Redis的消息交互

客户端将请求传送给服务器，服务器处理完毕后，再将响应回复给客户端。这要花费一个网络数据包来回的时间。

详细过程是这样的：

1、客户端进程调用 write 将消息写到操作系统内核为套接字分配的发送缓冲 send buffer。

2、客户端操作系统内核将发送缓冲的内容发送到网卡，网卡硬件将数据通过「网际路由」送到服务器的网卡。

3、服务器操作系统内核将网卡的数据放到内核为套接字分配的接收缓冲 recv buffer。

4、服务器进程调用 read 从接收缓冲中取出消息进行处理。

5、服务器进程调用 write 将响应消息写到内核为套接字分配的发送缓冲 send buffer。

6、服务器操作系统内核将发送缓冲的内容发送到网卡，网卡硬件将数据通过「网际路由」送到客户端的网卡。

7、客户端操作系统内核将网卡的数据放到内核为套接字分配的接收缓冲 recv buffer。

8、客户端进程调用 read 从接收缓冲中取出消息返回给上层业务逻辑进行处理。

9、结束。

由上面的详细过程，可以知道：

【write操作】只负责将数据写到发送缓冲。但是如果发送缓冲满了，那么就需要等待缓冲空出空闲空间。这个就是写操作 IO 操作的真正耗时。

【read操作】只负责将数据从接收缓冲中取出来就完事了。如果缓冲是空的，那么就需要等待数据到来，这个就是读操作 IO 操作的真正耗时。

因此，一条消息的消息交互所花费的时间主要在于：

write 操作几乎没有耗时，直接写到发送缓冲就返回，而 read 就会比较耗时了，因为它要等待消息经过网络路由到目标机器处理后的响应消息,再回送到当前的内核读缓冲才可以返回。这才是一个网络来回的真正开销。

2）连续多条消息指令的消息交互

（1）普通方式：

如果要执行多条消息指令，则需要花费多个网络来回的时间。

（2）使用redis管道：

管道方式是客户端通过将多条命令white到发送缓存中，然后再一次性发送到服务端的缓存中，服务端处理完这些命令后将响应结果写到发送缓存中，最后再发送给接收缓存，客户端第一次read的时候，需要等待一个网络来回，而后续的read操作直接从接收缓存中取就行了，因此总的花费时间只有一个网络来回的时间。

以上两种方式的示例代码：

public class RedisPipelineTestDemo {
    public static void main(String[] args) {
        //连接redis
        Jedis jedis = new Jedis("10.101.17.180", 6379);
 
        //jedis逐一给每个set新增一个value
        String zSetKey = "Pipeline-test-set";
        int size = 100000;
        
        //普通方式
        long begin = System.currentTimeMillis();
        for (int i = 0; i < size; i++) {
            jedis.sadd(zSetKey + i, "aaa");
        }
        log.info("Jedis逐一给每个set新增一个value耗时：{}ms", (System.currentTimeMillis() - begin));
 
        //管道方式      
        Pipeline Pipeline = jedis.Pipelined();
        begin = System.currentTimeMillis();
        for (int i = 0; i < size; i++) {             
            Pipeline.sadd(zSetKey + i, "bbb");
        }         
        Pipeline.sync();
        log.info("Jedis Pipeline模式耗时：{}ms", (System.currentTimeMillis() - begin));
    }
}

普通方式：162655ms
管道方式：504ms

2.2.使用redis键空间通知

redis的PubSub，发布者订阅者模型。

摘自javaguide：

在 pub/sub 模式下，生产者需要指定消息发送到哪个 channel 中，而消费者则订阅对应的 channel 以获取消息。

Redis 中有很多默认的 channel，这些 channel 是由 Redis 本身向它们发送消息的，而不是我们自己编写的代码。其中，__keyevent@0__:expired 就是一个默认的 channel，负责监听 key 的过期事件。也就是说，当一个 key 过期之后，Redis 会发布一个 key 过期的事件到__keyevent@<db>__:expired这个 channel 中。

我们只需要监听这个 channel，就可以拿到过期的 key 的消息，进而实现了延时任务功能。

这个功能Redis官方称为keyspace notifications，字面意思就是键空间通知。

代码实现：

Spring已经实现了监听__keyevent@*__:expired这个channel这个功能，__keyevent@*__:expired中的*代表通配符的意思，监听所有的数据库。

在配置类中：

@Configuration
public class RedisConfiguration {

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(connectionFactory);
        return redisMessageListenerContainer;
    }

    @Bean
    public KeyExpirationEventMessageListener redisKeyExpirationListener(RedisMessageListenerContainer redisMessageListenerContainer) {
        return new KeyExpirationEventMessageListener(redisMessageListenerContainer);
    }
    
    /**
    KeyExpirationEventMessageListener实现了对__keyevent@*__:expiredchannel的监听
    
    当KeyExpirationEventMessageListener收到Redis发布的过期Key的消息的时候，会发布RedisKeyExpiredEvent事件
    **/

}

所以我们只需要监听RedisKeyExpiredEvent事件就可以拿到过期消息的Key，也就是延迟消息。对RedisKeyExpiredEvent事件的监听实现MyRedisKeyExpiredEventListener。

@Component
public class MyRedisKeyExpiredEventListener implements ApplicationListener<RedisKeyExpiredEvent> {

    @Override
    public void onApplicationEvent(RedisKeyExpiredEvent event) {
        byte[] body = event.getSource();
        System.out.println("获取到延迟消息：" + new String(body));
    }

}

缺点：

1、能否及时的监听到过期键过期，取决于【redis的过期键删除策略】。由于可能出现redis过期键已经过期，但是redis还未将其删除从而导致无法监听到过期键过期。使得最后消息延迟。