Ensuring Fair Processing with Celery - Part II

This article explores task priorities in Celery, building upon the previous post about fair processing. Task priorities offer a way to enhance fairness and efficiency in background processing by assigning different priority levels to tasks based on custom criteria.

Why Task-Level Priority?

Task-level priority provides fine-grained control over task execution without complex implementation. By submitting all tasks to a single queue with assigned priority values, workers can process tasks based on their urgency. This ensures fair handling regardless of submission time.

For example, if one tenant submits 100 tasks and another submits 5 shortly after, task-level priority prevents the second tenant from waiting for all 100 tasks to complete.

This approach dynamically assigns priority based on a tenant's task count. Each tenant's first task starts with high priority, but with every 10 concurrent tasks, the priority decreases. This ensures that tenants with fewer tasks don't experience unnecessary delays.

Implementing Task Priority

First, install Celery and Redis:

pip install celery redis

Configure Celery to use Redis as the broker and enable priority-based task processing:

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",
}

Define a method to calculate dynamic priority using Redis to cache each tenant's task count:

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)

Create a custom task class to decrement the task count upon successful completion:

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)

Trigger tasks for different tenants, ensuring the tenant_id is included in the task's keyword arguments:

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
        )

You can see the full code here.

Start the Celery worker and trigger the tasks:

# Run the worker
celery -A tasks worker --loglevel=info

# Trigger the tasks
python tasks.py

This setup demonstrates how Celery's priority queue, combined with Redis, ensures fair task processing by dynamically adjusting priorities based on tenant activity. Let’s see a simplified output of the worker:

Conclusion

Task-level priority with Celery and Redis provides a robust solution for ensuring fair processing in multi-tenant systems. By dynamically assigning priorities and leveraging a single queue, you can maintain simplicity while meeting business requirements.

There are many ways to implement task-level priority, using RabbitMQ for example is more efficient since it support priority at its core but since we are also using Redis for task counting, it simplifies our overall architecture.

Hope you find this useful and see on the next one!

The above is the detailed content of Ensuring Fair Processing with Celery - Part II. For more information, please follow other related articles on the PHP Chinese website!