前言：

   这里介绍一个python下，比celery更加简单的异步工具，真的是很简单，当然他的功能没有celery多，复杂程度也没有celery大，文档貌似也没有celery多，但是为啥会介绍rq这个东西 因为他够简单。

当然他虽然简单，但是也是需要中间人的，也就是 Broker，这里只能是redis了。 他没有celery支持的那么多，比如 redis rabbitmq mongodb mysql之类的。 说回来，咱们用rq，就是看重他的简单。

如对celery有兴趣，可以看看我以前写过的博文。 http://rfyiamcool.blog.51cto.com/1030776/1325062

安装redis以及python-rq包，redis的话，直接yum就行，python rq需要pip来搞定。

[root@67 ~]# pip install rq
Downloading/unpacking rq
  Downloading rq-0.4.5.tar.gz
  Running setup.py egg_info for package rq
    warning: no previously-included files matching ‘*‘ found under directory ‘tests‘
Requirement already satisfied (use --upgrade to upgrade): redis>=2.7.0 in /usr/lib/python2.6/site-packages (from rq)
Downloading/unpacking importlib (from rq)
  Downloading importlib-1.0.3.tar.bz2
  Running setup.py egg_info for package importlib
Requirement already satisfied (use --upgrade to upgrade): argparse in /usr/lib/python2.6/site-packages (from rq)
Installing collected packages: rq, importlib
  Running setup.py install for rq
    warning: no previously-included files matching ‘*‘ found under directory ‘tests‘
    Installing rqinfo script to /usr/bin
    Installing rqworker script to /usr/bin
  Running setup.py install for importlib
Successfully installed rq importlib
Cleaning up...

先开始官方的demo：

这个是咱们要后端异步的模块：

import requests
def count_words_at_url(url):
    resp = requests.get(url)
    return len(resp.text.split())

创建队列

from redis import Redis
from rq import Queue
q = Queue(connection=Redis())

然后，直接rqworker ！

一直往队列里面扔任务。

In [238]: result = q.enqueue(
             count_words_at_url, ‘http://nvie.com‘
)
In [241]: result = q.enqueue(
             count_words_at_url, ‘http://nvie.com‘
)
In [244]: result = q.enqueue(
             count_words_at_url, ‘http://nvie.com‘
)
In [247]: result = q.enqueue(
             count_words_at_url, ‘http://xiaorui.cc‘
)
In [250]: result = q.enqueue(
             count_words_at_url, ‘http://xiaorui.cc‘
)
In [253]: result = q.enqueue(
             count_words_at_url, ‘http://xiaorui.cc‘
)
In [256]: result = q.enqueue(
             count_words_at_url, ‘http://xiaorui.cc‘
)

rqworker的接口任务并执行：

（下面的log已经说明了一切，任务确实执行了，而且我在ipython下，很是流畅，我不需要担心任务是否很好的执行，我只需要把任务一扔，就甩屁股走人了。）

00:42:13 *** Listening on default...
00:42:22 default: nima.count_words_at_url(‘http://xiaorui.cc‘) (84f9d30f-8afc-4ea6-b281-4cb75c77779f)
00:42:22 Starting new HTTP connection (1): xiaorui.cc
00:42:23 Starting new HTTP connection (1): rfyiamcool.blog.51cto.com
00:42:23 Job OK, result = 2632
00:42:23 Result is kept for 500 seconds.
00:42:23
00:42:23 *** Listening on default...
00:42:27 default: nima.count_words_at_url(‘http://xiaorui.cc‘) (9fdaa934-e996-4719-8fb5-d619a4f15237)
00:42:27 Starting new HTTP connection (1): xiaorui.cc
00:42:28 Starting new HTTP connection (1): rfyiamcool.blog.51cto.com
00:42:28 Job OK, result = 2632
00:42:28 Result is kept for 500 seconds.
00:42:28
00:42:28 *** Listening on default...
00:42:28 default: nima.count_words_at_url(‘http://xiaorui.cc‘) (952cc12b-445e-4682-a12a-96e8019bc4a8)
00:42:28 Starting new HTTP connection (1): xiaorui.cc
00:42:28 Starting new HTTP connection (1): rfyiamcool.blog.51cto.com
00:42:28 Job OK, result = 2632
00:42:28 Result is kept for 500 seconds.
00:42:28
00:42:28 *** Listening on default...
00:42:29 default: nima.count_words_at_url(‘http://xiaorui.cc‘) (c25803e4-a3ad-4889-bbec-06cf1e77a11e)
00:42:29 Starting new HTTP connection (1): xiaorui.cc
00:42:29 Starting new HTTP connection (1): rfyiamcool.blog.51cto.com
00:42:29 Job OK, result = 2632
00:42:29 Result is kept for 500 seconds.
00:42:29
00:42:29 *** Listening on default..

紧接着咱们再跑一个我自己测试的模块，逻辑很简单在sleep情况下，是否会很好的执行，来测试他的异步任务执行。 当然你也可以rqworker执行的运行，下面的代码更像是event事件的感觉。

[root@67 ~]# cat worker.py
#xiaorui.cc
import os
import redis
from rq import Worker, Queue, Connection
listen = [‘high‘, ‘default‘, ‘low‘]
redis_url = os.getenv(‘REDISTOGO_URL‘, ‘redis://localhost:6379‘)
conn = redis.from_url(redis_url)
if __name__ == ‘__main__‘:
    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

下面是自己需要异步执行的模块代码~

[root@67 ~]# cat utils.py
#xiaorui.cc
import requests
import time
def tosleep(num):
    time.sleep(num)
    return num

咱们在ipython测试下吧：

In [53]: from redis import Redis
In [54]: from rq import Queue
In [55]:
In [56]: q = Queue(connection=Redis())
In [57]: from utils import tosleep
In [58]: for i in range(5):
    q.enqueue(tosleep,5)
   ....:
   ....:
Out[59]: Job(u‘8d71a0ee-695a-4708-b6cf-15821aac7299‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 42, 4, 47779))
Out[59]: Job(u‘27419b10-8b12-418c-8af1-43c290fc2bf3‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 42, 4, 51855))
Out[59]: Job(u‘7c98f0d1-7317-4c61-8bfa-10e223033948‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 42, 4, 53606))
Out[59]: Job(u‘0a84a48f-3372-4ef0-8aa8-d868de2e0c11‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 42, 4, 57173))
Out[59]: Job(u‘ad1986b9-a2fa-4205-93ab-a1b685d7cf88‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 42, 4, 58355))

