本文在上一篇有关公平处理的文章的基础上探讨了 celery 中的任务优先级。任务优先级提供了一种通过根据自定义标准为任务分配不同优先级来增强后台处理的公平性和效率的方法。
为什么任务级优先级？
任务级优先级提供对任务执行的细粒度控制，无需复杂的实现。通过将所有任务提交到具有指定优先级值的单个队列，工作人员可以根据任务的紧急程度来处理任务。这确保了公平处理，无论提交时间如何。
例如，如果一个租户提交了 100 个任务，而另一个租户不久后提交了 5 个任务，则任务级别优先级会阻止第二个租户等待所有 100 个任务完成。
这种方法根据租户的任务计数动态分配优先级。 每个租户的第一个任务以高优先级开始，但每有 10 个并发任务，优先级就会降低。这可确保任务较少的租户不会遇到不必要的延误。
实施任务优先级
首先，安装 celery 和 redis：
pip install celery redis
配置 celery 使用 redis 作为代理并启用基于优先级的任务处理：
from celery import celery
app = celery(
"tasks",
broker="redis://localhost:6379/0",
broker_connection_retry_on_startup=true,
)
app.conf.broker_transport_options = {
"priority_steps": list(range(10)),
"sep": ":",
"queue_order_strategy": "priority",
}
定义一个方法来计算动态优先级，使用redis来缓存每个租户的任务计数：
import redis
redis_client = redis.strictredis(host="localhost", port=6379, db=1)
def calculate_priority(tenant_id):
"""
calculate task priority based on the number of tasks for the tenant.
"""
key = f"tenant:{tenant_id}:task_count"
task_count = int(redis_client.get(key) or 0)
return min(10, task_count // 10)
创建自定义任务类以在成功完成后减少任务计数：
from celery import task
class tenantawaretask(task):
def on_success(self, retval, task_id, args, kwargs):
tenant_id = kwargs.get("tenant_id")
if tenant_id:
key = f"tenant:{tenant_id}:task_count"
redis_client.decr(key, 1)
return super().on_success(retval, task_id, args, kwargs)
@app.task(name="tasks.send_email", base=tenantawaretask)
def send_email(tenant_id, task_data):
"""
simulate sending an email.
"""
sleep(1)
key = f"tenant:{tenant_id}:task_count"
task_count = int(redis_client.get(key) or 0)
logger.info("tenant %s tasks: %s", tenant_id, task_count)
为不同租户触发任务，确保tenant_id包含在任务的关键字参数中：
if __name__ == "__main__":
tenant_id = 1
for _ in range(100):
priority = calculate_priority(tenant_id)
key = f"tenant:{tenant_id}:task_count"
redis_client.incr(key, 1)
send_email.apply_async(
kwargs={"tenant_id": tenant_id, "task_data": {}}, priority=priority
)
tenant_id = 2
for _ in range(10):
priority = calculate_priority(tenant_id)
key = f"tenant:{tenant_id}:task_count"
redis_client.incr(key, 1)
send_email.apply_async(
kwargs={"tenant_id": tenant_id, "task_data": {}}, priority=priority
)
您可以在此处查看完整代码。
启动 celery worker 并触发任务：
# Run the worker celery -A tasks worker --loglevel=info # Trigger the tasks python tasks.py
此设置演示了 celery 的优先级队列如何与 redis 相结合，通过根据租户活动动态调整优先级来确保公平的任务处理。让我们看看工作人员的简化输出：
结论
celery 和 redis 的任务级优先级为确保多租户系统中的公平处理提供了强大的解决方案。通过动态分配优先级并利用单个队列，您可以在满足业务需求的同时保持简单性。
实现任务级优先级的方法有很多，例如使用 rabbitmq 效率更高，因为它的核心支持优先级，但由于我们还使用 redis 进行任务计数，因此它简化了我们的整体架构。
希望您觉得这篇文章很有用，并请参阅下一篇！
