Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
# 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 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 =
while 1:
i += 1
# print i
except KeyboardInterrupt:
stop =
delta = stop - start
conn_per_sec = float(i / (delta.days*24*3600 + delta.seconds))
print settings.BROKER_URL
print i, "tasks created for", delta, "seconds"
print "Tasks created per sec:", conn_per_sec
# 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.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
def debug_task(self):
print('Request: {0!r}'.format(self.request))

This comment has been minimized.

Copy link
Owner Author

atodorov commented Nov 5, 2014 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. defines the sample task that is used for profiling.

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

$ ./ celery_load_test
6754 tasks created for 0:00:02.179551 seconds
Tasks created per sec: 3377.0
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.