文档章节

redis06消息队列

AllenOR灵感
 AllenOR灵感
发布于 2017/09/10 01:18
字数 1990
阅读 12
收藏 0

任务队列

在web开发的场景中撰写常规的代码,一般是在一个视图层(view)将所有业务逻辑完成之后直接返回数据(可以是一个渲染后的html也可以是一组json数据)给用户的浏览器;业务比较简单且单一的逻辑处理响应时间通常都比较短,但涉及到发邮件、数据爬取、密集计算、复杂业务这种情况耗时较长的操作,如果不采取任何措施的话,那么在用户这边最直观的感受就是提交了请求后页面就僵在那里了(如果时间大于10秒钟,可能就会认为出问题了)。

redis的任务队列(列表)就是这种耗时等待比较长场景的解决办法之一,视图层是一个负责塞消息的对象(Producer角色),然后在单独写一个独立的进程来读取这个列表(Consumer)去完成后续的工作;那么视图层就可以直接返回一个页面给用户,告诉用户这个事情的状态是正在处理中,当处理完成之后将状态改成已完成,并提供一个链接让用户可以去查看任务明细结果。

将一个耗时操作拆开要从好几个方面进行考虑,用户的等待体验、将相同功能相同作用的代码抽象出来解耦、简化逻辑层的代码复杂度。

 

非堵塞读取(rpop)

queue_rpop/producer.py

import redis

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)

# 清空所有键
for number, key in enumerate(r.keys()):
    r.delete(key)

# 准备数据
r.rpush('queue:send_mail', *['3330356463@qq.com', 'zhengtong0898@aliyun.com'])

queue_rpop/consumer1.py

import redis
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)


while True:
    email = r.rpop('queue:send_mail')
    print('[{} {} INFO]: {}'.format(
        dt.now(),
        'queue:send_mail',
        email
    ))

queue_rpop/consumer2.py

import redis
import time
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)


def send_mail(email):
    print('[{} {} INFO]: send mail to {}'.format(
        dt.now(),
        'queue:send_mail',
        email
    ))

if __name__ == '__main__':

    while True:
        mail = r.rpop('queue:send_mail')
        if not mail:

            print('[{} {} INFO]: {}'.format(
                dt.now(),
                'queue:send_mail',
                mail
            ))

            time.sleep(1)
            continue
        send_mail(mail)

运行

# 运行consumer1.py
python consumer1.py
# 显示结果,这种方式对CPU消耗非常大,整体的IO消耗也很高。
[2017-04-26 18:59:31.822200 queue:send_mail INFO]: None
[2017-04-26 18:59:31.822700 queue:send_mail INFO]: None
[2017-04-26 18:59:31.822700 queue:send_mail INFO]: None
[2017-04-26 18:59:31.823200 queue:send_mail INFO]: None
[2017-04-26 18:59:31.823200 queue:send_mail INFO]: None
[2017-04-26 18:59:31.823700 queue:send_mail INFO]: None
[2017-04-26 18:59:31.824200 queue:send_mail INFO]: None
...
...
...


# 运行consumer2.py (为了降低没有必要的系统消耗)
python consumer2.py

# 再开一个窗口运行 producer.py
python producer.py

# 显示结果
[2017-04-26 19:15:29.695200 queue:send_mail INFO]: None
[2017-04-26 19:15:30.695700 queue:send_mail INFO]: None
[2017-04-26 19:15:31.695700 queue:send_mail INFO]: None
[2017-04-26 19:15:32.696700 queue:send_mail INFO]: None
[2017-04-26 19:15:33.697200 queue:send_mail INFO]: send mail to 3330356463@qq.com
[2017-04-26 19:15:33.697700 queue:send_mail INFO]: send mail to zhengtong0898@aliyun.com
[2017-04-26 19:15:33.698200 queue:send_mail INFO]: None
[2017-04-26 19:15:34.698700 queue:send_mail INFO]: None
[2017-04-26 19:15:35.699200 queue:send_mail INFO]: None

 

堵塞读取(brpop)