看到没有，本来咱们调用了一个函数是sleep5s，但他不影响其他的代码的堵塞，会扔到队列里面后，迅速的执行后面的代码。

如果我想像celery那样，查看结果的话，也是用result方法的。

#xiaorui.cc
In [67]: job=q.enqueue(tosleep,5)
In [68]: job.result
In [69]: job.result
In [70]: job.result
In [71]: job.result
In [72]: job.result
Out[72]: 5

但是有个缺点，任务是异步方式的放到了redis的队列里面了，但是后端的work貌似是单进程的。。。当然也很好改，用threading针对每个任务进行fork线程就可以了。

#xiaorui.cc
In [47]: for i in range(5):
   ....:     q.enqueue(tosleep,5)
   ....:
   ....:
Out[47]: Job(u‘5edb3690-9260-4aba-9eaf-fa75fbf74a13‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 24, 54, 229289))
Out[47]: Job(u‘e91cfcb8-850b-4da4-8695-13f84a6a0222‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 24, 54, 233016))
Out[47]: Job(u‘cc6c78d4-e3b5-4c22-b027-8c070b6c43db‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 24, 54, 234333))
Out[47]: Job(u‘569decc8-7ad2-41eb-83cc-353d7386d2b9‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 24, 54, 235954))
Out[47]: Job(u‘155c493e-5a2c-4dcf-8d9b-3ae2934bf9e5‘, enqueued_at=datetime.datetime(2014, 5, 14, 15, 24, 54, 238030))
#xiaorui.cc

这个是worker.py打出来的日志：

23:24:59 Job OK, result = 5
23:24:59 Result is kept for 500 seconds.
23:24:59
23:24:59 *** Listening on high, default, low...
23:24:59 default: utils.tosleep(5) (e91cfcb8-850b-4da4-8695-13f84a6a0222)
23:25:04 Job OK, result = 5
23:25:04 Result is kept for 500 seconds.
23:25:04
23:25:04 *** Listening on high, default, low...
23:25:04 default: utils.tosleep(5) (cc6c78d4-e3b5-4c22-b027-8c070b6c43db)
23:25:09 Job OK, result = 5
23:25:09 Result is kept for 500 seconds.
23:25:09
23:25:09 *** Listening on high, default, low...
23:25:09 default: utils.tosleep(5) (569decc8-7ad2-41eb-83cc-353d7386d2b9)
23:25:14 Job OK, result = 5
23:25:14 Result is kept for 500 seconds.
23:25:14
23:25:14 *** Listening on high, default, low...
23:25:14 default: utils.tosleep(5) (155c493e-5a2c-4dcf-8d9b-3ae2934bf9e5)
23:25:19 Job OK, result = 5
23:25:19 Result is kept for 500 seconds.
23:25:19
23:25:19 *** Listening on high, default, low...

这里在看下官方给的例子：

from rq import Connection, Queue
from redis import Redis
from somewhere import count_words_at_url
# 创建redis的一个连接对象
redis_conn = Redis()
q = Queue(connection=redis_conn)  # 默认是用redis的default队列名
# 封装任务
job = q.enqueue(count_words_at_url, ‘http://xiaorui.cc‘)
print job.result   # => None
# Now, wait a while, until the worker is finished
time.sleep(2)
print job.result   # => 889

rq可以设置任务的优先级别的，比如一个low级别的。

q = Queue(‘low‘, connection=redis_conn)
q.enqueue(count_words_at_url, ‘http://nvie.com‘)

好了先这么着吧，官方 http://python-rq.org/docs/  还提供了很多实用的东西，比如装饰器啥的。

对了，官方提供了一个rq的管理平台页面。

地址是 https://github.com/nvie/rq-dashboard

本文出自 “峰云，就她了。” 博客，谢绝转载！