celery_load_test.py

# place this file under djapp/management/commands/

from djapp.tasks import *
from datetime import datetime
from django.conf import settings
from optparse import make_option
from django.core.management.base import BaseCommand, CommandError


class Command(BaseCommand):
    help = 'Basic load test for Celery backends. Ctrl+C to abort'

    def handle(self, *args, **options):
        i = 0
        start = datetime.now()
        try:
            while 1:
                debug_task.delay()
                i += 1
#                print i
        except KeyboardInterrupt:
            stop = datetime.now()
            delta = stop - start
            conn_per_sec = float(i / (delta.days*24*3600 + delta.seconds))

            print
            print settings.BROKER_URL
            print i, "tasks created for", delta, "seconds"
            print "Tasks created per sec:", conn_per_sec

tasks.py

# place this file under djapp/

from __future__ import absolute_import

import os
from celery import Celery
from django.conf import settings

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'djapp.settings')

app = Celery('djapp')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

celery_load_test.py is a management command (for Django 1.6.X) which creates delayed tasks using Celery and measures how many of them can be created per second. I use it to test performance of various transport backends.

tasks.py defines the sample task that is used for profiling.

To use just run the command and terminate with Ctrl+C:

$ ./manage.py celery_load_test
^C
memory://
6754 tasks created for 0:00:02.179551 seconds
Tasks created per sec: 3377.0