参考Redis实现简单消息队列

Redis提供了两种方式来作消息队列。一个是使用生产者消费模式模式，另外一个方法就是发布订阅者模式。前者会让一个或者多个客户端监听消息队列，一旦消息到达，消费者马上消费，谁先抢到算谁的，如果队列里没有消息，则消费者继续监听。后者也是一个或多个客户端订阅消息频道，只要发布者发布消息，所有订阅者都能收到消息，订阅者都是ping的。

生产消费模式

主要使用了redis提供的blpop获取队列数据，如果队列没有数据则阻塞等待，也就是监听。

#coding:utf-8
"""
生产者-消费者模式中的消费者
"""
import redis

class Task(object):
    def __init__(self):
        self.rcon = redis.StrictRedis(host='localhost', db=5)
        self.queue = 'task:prodcons:queue'

    def listen_task(self):
        while True:
            task = self.rcon.blpop(self.queue, 0)[1]
            print "Task get", task

if __name__ == '__main__':
    print '监听任务队列'
    Task().listen_task()

发布订阅模式

使用redis的pubsub功能，订阅者订阅频道，发布者发布消息到频道了，频道就是一个消息队列。

#coding:utf-8
"""
发布订阅模式中的订阅者
"""
import redis

class Task(object):

    def __init__(self):
        self.rcon = redis.StrictRedis(host='localhost', db=5)
        self.ps = self.rcon.pubsub()
        self.ps.subscribe('task:pubsub:channel')

    def listen_task(self):
        for i in self.ps.listen():
            if i['type'] == 'message':
                print "Task get", i['data']

if __name__ == '__main__':
    print '监听任务频道'
    Task().listen_task()

提供生产者和消息发布者

这里使用Flask做一个简单的页面作为生产者和消息发布者

# coding:utf-8
"""
提供一个页面作为生产者/消息发布者
"""
import redis
import random
import logging
from flask import Flask, redirect

app = Flask(__name__)

rcon = redis.StrictRedis(host='localhost', db=1)
prodcons_queue = 'task:prodcons:queue'  #生产者消费者队列名称
pubsub_channel = 'task:pubsub:channel' #发布订阅频道名称

@app.route('/')
def index():

    html = """
<br>
<center><h3>Redis Message Queue</h3>
<br>
<a href="/prodcons">生产消费者模式</a>
<br>
<br>
<a href="/pubsub">发布订阅者模式</a>
</center>
"""
    return html

#生产者
@app.route('/prodcons')
def prodcons():
    elem = random.randrange(10)
    rcon.lpush(prodcons_queue, elem)
    logging.info("lpush {} -- {}".format(prodcons_queue, elem))
    return redirect('/')
#消息发布者
@app.route('/pubsub')
def pubsub():
    ps = rcon.pubsub()
    ps.subscribe(pubsub_channel)
    elem = random.randrange(10)
    rcon.publish(pubsub_channel, elem)
    return redirect('/')

if __name__ == '__main__':
    app.run(debug=True)

测试

将三短代码分别保存在三个脚本文件中启动.

打开浏览器,会看到如下页面:

点击其中的两个超链接,两个一直在监听的消费者和订阅者会立即得到消息

消费者

订阅者