using app.worker_main method (v3.1.12): ± cat start_celery.py #!/usr/bin/python from myapp import app if __name__ == “__main__”: argv = [ ‘worker’, ‘–loglevel=DEBUG’, ] app.worker_main(argv)
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
According to https://docs.celeryq.dev/en/stable/userguide/workers.html#restarting-the-worker you can restart a worker by sending a HUP signal ps auxww | grep celeryd | grep -v “grep” | awk ‘{print $2}’ | xargs kill -HUP
Unacknowledged messages are those which have been delivered across the network to a consumer but have not yet been ack’ed or rejected — but that consumer hasn’t yet closed the channel or connection over which it originally received them. Therefore the broker can’t figure out if the consumer is just taking a long time to … Read more
You can set task time limits (hard and/or soft) either while defining a task or while calling. from celery.exceptions import SoftTimeLimitExceeded @celery.task(time_limit=20) def mytask(): try: return do_work() except SoftTimeLimitExceeded: cleanup_in_a_hurry() or mytask.apply_async(args=[], kwargs={}, time_limit=30, soft_time_limit=10)
Prefetching can improve the performance. Workers don’t need to wait for the next message from a broker to process. Communicating with a broker once and processing a lot of messages gives a performance gain. Getting a message from a broker (even from a local one) is expensive compared to the local memory access. Workers are … Read more