The following seems to have worked for a few people so I’m submitting it as the answer: python manage.py celerybeat –pidfile= –schedule=/var/my_app/celerybeat-schedule –pidfile= (an empty string as the pidfile arg) seems to stop one being created.
A lot has changed since the OP! There is now an option for high-availability aka “mirrored” queues. This goes pretty far toward solving the problem you described. See http://www.rabbitmq.com/ha.html.
Based on celery documentation you can import a structure of celery tasks like this: For example if you have an (imagined) directory tree like this: | |– foo | |– __init__.py | |– tasks.py | |– bar |– __init__.py |– tasks.py Then calling app.autodiscover_tasks([‘foo’, bar’]) will result in the modules foo.tasks and bar.tasks being imported.
You can get this kind of behavior using an intermediate task. Here’s a demonstration of creating a “map” like method that works like you’ve suggested. from celery import task, subtask, group @task def get_list(amount): return [i for i in range(amount)] @task def process_item(item): # do stuff pass @task def dmap(it, callback): # Map a callback … Read more
For Celery if the end point is amqp. Checkout Celery.js Github any node process started as amqp consumer would work fine. For every other self.conf.backend_type types you can have varied consumer. Following example is merely for amqp. One such example. The message below may be the Celery task object. var amqp = require(‘amqp’); var connection … Read more
I believe the recommended way to start the task is as follows. python manage.py celery worker -P gevent –loglevel=INFO Gevent needs to be patched as early as possible.
Celery have specific option -f –logfile which you can use: -f LOGFILE, –logfile=LOGFILE Path to log file. If no logfile is specified, stderr is used. To get information about other options, just use celery worker –help. If just want want celery worker with logging to file, your command may look like this: celery worker -f … Read more
From my testing, it processes multiple queues round-robin style. If I use this test code: from celery import task import time @task def my_task(item_id): time.sleep(0.5) print(‘Processing item “%s”…’ % item_id) def add_items_to_queue(queue_name, items_count): for i in xrange(0, items_count): my_task.apply_async((‘%s-%d’ % (queue_name, i),), queue=queue_name) add_items_to_queue(‘queue1’, 10) add_items_to_queue(‘queue2’, 10) add_items_to_queue(‘queue3’, 5) And start the queue with (using … Read more