redis内部并没有采用非堵塞的socket模型,因此初步判断它是一个堵塞请求,它的等待机制是由redis内部来负责响应的。堵塞让资源消耗更少,逻辑撰写更容易。

queue_brpop/consumer.py

import redis
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)


def send_mail(email):
    print('[{} {} INFO]: send mail to {}'.format(
        dt.now(),
        'queue:send_mail',
        email
    ))

if __name__ == '__main__':

    while True:
        queue_name, mail = r.brpop('queue:send_mail')
        send_mail(mail)

运行

# 运行queue_brpop/consumer.py
python queue_brpop/consumer.py

# 运行queue_rpop/producer.py

# 显示结果(没有多余的输出,挺好!)
[2017-04-26 19:31:23.947200 queue:send_mail INFO]: send mail to 3330356463@qq.com
[2017-04-26 19:31:23.947700 queue:send_mail INFO]: send mail to zhengtong0898@aliyun.com
优先级(lpush、blpop)

最普遍和常见的新增列表列表元素是在当前列表后面追加元素,取值的话则是从左到右读取,这种行为在消息队列中被称为先进先出(FIFO);那也会有一些场景要求后进先出(LIFO),这就需要利用lpush(从左边开始插入元素)和blpop(从左边开始读取)。

quque_priority/producer.py

import redis

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)

# 清空所有键
for number, key in enumerate(r.keys()):
    r.delete(key)

# 准备数据
r.lpush('queue:send_mail', '3330356463@qq.com')
r.lpush('queue:send_mail', 'zhengtong0898@aliyun.com')
r.lpush('queue:send_mail', 'zhengtong0898@a.com')
r.lpush('queue:send_mail', 'zhengtong0898@b.com')
r.lpush('queue:send_mail', 'zhengtong0898@c.com')
r.lpush('queue:send_mail', 'zhengtong0898@d.com')
r.lpush('queue:send_mail', 'zhengtong0898@e.com')

queue_priority/consumer.py

import redis
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)


def send_mail(email):
    print('[{} {} INFO]: send mail to {}'.format(
        dt.now(),
        'queue:send_mail',
        email
    ))

if __name__ == '__main__':

    while True:
        queue_name, mail = r.blpop('queue:send_mail')
        send_mail(mail)

运行

# 运行producer.py,先将数据塞到redis列表中(每次都是从左边插入)
python queue_priority/producer.py

# 运行consumer.py,消费数据(每次都是从左边读取)
python queue_priority/consumer.py

# 显示结果,仔细看就可以看得出来,确实是后进先出(LIFO)
[2017-04-26 20:29:18.735700 queue:send_mail INFO]: send mail to zhengtong0898@e.com
[2017-04-26 20:29:18.735700 queue:send_mail INFO]: send mail to zhengtong0898@d.com
[2017-04-26 20:29:18.736200 queue:send_mail INFO]: send mail to zhengtong0898@c.com
[2017-04-26 20:29:18.736700 queue:send_mail INFO]: send mail to zhengtong0898@b.com
[2017-04-26 20:29:18.736700 queue:send_mail INFO]: send mail to zhengtong0898@a.com
[2017-04-26 20:29:18.737200 queue:send_mail INFO]: send mail to zhengtong0898@aliyun.com
[2017-04-26 20:29:18.737700 queue:send_mail INFO]: send mail to 3330356463@qq.com

 
 

发布/订阅

消息队列是将消息放在列表中,由一个或多个consumer去消费列表中不同的元素;而发布/订阅是将消息放在一个频道中,有一个或多个consumer去消费频道中相同的元素;若该频道中没有任何consumer,这次producer生产了一条消息放入到频道中,这条消息将会自动消失,因此发布/订阅模式要求consumer必须时刻运行,否则producer发送的任何消息都是无效的。

publish_subscribe/producer.py

import redis
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)

r.publish('channel1.1', 'hi')
r.publish('channel1.1', 'my')
r.publish('channel1.1', 'name')
r.publish('channel1.1', 'is')
r.publish('channel1.1', 'publisher')

