Celery进阶_任务优先级分配

Background

书接上回Celery入门，简单使用了Celery的基本功能使我们的任务能够“并发”的跑起来，但是当所有的worker都被占用时我们想让一些任务优先的执行，这个时候我们应该怎么做呢？

• 设置任务的优先级(有时会不生效，具体原因未知)
• 设置消息队列的优先级

PS：本文使用的Celery版本为5.0.2，不同版本会有差异。

参数说明

CELERYD_PREFETCH_MULTIPLIER：
预取消息数量，默认会取4*并发数，在上面的例子中则会最多预取40个消息，
如果设置为1，则表示禁止预取。
CELERY_ACKS_LATE：
默认为FALSE，这个参数的理解需要先了解下rabbitMQ的ACK机制，
简单说就是如果设置True则会在任务执行完成后才会对消息队列发送确认，表明消息已被消费，
如果任务执行中发生了异常情况，未发送确认消息，则消息队列会继续保留此消息，直到下一个worker取走并成功执行。
此时衍生出了一个问题，需要保证此消息是幂等的，
也就是无论执行多少次结果都一样，否则可能会得到一些预料之外的结果。

Code

# @Time : 2022/8/5 10:31
# @Author : 
# @FileName: my_task.py
# @Description:

import time

from celery import Celery, signature, group
from celery.exceptions import TimeoutError
from celery.result import AsyncResult
from kombu import Queue, Exchange

CONFIG = {
    
    # 优先级需设置这些参数
    'CELERY_ACKS_LATE': True,
    'CELERYD_PREFETCH_MULTIPLIER': 2,
    # 设置时区
    'CELERY_TIMEZONE': 'Asia/Shanghai',
    # 默认为true，UTC时区
    'CELERY_ENABLE_UTC': True,
    # broker
    'BROKER_URL': 'redis://10.11.ip.ip:6379/1',
    # backend配置，注意指定redis数据库
    'CELERY_RESULT_BACKEND': 'redis://10.11.ip.ip:6379/2',
    # worker最大并发数
    'CELERYD_CONCURRENCY': 1,
    # 如果不设置，默认是celery队列，此处使用默认的直连交换机，routing_key完全一致才会调度到celery_demo队列
    # 此处注意，元组中只有一个值的话，需要最后加逗号
    'CELERY_QUEUES': (
        Queue("celery_demo", Exchange("celery_demo"), routing_key="celery_demo", priority=3),
        Queue("celery_demo_high", Exchange("celery_demo_high"), routing_key="celery_demo_high", priority=9)
    )
}


app = Celery()
app.config_from_object(CONFIG)


@app.task(name='demo_task')
def demo_task(x, y):
    print(f"这是一个demo任务，睡了2秒，并返回了{
      x}+{
      y}的结果。")
    time.sleep(2)
    return x + y


if __name__ == '__main__':
    # call()
    l1 = list()
    for i in range(1, 50):
        # sig = demo_task.s(i, i)
        # sig.apply_async(priority=9)
        l1.append(signature("demo_task", args=(i, i), app=app))

    task_group = group(l1).apply_async(priority=9)
    # task_group = group(l1).apply_async(queue='celery_demo')
    task_group.get()

    # sig = demo_task.s(3, 3)
    # sig.apply_async(priority=9)
    # sig = demo_task.s(4, 4)
    # sig.apply_async(priority=9)
    # sig = demo_task.s(5, 5)
    # sig.apply_async(priority=9)
    # sig = demo_task.s(6, 6)
    # sig.apply_async(priority=6)
    # sig = demo_task.s(7, 7)
    # sig.apply_async(priority=6)
    # sig = demo_task.s(8, 8)
    # sig.apply_async(priority=6)
    # sig = demo_task.s(9, 9)
    # sig.apply_async(priority=6)
    # sig = demo_task.s(10, 10)
    # sig.apply_async(priority=6)


## 另起一群任务
if __name__ == '__main__':
    # call()
    l2 = list()

    for i in range(101, 120):
        # sig = demo_task.s(i, i)
        # sig.apply_async(priority=6)
        l2.append(signature("demo_task", args=(i, i), app=app))

    task_group = group(l2).apply_async(priority=6)
    task_group.get()

效果

TIPS

官方文档中使用Redis做消息队列时priority数字越大优先级越低，而RabbitMQ正好相反。摊手.jpg

引用

https://docs.celeryq.dev/en/v5.0.2/userguide/calling.html
http://allynh.com/blog/flask-asynchronous-background-tasks-with-celery-and-redis/
https://segmentfault.com/a/1190000039939846
https://juejin.cn/post/6844904133514756109
https://www.celerycn.io/v/4.4.0/yong-hu-zhi-nan/lu-you-ren-wu-routing-tasks