Redis使用总结(3):实现简单的消息队列

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

参考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)