publish_subscribe/consumer.py

import redis
import time
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)
p = r.pubsub()
p.subscribe('channel1.1')    # 支持同时订阅多个频道: p.subscribe('channel1.1', 'channel1.2')

while True:
    msg = p.get_message(ignore_subscribe_messages=True)
    if not msg:
        time.sleep(1)
        print('sleep 1')
        continue
    print('debug: ', msg)

运行

# 打开两个窗口运行两次 consumer.py, 都订阅'channel1.1'频道.
python publish_subscribe/consumer.py # 窗口一
python publish_subscribe/consumer.py # 窗口二

# 运行producer.py, 将消息塞到'channel1.1'频道.
python publish_subscribe/producer.py


# 显示结果(两边都同时存在)
[2017-04-27 13:01:17.712700 publish/subscribe INFO]: None
[2017-04-27 13:01:18.712700 publish/subscribe INFO]: None
[2017-04-27 13:01:19.712700 publish/subscribe INFO]: None
[2017-04-27 13:01:19.712700 publish/subscribe INFO]: {'data': 'hi', 'type': 'message', 'pattern': None, 'channel': 'channel1.1'}
[2017-04-27 13:01:19.712700 publish/subscribe INFO]: {'data': 'my', 'type': 'message', 'pattern': None, 'channel': 'channel1.1'}
[2017-04-27 13:01:19.712700 publish/subscribe INFO]: {'data': 'name', 'type': 'message', 'pattern': None, 'channel': 'channel1.1'}
[2017-04-27 13:01:19.712700 publish/subscribe INFO]: {'data': 'is', 'type': 'message', 'pattern': None, 'channel': 'channel1.1'}
[2017-04-27 13:01:19.712700 publish/subscribe INFO]: {'data': 'publisher', 'type': 'message', 'pattern': None, 'channel': 'channel1.1'}
[2017-04-27 13:01:20.713200 publish/subscribe INFO]: None
[2017-04-27 13:01:21.714200 publish/subscribe INFO]: None

主题订阅

除了订阅多个频道之外,redis也支持consumer通过glob通配符(psubscribe)来订阅不同主题的频道。

publish_psubscribe/producer.py

import redis
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)

r.publish('channel1.1', 'hi')
r.publish('channel1.2', 'my')
r.publish('channel1.3', 'name')
r.publish('channel1.4', 'is')
r.publish('channel1.5', 'publisher')

publish_psubscribe/consumer.py

import redis
import time
from datetime import datetime as dt

# 连接redis
r = redis.StrictRedis('192.168.1.124', decode_responses=True)
p = r.pubsub()
p.psubscribe('channel1.*')

while True:
    msg = p.get_message(ignore_subscribe_messages=True)
    if not msg:
        time.sleep(1)
        msg = None

    print('[{} {} INFO]: {}'.format(
        dt.now(),
        'publish/psubscribe',
        msg
    ))

运行

# 运行consumer.py
python publish_psubscribe/consumer.py

# 运行producer.py
publish_psubscribe/producer.py

# 显示结果
[2017-04-27 13:31:00.683700 publish/psubscribe INFO]: None
[2017-04-27 13:31:01.683700 publish/psubscribe INFO]: None
[2017-04-27 13:31:02.683700 publish/psubscribe INFO]: None
[2017-04-27 13:31:02.683700 publish/psubscribe INFO]: {'data': 'hi', 'type': 'pmessage', 'pattern': 'channel1.*', 'channel': 'channel1.1'}
[2017-04-27 13:31:02.683700 publish/psubscribe INFO]: {'data': 'my', 'type': 'pmessage', 'pattern': 'channel1.*', 'channel': 'channel1.2'}
[2017-04-27 13:31:02.683700 publish/psubscribe INFO]: {'data': 'name', 'type': 'pmessage', 'pattern': 'channel1.*', 'channel': 'channel1.3'}
[2017-04-27 13:31:02.683700 publish/psubscribe INFO]: {'data': 'is', 'type': 'pmessage', 'pattern': 'channel1.*', 'channel': 'channel1.4'}
[2017-04-27 13:31:02.683700 publish/psubscribe INFO]: {'data': 'publisher', 'type': 'pmessage', 'pattern': 'channel1.*', 'channel': 'channel1.5'}
[2017-04-27 13:31:03.683700 publish/psubscribe INFO]: None
[2017-04-27 13:31:04.683700 publish/psubscribe INFO]: None

