zoukankan      html  css  js  c++  java
  • redis实现消息队列

    业务需求

    本文是以laravel框架来介绍redis队列,具体用法你可以参考http://www.cnblogs.com/lengthuo/p/7277260.html
    最近接受一个很简单的东西,(说起来很简单,硬是搞了2天。)我们业务中的一些定时是在晚上执行,但是有的定时必须推送微信消息给用户,为了不影响客户的休息,我们之后想把发去推迟这个任务。
    对于我们开发来说,我们只需要知道2件事,入队列和出队列。

    入队列

    非常简单,我们只需要把数据放到队列中就行了,这里选用redis来作为我们的容器来存储队列,当然你也可以选择用数据库,或者其他的,laraevel支持好多中方式。
    redis中用list去存储队列,但是当遇到延迟发送的时候,其实是使用redis的zset(有序的集合列表)来存储的队列。因为有延迟时间吗,我们知道redis的zset的数据结构,有一个value和score ,score是用来排序的,laravel很巧妙的用它来存储发送的时间戳。当然这些其实我们都不需要知道,在laravel框架中,QueueRedis已经封装的非常好了,我们只需要简单的调用

     //方法一 使用内置函数
     $job = (new SendWechatMessage($data))->onQueue('queue-name')->delay($delay);
     dispatch($job); 
    
    //方法二 使用门面
    //当没有延迟的时候,redis是直接存到list中的
    Queue::push(new SendWechatMessage($data), '', 'queue-name');
    //有延迟的时候存入的是zset类型中的
    Queue::later($delay, new SendWechatMessage($data), '', 'redpacket');
    

    在这推荐一个工具
    rdm.app 可视化的工具去操作redis,日志中还会有显示对redis操作的日志。很好用,极力推荐。

    出队列

    laravel中我们只需要简单的启动队列任务就行了,当然laravel文档讲解的更清楚,我就不想说了,我只是简单的提一下几点注意的,和一些原理的东西。

    启动队列的方式

    queue:work 默认只执行一次队列请求, 当请求执行完成后就终止;
    queue:listen 监听队列请求, 只要运行着, 就能一直接受请求, 除非手动终止;
    queue:work --daemon 同 listen 一样, 只要运行着, 就能一直接受请求, 不一样的地方是在这个运行模式下, 当新的请求到来的时候, 不重新加载整个框架, 而是直接 fire 动作.
    能看出来, queue:work --daemon 是最高级的, 一般推荐使用这个来处理队列监听.
    注意: 使用 queue:work --daemon , 当更新代码的时候, 需要停止, 然后重新启动, 这样才能把修改的代码应用上。

    原理

    首先对存入队列的值进行分析
    无延迟队列,直接使list存入,然后在queue:listen 的时候

        public function listen($connection, $queue, $delay, $memory, $timeout = 60)
        {
            $process = $this->makeProcess($connection, $queue, $delay, $memory, $timeout);
            while (true) {//无限循环,所以耗费资源,建议选择使用work命令
                $this->runProcess($process, $memory);
            }
        }
    

    一直从redis list中去取数据。所以一旦有生产者放入,就会很快的被消费之监听。也是非常简单的一种模式
    有延迟队列,我只说在laravel中的实现,可能每一种框架都会有不同的实现。
    在redis中是由zset来存储延迟队列值得,当有一个延迟队列推送来之后,会存入到redis的zset中,value为要存入的redis中的值,而score为time()+要推迟的时间,所以score存入的是时间戳,laravel框架中在解析的时候会组装一句命令发给redis服务器,
    zrangebyscore queues:queue-name:delayed -inf 1504105422
    这条命令的意思是:返回小于1504105422时间戳的值,也就是说每一次执行的时候,laravel都是拿当前的时间戳值和redis第一条比较(默认是顺序的,并且第一条是最小的)
    说到这是不是豁然开朗,如果不明白可以私聊我,也可以多看看,我相信读完你必有收获。

    如果你跟debug的话,就会发现其实laravel包装的所有操作只是来组装数据,最后直接发给redis服务器来处理,然后返回结果。其实就是这么简单。laravel包装的太过于完美,以至于我们什么都不用做。只是简单2行代码就实现了对redis消息队列的操作。

    总结

    有的时候我们放眼去看,对于这种多种服务交互的时候,原理都是一样的,就比如我们操作mysql,程序肯定不会认识,程序只负责组装sql语句,最后都是会交给mysql服务器来处理,redis也一样,我们只需要组装redis的命令,最后发给redis服务器,他只需要给我们返回结果就可以了。希望本篇文章对你有用。

          路漫漫其修远兮,吾将上下而求索
  • 相关阅读:
    邮件发送携带附件
    两个文件内容同行合并操作
    re模块,判断某行/某字符是否存在
    企业微信公众号告警Python脚本
    CodeForces 371D. Vessels 题解
    免安装 mysql
    kibana dev tools 操作 Elasticsearch
    win10 强制关掉被占用的端口
    值传递与引用传递
    微服务网关 soul
  • 原文地址:https://www.cnblogs.com/lengthuo/p/7455907.html
Copyright © 2011-2022 走看看