A Light Experience with Lightweight Message Queue Django-Q

Preface

In a recent project (based on Django), I needed a feature to periodically collect vehicle locations.

This reminded me of the OneCat project from a few years ago, when I used Celery, a rather heavy component

Celery is really too heavy. Later, when I built a WeChat official account scraping platform, I came across Django-RQ and Django-Q. The former is a wrapper around RQ that integrates RQ better with Django; the latter is a brand-new “multiprocessing task queue” component, much lighter than Celery, and it left a pretty good impression on me when I used it.

So for this project I decided to continue using Django-Q for some async operations and scheduled tasks.

About Django-Q

Official intro:

A multiprocessing task queue for Django

Quick Start

Installation

pip install django-q

Add to INSTALLED_APPS

INSTALLED_APPS = (
    # other apps
    'django_q',
)

Database Migration

Because Django-Q stores execution results in the database, you need to run a database migration.

python manage.py migrate

This will create three tables: django_q_ormq, django_q_schedule, and django_q_task

Configuration

Since the project already uses Redis for caching, I used Redis directly as the message queue backend (broker)

Django-Q supports many backends: besides Redis, there are also Disque, IronMQ, Amazon SQS, MongoDB, or Django’s ORM~

Add the following configuration in settings.py:

Q_CLUSTER = {
    'name': 'project_name',
    'workers': 4,
    'recycle': 500,
    'timeout': 60,
    'compress': True,
    'cpu_affinity': 1,
    'save_limit': 250,
    'queue_limit': 500,
    'label': 'Django Q',
    'redis': {
        'host': 127.0.0.1',
        'port': 6379,
        'db': 0,
    }
}

Start the Service

python manage.py qcluster

Done. The message queue service is now up and running

We can add async tasks or scheduled tasks

Async Tasks

The simplest way is to use the async_task method it provides to add a new async task to the queue

Let’s write an example: take a number, compute its factorial, then take the square root

import math

def demo_task(number: int):
    return math.sqrt(math.factorial(number))

Start a Task

Then let’s add an async task

from django_q.tasks import async_task, Task

def task_finish(task: Task):
    print(f'任务 {task.name}（ID：{task.id}）完成！')

task_id = async_task(
    demo_task, 10,
    task_name='任务名称',
    hook=task_finish,
)

As you can see, just call the async_task method

The method is defined as

async_task(func: Any, *args: Any, **kwargs: Any)

After passing in the function to run asynchronously, you can pass that function’s arguments right after it, or pass them as kwargs

Both of these work:

• async_task(demo_task, 10)
• async_task(demo_task, number=10)

I personally prefer the first, because Django-Q itself has several named parameters, such as task_name, hook, timeout, and so on. Passing arguments the first way is less likely to conflict with Django-Q’s default named parameters.

Getting Execution Results

There are two ways to get a task’s execution result:

• the admin backend
• use the result method in code

No need to elaborate on the first. After installing the Django-Q component and running migrations, three admin modules are created: Failed tasks, Scheduled tasks, and Successful tasks. As the names suggest, in Failed tasks and Successful tasks you can see the task execution results, i.e. the return value we wrote in demo_task.

The second way, the code is as follows:

from django_q.tasks import result

task_result = result(task_id)

Pass in task_id to query the task’s result. If the task hasn’t finished yet, the result is None

This result method also has a wait parameter to set a wait time, in milliseconds

Completion Callback

In the code above, we also set the hook parameter

Its purpose is to run the task_finish function after the task completes

Inside task_finish you can get task information via the task parameter

That’s it~

Other Parameters of async_task

This method for creating async tasks has many more parameters. The official docs are decent enough. Many of the parameters already exist in the Q_CLUSTER config; setting them in async_task will override the default configuration.

I’ll just copy them over, consider it a translation of the docs~

Besides task_name and hook covered above, there are these parameters:

• group: str the task’s group name
• save configures the storage backend for task results. The docs only have a one-sentence description, so how to actually configure it still needs some research. (I glanced at the source and didn’t get it—dynamic languages are such a pain.)
• timeout: int task timeout in seconds. Recalling the Q_CLUSTER config earlier, it has a timeout setting. Setting this parameter can override that config. If the task runs longer than this time, it will be terminated immediately.
• ack_failures: bool when set to True, also acknowledge failed tasks. This causes failed tasks to be treated as successfully delivered and thus removed from the task queue. Default is False. (Honestly I didn’t quite get what this means.)
• sync: bool when set to True, all async tasks run synchronously. This is useful for testing. Default is False.
• cached this parameter can be set to True, or passed a number representing cache expiration time. According to the docs, async task results are stored in the database. When this parameter is True, results are not written to the database but saved in cache. This is useful when you need to run a large number of async tasks in a short time and don’t need to write results to the database immediately, which can improve performance.
• broker you need to pass an instance of a Broker object, used to control which broker this async task runs on.
• q_options: dict this is the last parameter. I’ll introduce it separately below

The q_options Parameter

From the earlier section on starting tasks, when we start an async task we can pass arguments to the task function via named parameters, for example:

async_task(demo_task, number=10)

But the async_task method itself has many parameters. If a parameter name collides with a parameter of the task demo_task we want to run, those parameters get taken by async_task, and our task demo_task never receives them.

What to do?

The q_options parameter exists to solve this

You can wrap all the parameters meant for async_task in a dict, then pass them via the q_options parameter

Suppose our demo_task looks like this:

def demo_task(number: int, timeout: int):
  ...

Besides the number parameter, it also takes a timeout parameter that collides with async_task’s own parameter. The q_options solution is as follows