参考

本文转载自:http://www.jianshu.com/p/52b695769fab

共有 人打赏支持
上一篇: rust05模块
AllenOR灵感
粉丝 11
博文 2635
码字总数 83001
作品 0
程序员
私信 提问
rt-thread的IPC机制之消息队列源码分析

消息队列是另一种常用的线程间通讯方式,它能够接收来自线程的不固定长度的消息,并把消息缓存在自己的内存空间中。其他线程也能够从消息队列中读取相应的消息,而当消息队列是空的时候,可以...

长平狐
2013/03/19
595
0
进程间通信(IPC)——消息队列

消息队列  消息队列是在消息的传输过程中保存消息的容器。消息队列管理器在将消息从它的源中继到它的目标时充当。队列的主要目的是提供路由并保证消息的传递;如果发送消息时接收者不可用,...

qq_38646470
2018/05/02
0
0
消息队列属性及常见消息队列介绍

什么是消息队列? 消息队列是在消息的传输过程中保存消息的容器,用于接收消息并以文件的方式存储,一个队列的消息可以同时被多个消息消费者消费。分布式消息服务DMS则是分布式的队列系统,消...

中间件小哥
2018/06/22
0
0
LINUX-进程间通信(二、消息队列)

LINUX进程间通信(一、命名管道与匿名管道) 消息队列概述 消息队列是消息的链接表,存放在内核中并由消息队列标识符表示。主要涉及到msgget(),msgsnd(),msgrcv(),msgctl()四个函数,在具...

sssssuuuuu666
2017/12/22
0
0
【IPC通信】Posix消息队列读写

先给出创建消息队列的程序: #include <stdio.h> include <stdlib.h> include <mqueue.h> include <sys/types.h> include <sys/stat.h> include <unistd.h> include <fcntl.h> include <errn......

恋恋美食
2011/11/17
0
0

没有更多内容

加载失败,请刷新页面

加载更多

JavaScript继承(二)——借用构造函数

JavaScript继承(一)——原型链中提出原型链继承的两个问题:一是原型的数据共享问题,二是创建子类型的实例时,不能向父类型的构造函数中传递参数。这两个问题的根源还是在于使用原型模式创...

Bob2100
32分钟前
0
0
day25:比较两个文件内容|杀进程|并发备份数据库|并发进程备份mysql库|监控全国CDN网站节点|

1、有两个文件 a.txt 和 b.txt,把 a.txt 中有的但 b.txt 中没有的行找出来,并写入列 c.txt,然后计算 c 的行数; 注释:比较两个文件两个文件不同,可以用的方法:diff md5sum grep -vf $1 $...

芬野de博客
47分钟前
4
0
抓包工具Charles

背景介绍 Charles是一款很实用,界面很友好(至少跟fiddler比起来),功能强大的抓包神器,因为它是基于 java 开发的,所以跨平台,Mac、Linux、Windows下都是可以使用的,并且在Android和i...

Jack088
48分钟前
2
0
徒手撸一个简单的RPC框架

徒手撸一个简单的RPC框架 之前在牛逼哄哄的 RPC 框架,底层到底什么原理得知了RPC(远程过程调用)简单来说就是调用远程的服务就像调用本地方法一样,其中用到的知识有序列化和反序列化、动态...

不学无数的程序员
今天
2
0
Java 面试题目最全集合1000+ 大放送,能答对70%就去BATJTMD试试~

2019,相对往年我们会发现今年猎头电话少了,大部分企业年终奖缩水,加薪幅度也不如往年,选择好offer就要趁早,现在开始准备吧,刷一波Java面试题,能回答70%就去BATJTMD大胆试试~ 以下是2...

mikechen优知
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部