简单可靠的任务管理器。尝试一个类似芹菜的，异步友好的。
aio-task的Python项目详细描述
aio任务
可靠和执行异步友好的异步任务管理器。在
主要特点
- 一个简单的worker接口，用于将协程注册为任务。在
- 生成和获取任务的简单代理接口。在
- 可以在单个程序中设置代理和工作线程，避免外部服务依赖（通过使用虚拟队列和缓存）。在
- 如果工作线程在处理任务时崩溃，则不会丢失它，它将保留在队列中并重新处理，直到工作线程确认它为止。在
- 任务异常不会丢失：您将在任务的结果中检索它们。在
- 支持rabbitmq、redis和sentinel。在
- 可轻松添加新的队列/缓存服务
入门
使用docker-compose -f examples/docker-compose.yml up启动rabbitmq和redis来运行此示例。
安装
pip install aio-task
Worker→运行任务
^{pr2}$
代理→生成任务
importasynciofromaio_taskimportBrokerasyncdefsample_addition():# setup brokerrabbitmq_config={"url":"amqp://guest:guest@localhost:5672","routing_key":"tasks_queue"}redis_config={"address":"redis://localhost"}broker=awaitBroker.create("rabbitmq",rabbitmq_config,"redis",redis_config)# produce tasktask_id=awaitbroker.create_task("addition",{"a":1,"b":2})awaitasyncio.sleep(0.1)# fetch tasktask=awaitbroker.get_task(task_id)print(task)awaitbroker.close()# gracefull shutdownloop=asyncio.get_event_loop()loop.run_until_complete(sample_addition())loop.run_until_complete(broker.close())
pip install aio-task
示例/…中有更多示例。。。在
运行测试
单元测试
pip install -e .[test]
pytest -xvs tests/unit
集成测试
pip install -e .[test]
docker-compose -f tests/integration/compose/docker-compose up -d
pytest -xvs tests/integration
- 项目
标签：
欢迎加入QQ群-->： 979659372
