---
title: "rq (redis queue)实践"
url: "https://lerry.me/post/2013/05/rq-practice"
date: 2013-05-22T18:26:00.000Z
updated: 2026-09-07T00:12:15.595Z
tags:
  - "Queue"
  - "Python"
  - "RQ"
  - "Redis"
language: zh-CN
---

# rq (redis queue)实践

最近想写个东西，需要用到消息队列。Python支持的消息队列有好多，以前工作中也用过Celery，配置起来挺麻烦的。然后就想到曾经在V2EX上看到过的RQ，RQ是一个Python实现的消息队列系统，使用Redis作为后端存储，依赖挺少的，就这个吧。
RQ有多种使用方式，文档里已有介绍：[RQ Documentation](http://python-rq.org/docs/)。我使用了装饰器的方法，这样比较方便。
文档中是这样的，

```python
from rq.decorators import job

@job('low', conn=my_redis_conn, timeout=5)
def add(x, y):
    return x + y

job = add.delay(3, 4)
time.sleep(1)
print job.result
```

需要在装饰器中传入配置参数，这样岂不是很麻烦，能不能在一处定义呢？通过查看源码，发现job其实是一个类，传入参数对其实例化，然后在"__call__"方法里为被装饰的函数增加一个"delay"的方法，这个方法其实就是使用"enqueue_call"方法将函数调用添加到队列中，既然知道了原理就好办了。

```python
from model.connect import redis
from rq.decorators import job

def rq(f):
    _job = job(queue='default', connection=redis)
    return _job(f)
```
 
在此，我定义了一个叫 `rq` 的函数，作为新的装饰器。传入的函数 `f` 将被 `job` 的 `__call__` 方法修改，实现了和文档相同的效果，使用很简单。
 

```python
from somewhere import rq

@rq
def count_words_at_url(x):
    return len(x)
```


调用函数时和普通调用相同，如果需要将其加入消息队列则使用其"delay"方法

```python
from func import count_words_at_url

count_words_at_url.delay('hello')
```

需要注意的是，无论使用哪种方法加入消息队列，函数的定义和调用都不能放在同一个文件里，不然会报错：

```text
Functions from the __main__ module cannot be processed by workers.
```

我猜是因为 rq 调用函数时需要 import 函数所在文件，而 import 时又会运行函数调用，造成死循环。