Redis应用-异步消息队列与延时队列

系列文章

• Redis应用-分布式锁
• Redis应用-异步消息队列与延时队列
• Redis应用-位图
• Redis应用-HyperLogLog
• Redis应用-布隆过滤器
• Redis应用-限流
• Redis应用-Geo

异步消息队列

说道消息队列，你确定会想到Kafka、Rabbitmq等消息中间件，这些专业的消息中间件提供了不少功能特性，固然他的部署使用维护都是比较麻烦的。若是你对消息队列没那么高要求，想要轻量级的，使用Redis就没错啦。

Redis经过list数据结构来实现消息队列.主要使用到以下命令：

• lpush和rpush入队列
• lpop和rpop出队列
• blpop和brpop阻塞式出队列

废话补很少说上代码：

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

//发送消息
$redis->lPush($list, $value);

//消费消息
while (true) {
    try {
        $msg = $redis->rPop($list);
        if (!$msg) {
            sleep(1);
        }
        //业务处理
     
    } catch (Exception $e) {
        echo $e->getMessage();
    }
}
复制代码

上面代码会有个问题若是队列长时间是空的，那pop就不会不断的循环，这样会致使redis的QPS升高，影响性能。因此咱们使用sleep来解决，当没有消息的时候阻塞一段时间。但其实这样还会带来另外一个问题，就是sleep会致使消息的处理延迟增长。这个问题咱们能够经过blpop/brpop 来阻塞读取队列。

blpop/brpop在队列没有数据的时候，会当即进入休眠状态，一旦数据到来，则马上醒过来。消息的延迟几乎为零。用blpop/brpop替代前面的lpop/rpop，就完美解决了上面的问题。

还有一个须要注意的点是咱们须要是用try/catch来进行异常捕获，若是一直阻塞在那里，Redis服务器通常会主动断开掉空连接，来减小闲置资源的占用。

延迟队列

你是否在作电商项目的时候会遇到以下场景：

• 订单下单后超过一小时用户未支付，须要关闭订单
• 订单的评论若是7天未评价，系统须要自动产生一条评论

这个时候咱们就须要用到延时队列了，顾名思义就是须要延迟一段时间后执行。Redis可经过zset来实现。咱们能够将有序集合的value设置为咱们的消息任务，把value的score设置为消息的到期时间，而后轮询获取有序集合的中的到期消息进行处理。

实现代码以下：

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

$redis->zAdd($delayQueue,$tts, $value);


while(true) {
    try{
        $msg = $redis->zRangeByScore($delayQueue,0,time(),0,1);
        if($msg){
            continue;
        }
        //删除消息
        $ok = $redis->zrem($delayQueue,$msg);
        if($ok){
            //业务处理
        }
    } catch(\Exception $e) {

    }
}
复制代码

这里又产生了一个问题，同一个任务可能会被多个进程取到以后再使用 zrem 进行争抢，那些没抢到的进程都是白取了一次任务，这是浪费。解决办法：将 zrangebyscore和zrem使用lua脚本进行原子化操做,这样多个进程之间争抢任务时就不会出现这种浪费了。

本文亦在微信公众号【小道资讯】发布，欢迎扫码关注！