task.py 0.3

task.py

在后台异步运行任务，并与它们进行控制/通信。

啊！[生成状态]（https://api.travis-ci.org/abhinavsingh/task.py.png）

安装

pip install task.py

用法

task.py的基本用例是异步运行某个工作单元。 （在单独的线程或进程中）同时能够控制和 与工作单位沟通。示例：

让我们定义一个简单的任务

• 需要一些参数
• 在发出停止信号之前做一些工作（睡眠）
• 返回“完成”。作为最终结果

` >>> def simple_task(t): # will receive a Task instance as first argument ...     assert t.args[0] == 'hello' ...     assert t.kwargs['world'] == 'world' ...     while not t.stopped(): ...         time.sleep(0.1) ...     return 'done.' `

接下来，创建一个以简单任务为单位的实例。 另外指定睡眠任务所需的参数：

` >>> t1 = Task(simple_task, 'hello', world='world') >>> >>> # start our task asynchronously >>> t1.start() >>> >>> # after 1 second signal our task to stop >>> time.sleep(1) >>> t1.stop() >>> >>> t1.done True >>> >>> t1.result 'done.' >>> >>> t1.exception == None True `

让我们创建一个新的任务实例，这次跳过预期的参数：

` >>> t2 = Task(simple_task) >>> >>> # start our task asynchronously >>> t2.start() >>> >>> t2.done True >>> >>> # No result found >>> t2.result == None True >>> >>> # our task raised an exception >>> t2.exception IndexError('tuple index out of range',) `

与任务通信

任务提供一个通信队列，外部线程可以通过该队列发送数据 致工作执行单位。示例：

让我们定义一个睡眠任务，它将接收睡眠时间 通过通信队列：

` >>> def sleep_task(t): ...     secs = t.recv() # recv API, default blocks until we have some data ...     time.sleep(int(secs)) ... >>> t3 = Task(sleep_task) >>> t3.start() >>> >>> # our task is waiting >>> t3.done False >>> >>> # send 1 sec sleep time >>> t3.send(1) >>> >>> # alright, done >>> t3.done True `

控制器

• [等待完成]（https://github.com/abhinavsingh/task.py/blob/master/examples/wait_until_done_task.py）
• [红地毯]（https://github.com/abhinavsingh/task.py/blob/master/examples/redis_brpop_task.py）
• [redis_pubsub]（https://github.com/abhinavsingh/task.py/blob/master/examples/redis_pubsub_task.py）

任务和芹菜

• [芹菜任务]（https://github.com/abhinavsingh/task.py/blob/master/examples/celery_task.py）