为什么 RabbitMQ 不在持久队列上持久化消息?
问题描述
我通过 Celery 将 RabbitMQ 与 Django 一起使用.我使用的是最基本的设置:
I am using RabbitMQ with Django through Celery. I am using the most basic setup:
# RabbitMQ connection settings
BROKER_HOST = 'localhost'
BROKER_PORT = '5672'
BROKER_USER = 'guest'
BROKER_PASSWORD = 'guest'
BROKER_VHOST = '/'
我导入了一个 Celery 任务并将其排入队列以在一年后运行.从 iPython 外壳:
I imported a Celery task and queued it to run one year later. From the iPython shell:
In [1]: from apps.test_app.tasks import add
In [2]: dt=datetime.datetime(2012, 2, 18, 10, 00)
In [3]: add.apply_async((10, 6), eta=dt)
DEBUG:amqplib:Start from server, version: 8.0, properties: {u'information': 'Licensed under the MPL. See http://www.rabbitmq.com/', u'product': 'RabbitMQ', u'version': '2.2.0', u'copyright': 'Copyright (C) 2007-2010 LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd.', u'platform': 'Erlang/OTP'}, mechanisms: ['PLAIN', 'AMQPLAIN'], locales: ['en_US']
DEBUG:amqplib:Open OK! known_hosts []
DEBUG:amqplib:using channel_id: 1
DEBUG:amqplib:Channel open
DEBUG:amqplib:Closed channel #1
Out[3]: <AsyncResult: cfc507a1-175f-438e-acea-8c989a120ab3>
RabbitMQ 在 celery 队列中收到这条消息:
RabbitMQ received this message in the celery queue:
$ rabbitmqctl list_queues name messages durable
Listing queues ...
KTMacBook.local.celeryd.pidbox 0 false
celery 1 true
celeryctl_KTMacBook.local 0 true
...done.
然后我通过按 control-C 然后按a"中止来杀死 RabbitMQ.当我再次启动服务器并使用rabbitmqctl检查时,它说芹菜队列中没有消息:
I then killed RabbitMQ by hitting control-C followed by 'a' to abort. When I start the server again and check it with rabbitmqctl, it says that there are no messages in the celery queue:
$ rabbitmqctl list_queues name messages durable
Listing queues ...
celery 0 true
celeryctl_KTMacBook.local 0 true
...done.
芹菜队列是持久的.为什么消息没有持久化?我需要做什么才能使消息持久化?
The celery queue was durable. Why were the messages not persisted? What do I need to do to make the messages persistent?
解决方案
要找出消息 delivery_mode
你可以使用它并查看消息属性:
To find out the messages delivery_mode
you can consume it and look at the message properties:
>>> from tasks import add
>>> add.delay(2, 2)
>>> from celery import current_app
>>> conn = current_app.broker_connection()
>>> consumer = current_app.amqp.get_task_consumer(conn)
>>> messages = []
>>> def callback(body, message):
... messages.append(message)
>>> consumer.register_callback(callback)
>>> consumer.consume()
>>> conn.drain_events(timeout=1)
>>> messages[0].properties
>>> messages[0].properties
{'application_headers': {}, 'delivery_mode': 2, 'content_encoding': u'binary', 'content_type': u'application/x-python-serialize'}
相关文章