RQ异步任务队列介绍

[TOC]
> **引用:**

> - RQ (Redis Queue) is a simple Python library for queueing jobs and processing them in the background with workers. It is backed by Redis and it is designed to have a low barrier to entry. It should be integrated in your web stack easily.

#### :fa-file-o: 概述

目前后台业务系统有相当一部分数据任务是异步处理的，比如：通知转发的数据处理，微信主动反馈客户当前余额的任务，以及分享人与被分享人“得到了奖品”和“领取了奖品”的消息通知等等，而目前所有异步处理均是基于数据库来实现，关键是还没有做到统一结构和重用，使得在新的涉及异步处理的开发需求时都得重新各自实现一遍，这种做法的缺点是显而易见的，鉴于这些原因我们在项目中引入一个 **异步任务队列** 的统一框架

#### :fa-list: 特性
- 实现了python扩展的基于Redis的队列数据结构，可以像操作list一样使用

- 实现了方法参数级调用队列，即以方法参数为队列的方法重复调用

- 实现了异步队列服务，可以将所有要异步执行的各个方法直接加入到队列，由工作服务进程负责这些方法的统一执行

- 所有任务执行情况均可监控，有命令行脚本工具可以查询当前各队列的情况，更值得一提的是该模块提供一个web监控界面来查看当前所有队列以及各队列的任务执行情况，对于失败的任务，可以直接查看其在执行时的错误跟踪堆栈信息

#### :fa-book: 概念定义
- Job：任务，要执行的function
- Worker：工作进程，实际执行队列的任务的进程
- Queue：队列，任务的集合

---
#### :fa-qrcode: 接口描述

##### 扩展的队列数据结构

- 基本操作
定义一个队列结构
```python
>>> queue = HotQueue("myqueue")
```
入队操作(支持python内置的几种数据类型)
```python
>>> queue.put("my message")
>>> queue.put({'name': "Richard Henry", 'eyes': "blue"})
```
出队操作
```python
>>> queue.get()
"my message"
>>> queue.get()
{'name': 'Richard Henry', 'eyes': 'blue'}
```

队列迭代
```python
>>> for item in queue.consume():
...     print item
```

- 详细接口

见 **hotqueue.py** 源码

##### 方法参数级调用队列
- 基本示例

定义调用队列
```python
SquareQ = HotQueue("my_queue")
@SquareQ.worker
def square(num):
    print num * num
```

队列操作
```python
SquareQ.put(2)
SquareQ.put(3)
SquareQ.put(4)
```
工作进程执行
```python
>>> square()
4
9
16
```
当执行到为空时工作进程进入等待状态，直到有新的参数put到队列

##### 异步队列服务

定义队列
```python
Q = Queue('queuename')

def foo(s):
	print 'foo',s
	return s

def boo(s):
	print 'boo',s
	return s
```

> **Tip:** 其中queuename为队列名称，在zkeco实际的应用中我们可以分别为数据上传，命令下发和自动计算分别命名为zksaas_adms、writedata和autocalculate三个队列，这样方便在web监控界面上分类查询

入队操作
```python
Q.put(foo,'one')
Q.put(boo,'two')
Q.put(foo,'three')
```
运行工作进程服务
```
# python manage.py runQueue
```
> **备注:**
 
> - 入队操作时的function必须为非绑定方法，参数必须为可序列化的，比如模型实例object就不能作为参数序列化。

#### :fa-file-text: 说明
1.	执行失败的任务会转入failed 队列中
2.	消息数据存储方式：目前支持数据库存储和Redis高速存储两种，暂用mysql方式
3.	该“异步任务队列”支持执行异常消息的记录保持和重新入队执行功能，默认执行成功的消息记录直接删除

原开源项目地址：<https://github.com/nvie/rq>

Information

Tag

官方订阅号

分类

About This Entry

About This Blog

最近发布

Friend Links