opts = {
    'hook': 'hooks.print_result',
    'group': 'math',
    'timeout': 30
}

async_task(demo_task, number=10, timeout=100, q_options=opts)

This way you can both… and…, perfect~

Of course I still recommend passing arguments via *args, so there’s no parameter name collision.

Scheduled Tasks

There are two ways to add scheduled tasks

• add in code
• the admin backend

Adding in Code

Pretty simple, here’s the code

from django_q.tasks import schedule

schedule(
  'demo_task',
  schedule_type=Schedule.MINUTES,
  minutes=1,
  task_name='任务名称',
)

One thing to note: because scheduled tasks are stored in the database after being added

you need to pass the function to execute (including the full package name) as a string

Suppose in our Django project we want to run the demo_task function in the apps/test/tasks.py file

Then we need to pass the full name apps.test.tasks.demo_task

Same when adding in the admin

Interval Settings

Django-Q scheduled tasks have many types:

• once
• every x minutes
• hourly
• daily
• weekly
• monthly
• quarterly
• yearly
• Cron expression

Note that even with a Cron expression, the shortest interval for scheduled tasks is still 1 minute

I didn’t know this at first. I wrote a 15-second task with a Cron expression, but the execution timing was completely off. Then I dug through the GitHub issues and only understood after seeing the author’s reply~

The issue URL: https://github.com/Koed00/django-q/issues/179

The author’s reply:

The current design has a heartbeat of 30 seconds, which means the schedule table can't have schedules below that. Most of this is explained in the architecture docs. Because of the way the internal loop is set up, a resolution under a dozen seconds or so, quickly becomes unreliable.

I always imagined tasks that need accuracy measured in seconds, would use a delayed tasks strategy where a few seconds delay is either added through the broker or inside the task itself.

The problem with all this, is that a task is currently always added to the back of the queue. So even with a 1 second resolution on the schedule, the task still has to wait it's execution time. Which can of course vary wildly depending on the broker type, worker capacity and current workload.

This feels a bit of a letdown. If you need high-frequency tasks, you have no choice but to use Celery

Adding in the Admin Backend

This is even simpler—foolproof, point-and-click

So I’ll skip this part~

Docker Deployment

These days backend services are basically deployed with Docker

To use Django-Q in Docker

we need to spin up another identical container based on the original Django container, then change the entrypoint to the qcluster start command

There’s an issue that discusses this as well: https://github.com/Koed00/django-q/issues/513

Here’s a docker-compose.yml example

version: "3.9"
services:  
  redis:
    image: redis:alpine
    ports:
      - 6379:6379
  web:
    build: .
    command: python manage.py runserver 0.0.0.0:8000
    volumes:
      - .:/code
    ports:
      - "8000:8000"
    depends_on:
      - redis
      - django_q
  django_q:
    build: .
    command: python manage.py qcluster
    volumes:
      - .:/code
    depends_on:
      - redis

A simple example

Other things like environment variables should be set according to your actual situation

Notes:

• Whatever the Django container has (environment variables, dependencies) should be added to Django-Q as well
• After modifying Django project code, if the service auto-restarts via something like uwsgi, note that Django-Q will not restart automatically. You need to manually run docker-compose restart django_q for the code changes to take effect

Other

Command-Line Tools

Django-Q also provides some command-line tools

• Monitor cluster execution: python manage.py qmonitor
• Monitor content: python manage.py qmemory
• View current status: python manage.py qinfo

Besides monitoring via commands, you can also monitor in code. I haven’t needed that yet so I haven’t looked into it. If you need it, you can check the docs directly

Customizing Admin

After installing Django-Q, three menus appear in the admin. Like a regular Django app, these are also registered via admin, so we can re-register these ModelAdmins to customize the admin UI

Here’s a snippet of official code for the failed-tasks interface:

from django_q import models as q_models
from django_q import admin as q_admin

admin.site.unregister([q_models.Failure])
@admin.register(q_models.Failure)
class ChildClassAdmin(q_admin.FailAdmin):
    list_display = (
        'name',
        'func',
        'result',
        'started',
        # add attempt_count to list_display
        'attempt_count'
    )

It's the same as a regular ModelAdmin

We can add search boxes, filter fields, and the like ourselves. Remember to first run admin.site.unregister([q_models.Failure]) to unregister the ModelAdmin object that Django-Q registered itself.

Signals

Django has a built-in signal system. I previously wrote a short article introducing it: Understand Django's Signal Mechanism That Python Backend Developers Must Know in 3 Minutes

Django-Q provides two types of signals:

• Before a task is added to the message queue
• Before a task is executed

Sample code is as follows:

from django.dispatch import receiver
from django_q.signals import pre_enqueue, pre_execute

@receiver(pre_enqueue)
def my_pre_enqueue_callback(sender, task, **kwargs):
    print("Task {} will be enqueued".format(task["name"]))

@receiver(pre_execute)
def my_pre_execute_callback(sender, func, task, **kwargs):
    print("Task {} will be executed by calling {}".format(
          task["name"], func))

If needed, you can register message receivers to do some processing. (Though I haven't used this for now)

Summary

That's it~

The experience of using Django-Q has been pretty good. It's lightweight enough, convenient enough to deploy, and enough to handle most scenarios~

References

• Django-Q official documentation: https://django-q.readthedocs.io/en/latest/index.html
• Django-Q project homepage: https://github.com/Koed00/django-q
• Related issues: https://github.com/Koed00/django-q/issues/179
• Related issues: https://github.com/Koed00/django-q/issues/513