ztq_worker 1.3.2

Zopen Task Queue Worker

简介

ZTQ 队列服务, 分为3个包：ztq_core, ztq_worker, ztq_console。默认使用redis作为队列的后端。

ztq_core

提供一系列的方法把任务push到队列中，由ztq_worker去获取队列任务并且执行。

你可以在这里找到它： http://pypi.python.org/pypi/ztq_core/

ztq_worker:

队列的接收端，以线程为单位阻塞式的去监视一个队列。每一个线程称为Worker
当有任务push到了队列中，相应的Worker会自动pull下来去执行。

你可以在这里找到它： http://pypi.python.org/pypi/ztq_worker/

ztq_console:

对每一个队列的每一个任务执行情况进行监控、下达指令。这个包是可选的

你可以在这里找到它： http://pypi.python.org/pypi/ztq_console/

关于 ZTQ

* 开源, 使用MIT 许可
* 基于Python, 容易使用和修改
* 支持linux 和 windows
* 可靠，可以应付突然断电等情况
* 可管理，自身带有ztq_console 监控后台
* 灵活，可以在不同的机器上运行多个Worker, 并且随时热插拔Worker
* 使用简单

安装

pip install ztq_core
pip install ztq_worker
pip install ztq_console

使用

1. 先定义一个普通的任务

   #  my_send.py

   def send(body):

   print ‘START: ‘, body sleep(5) print ‘END:’, body

   def send2(body):

   print ‘START2’, body raise Exception(‘connection error’)

2. 将普通的任务改成队列任务

   # my_send.py
   
   import time
   from ztq_core import async
   
   @async                            # 使用默认队列default
   def send(body):
          print ‘START: ‘, body
          sleep(5)
          print ‘END:’, body
   
   @async(queue=‘mail’)            # 使用队列mail
   def send(body):
          print ‘START2’, body
          raise Exception(‘connection error’)

3. 运行worker

   # 运行：bin/ztq_worker app.ini
   
   # app.ini 例子, 在ztq_worker 包里面有个config 目录放有app.ini 这个文件
   
   [server]
   host = localhost
   port = 6379
   db = 0
   alias = w01
   active_config = false
   modules = my_send                   # 所有需要import的任务模块，每个一行
   
   [queues]
   default= 0                          # default队列，起1个处理线程
   mail = 0, 0                         # mail队列，起2个处理线程
   
   [log]
   handler_file = ./ztq_worker.log
   level = ERROR

4. 运行

   import ztq_core
   from my_send import send
   
   # 设置 Redis 连接
   ztq_core.setup_redis(‘default’, ‘localhost’,  6379, 0)
   
   send(‘hello, world’)
   
   # 动态指定queue
   send(‘hello world from mail’, ztq_queue=‘mail’)

5. 更详细的测试例子可见ztq_core包下的demo.py

使用更高级的特征

1. 抢占式执行

   # 后插入先执行。如果任务已经在队列，会优先
   send (body, ztq_first=True)

2. 探测任务状态

   # ztq_first存在就优先, ztq_run不存在就运行
   # 返回的是"running" 代表正在运行, 是"queue" 代表正在排队
   # 如果是"error" 代表出错, 是"none" 代表这个任务不在排队，也没在执行
   ping_task(send, body, ztq_first=True, ztq_run=True)

3. 支持事务

   import transaction
   ztq_core.enable_transaction(True)
   send_mail(from1, to1, body1)
   send_mail(from2, to2, body2)
   transaction.commit()
   # 也可以单独关闭事务
   send_mail(from2, to2, body2, ztq_transaction=False)

4. 定时任务

   from ztq_core.async import async
   from ztq_core import redis_wrap
   from ztq_core.cron import has_cron, add_cron_job
   
   @async(queue='clock-0')
   def bgrewriteaof():
       """ 将redis的AOF文件压缩 """
       redis = redis_wrap.get_redis()
       redis.bgrewriteaof()
   
   
   # 如果队列上没有这个定时任务，就加上。自动定时压缩reids
   if not has_cron(bgrewriteaof):
        add_cron({'hour':1}, bgrewriteaof)

5. 任务串行

   from ztq_core import prepare_task
   # 根据(方法，参数)生成一个任务
   callback = prepare_task(send, body)
   # 执行完 send_mail 之后队列会自动将callback 放入指定的队列
   send_mail(body, ztq_callback=callback)

6. 异常处理

   from ztq_core import prepare_task
   
   @async(queue='mail')
   def fail_callback(return_code, return_msg):
          print return_code, return_msg
   
   fcallback = prepare_task(send2)
   
   # 如果任务 send 抛出了任何异常，都会将fcallback 放入指定队列
   send(body, ztq_fcallback=fcallback)

7. 进度回调

   import ztq_worker
   @async(queue='doc2pdf')
   def doc2pdf(filename):
       ...
       # 可被进度回调函数调用
       ztq_worker.report_progress(page=2)
       ...
   
   from ztq_core import prepare_task
   pcallback = prepare_task(send2, body)
   doc2pdf(filename,  ztq_pcallback=pcallback)

8. 批处理

   # 为提升性能，需要多个xapian索引操作，一次性提交数据库
   @async(queue=‘xapian’)
   def index(data):
       pass
   
   def do_commit():
       xapian_conn.commit()
   
   # 每执行20个索引任务之后，一次性提交数据库
   # 不够20个，但队列空的时候，也会提交
   register_batch_queue(‘xapian’, 20, batch_func=do_commit)

1.0dev —

• Initial version