Direct to a specific Rabbitmq queue from celery

user_11828333 0
Viewed 330

I have a RabbitMQ server running and I'm trying to execute a task in it. I use celery for this and want to direct to a specific queue with specific exchange.

celery code

broker_uri='amqp://xxxx:xxxx@xxxx:5672;amqp://xxxx:xxxx@xxxx:5672;amqp://xxxx:xxxx@xxxx:5672/'
backend_uri="mongodb+srv://xxxxxx.mongodb.net/celery_test?retryWrites=true&w=majority"

app = Celery('TestApp', broker=broker_uri,backend=backend_uri)

@app.task
def reverse(text):
    sleep(10)
    return text[:-1]

when I run this, it automatically access a queue and an exchange which I haven't defined. How do I change this?

execution result

- *** --- * --- .> concurrency: 3 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
django-celery celery rabbitmq
2 Answers

you can define queue by using kombu.

from kombu import Queue, Exchange
sample_queue = Queue(name='sample_queue', exchange=Exchange('sample_exchange', <set exchange type>'topic', durable=True),
                routing_key='sample')

To route a task to a queue set task_routes

task_routes = {
    'reverse': {
        'queue': 'sample_queue',
        'routing_key': 'sample',
    },
}

check documentation exchanges-queues-and-routing-keys

user_94093240

Apparently, queue and exchange names can be defined this way,

app.conf.task_default_queue='queue'
app.conf.task_default_exchange='exchange'
app.conf.task_default_routing_key='key'

Seems like it's not possible to specify the types of queue and exchange.

user_118283330
Related