Python社区  »  Redis

[精华] python下使用redis构造一个简单的队列

爱情的枪 • 1 年前 • 5562 次点击  

python下使用redis构造一个简单的队列

本文展示了如何使用redis构建一个简单的多生产者,多消费者队列并且提供类似python标准库queue一样的接口。你可以使用这个队列方便的从多个进程或者耗时的计算到多个消费者进程之间共享数据。

我们使用redis列表来保存数据。redis列表按照字符串插入的顺序保存数据。

下面的redis命令会被用到:

  • rpush 在列表的末尾插入一个元素
  • blpop 从列表开头获取一个元素,如果列表是空则阻塞
  • lpop 从列表开头获取一个元素,如果列表是空则返回空
  • llen 返回列表的长度

实现过程使用了redis-py库和服务器进行交互:

RedisQueue.py

import redis

class RedisQueue(object):
    """Simple Queue with Redis Backend"""
    def __init__(self, name, namespace='queue', **redis_kwargs):
        """The default connection parameters are: host='localhost', port=6379, db=0"""
        self.__db= redis.Redis(**redis_kwargs)
        self.key = '%s:%s' %(namespace, name)

    def qsize(self):
        """Return the approximate size of the queue."""
        return self.__db.llen(self.key)

    def empty(self):
        """Return True if the queue is empty, False otherwise."""
        return self.qsize() == 0

    def put(self, item):
        """Put item into the queue."""
        self.__db.rpush(self.key, item)

    def get(self, block=True, timeout=None):
        """Remove and return an item from the queue.

        If optional args block is true and timeout is None (the default), block
        if necessary until an item is available."""
        if block:
            item = self.__db.blpop(self.key, timeout=timeout)
        else:
            item = self.__db.lpop(self.key)

        if item:
            item = item[1]
        return item

    def get_nowait(self):
        """Equivalent to get(False)."""
        return self.get(False)

使用:

>>> from RedisQueue import RedisQueue
>>> q = RedisQueue('test')
>>> q.put('hello world')

现在我们使用redis-cli客户端查看数据库,期望的结果如下:

redis 127.0.0.1:6379> keys *
1) "queue:test"
redis 127.0.0.1:6379> type queue:test
list
redis 127.0.0.1:6379> llen queue:test
(integer) 1
redis 127.0.0.1:6379> lrange queue:test 0 1
1) "hello world"

我们可以使用一个不同的脚本来获取数据:

>>> from RedisQueue import RedisQueue
>>> q = RedisQueue('test')
>>> q.get()
'hello world'

随后的q.get()调用会一直阻塞直到某人重新向队列发送一个新的数据。

接下来的工作将是到队列的 编码/解码(例如python-json),这样你就可以不受限制的发送任何字符串。

现在已经存在漂亮而又简单的hotqueue库,它具有像上面例子中的接口别且提供编码/解码功能。

其他值得提到的使用redis做后端的有:

  • flask-redis flask里使用redis做后端的一个基本的消息队列。
  • celery 一个基于分布式消息传递的异步任务队列/工作队列。比其他类库更高级点,可以配合不同的后端工作。
  • rq 简单的python类库作用是队列化任务并且在后端使用消费者进程处理它们。
  • resque 一个使用redis做后端的ruby库,主要为了创建后台工作,把他们放到多个队列,稍后处理他们。github在使用,并且有一个漂亮的web监控页面。
  • pyres python下resque的克隆版。

原文中的一个小bug,如下:

生产者:

In [1]: from RedisQueue import RedisQueue
In [2]: q = RedisQueue('test')
In [3]: q.put('1')
In [4]: q.put('2')

消费者:

In [10]: from RedisQueue import RedisQueue
In [11]: q = RedisQueue('test')
In [12]: q.get()
Out[12]: '1'
In [13]: q.get_nowait()

IndexError                                Traceback (most recent call last)
<ipython-input-13-4f5e3b567b9e> in <module>()
----> 1 q.get_nowait()

/data9/RedisQueue.py in get_nowait(self)
     36     def get_nowait(self):
     37         """Equivalent to get(False)."""
---> 38         return self.get(False)
     39

/data9/RedisQueue.py in get(self, block, timeout)
     31
     32         if item:
---> 33             item = item[1]
     34         return item
     35

IndexError: string index out of range

使用redis-py库连接redis数据库,获取下数据,很容易就看到问题出在哪里。

In [55]: r = redis.Redis()
In [56]: r.blpop('queue:test')
Out[56]: ('queue:test', '1')
In [57]: r.lpop('queue:test')
Out[57]: '2'

当使用非阻塞方式获取数据时,redis客户端返回的是一个string;当使用阻塞方式获取数据时,redis返回的数据是一个tuple。因此,修改下get函数中对item的判断条件就可以了:

if isinstance(item,tuple):
            item = item[1]
Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/1054
 
5562 次点击  
分享到微博
文章 [ 2 ]  |  最新文章 1 年前
younger.x.shen
Reply   •   1 楼
younger.x.shen    1 年前

根本没法用,直接考虑celery把,而且真正的队列也不是这样实现的

Py站长
Reply   •   2 楼
Py站长    1 年前

分享
社区所有版块导航
Python
python开源   Django   Python   DjangoApp   pycharm  
DATA
docker   Elasticsearch  
WEB开发
linux   MongoDB   tornado   bottle   Redis   IE   NoSql   DATABASE   MQ   Bootstrap   NGINX   js   其他Web框架   peewee   web工具   Git   zookeeper  
机器学习
机器学习算法  
Python88.com
反馈   公告   社区推广