django-celery 创建多个broker队列异步执行任务时指定队列

向右看齐 2022-02-20 05:49 343阅读 0赞

一.这里不再详细述说 django 框架中如何使用celery, 重点放在如何实现创建多个队列, 并指定队列存放异步任务

笔者使用   django-celery==3.2.2 模块, 配置项及配置参数, 启动方式请查看官方文档.

二.在settings.py配置文件中添加配置, 开启多队列(前提是已经配置好启动celery worker的相关配置项, 建议使用多队列, 一个队列对应一个worker, 而不是多个worker对应一个队列)

from kombu import Exchange, Queue
    
    # 定义celery各个队列的名称
    CELERY_QUEUES = (
        Queue("import_task", Exchange("import_task"), routing_key="task_a"),
        Queue("normal_task", Exchange("normal_task"), routing_key="task_b")
    )
    
    CELERY_ROUTES = {
        "tasks.taskA": {"queue": "import_task", "routing_key": "task_a"},
        "tasks.taskB": {"queue": "normal_task", "routing_key": "task_b"}
    }
    
    # 注意: 使用 redis 作为 broker 时, 队列名称,Exchenge名称,queue名称 必须保持一致
    
    """
    这里只定义了两个队列
        import_task: 用来存放需要优先执行的重要任务, 如果队列仍然存在堵塞的情况, 可以根据更小颗粒度划分出更多的队列
        normal_task: 用来存放执行级别较低的任务, 该类型的任务可以允许存在较长时间的延迟
    """

三.指定队列启动worker

进入manage.py 文件所在目录下, 执行以下命令, 开启worker, 因为我使用了django-celery模块, 所以可以使用manage.py 入口文件进行启动:    -Q  queue\_name   指定队列名称

python manage.py celery worker -B -l info -Q import_task
    python manage.py celery worker -B -l info -Q normal_task
    python manage.py celery worker -B -l info -Q celery

如果需要后台运行, 可在命令的最后加上 "&", 如果使用supervisor进行进程管理, 则不可以加上 "&", docker部署请自行参考docker 官方文档对 dockerfile 使用方式的说明.

***注意: 这里添加了一个使用 celery 队列的worker, 因为在进行任务发送时, 如果没有指明队列, 将默认发送至队列名称为celery的队列中.***

四.调用异步任务

from .tasks import send_emailMes_task
    
    send_emailMes_task.apply_async((params_1, params_2), {"params_3_key": params_3_value}, queue="import_task")

这里笔者导入了自定义的需要异步执行的任务: send\_emailMes\_task

需要注意的是:

使用异步任务对象下的apply\_async(), 而不是delay(), 后者无法指定队列名称

参数:  (params\_1, params\_2),  使用这样的方式传递实参, 需要使用\*agrs接收

参数:  \{"active\_token":token\},  使用这样的方式传递命名参数, 需要使用\*\*kwagrs接收

参数:  queue,  指定将任务发送至那个队列

五.完成以上操作以后就可以进行程序的执行了.