dotask 0.3.0

发布订阅注解方式

版本说明

0.1.0

• 在方法上添加@task就可以直接开启一个线程执行任务
• 常用参数：
  • 生产者: role="producer",topic="自定义"
  • 消费者: role="consumer",subscribe="自定义",publish_after="自定义"
• 说明:
  • role: 用于区分上下游 producer/consumer
  • topic: 发布的主题
  • subscribe: 对哪些主题感兴趣
  • publish_after: 消费完后继续发布新主题
• 特别的：方法的返回值可以进行传递,若没有返回值则不会发布新主题（只针对消费者和消费者之间）

0.2.x

1. dotask暴露topic_manager=>可以更好的管理生产消费者
2. 迁移logger至util包下
3. 添加本地调用shell工具函数->可以很好的适配@task消费者
   1. max_concurrent:控制并发执行的shell数量
   2. 引入dotask.Shell 调用Shell(max_concurrent=?).local_shell_execute(cmd,callback)来使用

0.3.x

1. 添加令牌桶,通过from dotask.util import TokenBucket使用,可以更好的限制生产消费速率

• 参数说明
  • TokenBucket(capacity=1, rate=0),capacity:令牌桶容量,rate:每秒生成速率
  • get_token(block=True),block:决定令牌桶无令牌时是阻塞还是返回false

快速开始

安装

pip install dotask

示例

1. 简单的生产发布模式

from dotask import task  
from dotask.util import logger  
import random  
  
if __name__ == '__main__':  
  
    @task(role="producer",topic="scan")  
    def a():  
        return random.randint(1,10)  
  
  
    @task(role="consumer",subscribe="scan",publish_after="sayHi")  
    def b(data):  
        if data >5:  
            return data  
        logger.warning(f"本次生成数字:{data},不会继续发布sayHi主题")  
  
    @task(role="consumer",subscribe="sayHi")  
    def c(data):  
        logger.debug(f"c触发")  
        logger.info(f"消费:{data}")
//=====================================================================
2026-02-08 17:30:37 - INFO - 发布主题-[scan]:5
2026-02-08 17:30:37 - WARNING - 本次生成数字:5,不会继续发布sayHi主题
2026-02-08 17:30:38 - INFO - 发布主题-[scan]:8
2026-02-08 17:30:38 - INFO - 发布主题-[sayHi]:8
2026-02-08 17:30:38 - DEBUG - c触发
2026-02-08 17:30:38 - INFO - 消费:8
2026-02-08 17:30:38 - INFO - 发布主题-[scan]:6
2026-02-08 17:30:38 - INFO - 发布主题-[sayHi]:6
2026-02-08 17:30:38 - DEBUG - c触发
2026-02-08 17:30:38 - INFO - 消费:6
//======================================================================