Я использую celery+rabbitmq. Я не могу найти удобный способ очистить очередь в celery+rabbitmq. Я делаю это с удалением и созданием vhost.
rabbitmqctl delete_vhost <vhostpath>
rabbitmqctl add_vhost <vhostpath>
Это предпочтительный способ очистить очередь сельдерея?
Я не совсем уверен, как работает сельдерей, но я подозреваю, что вы хотите очистить очередь RabbitMQ (в настоящее время вы имитируете это, удаляя очереди и заставляя сельдерей воссоздавать их).
Вы можете установить плагин управления RabbitMQ. Его WebUI позволит вам очистить необходимую очередь. Это также должно сказать вам, к какой очереди вы стремитесь, поэтому вам не нужно будет удалять все.
Как только вы узнаете, какая это очередь, вы можете очистить ее программно. Например, используя py-amqplib, вы должны сделать что-то вроде:
from amqplib import client_0_8 as amqp
conn = amqp.Connection(host="localhost:5672", userid="guest", password="guest", virtual_host="/", insist=False)
conn = conn.channel()
conn.queue_purge("the-target-queue")
Хотя, вероятно, есть лучший способ сделать это.
Если вы столкнулись с этой проблемой из-за того, что использовали rabbitmq для обработки результатов, и в результате у вас слишком много очередей, я бы предложил использовать другой результат обработки (redis или mongodb).
Это один хорошо известный недостаток сельдерея. Он создаст отдельную очередь для каждого результата, если вы используете amqp для обработки результатов.
Если вы все еще хотите придерживаться amqp в качестве бэкэнда. Он очистится через 24 часа. Однако вы можете установить меньшее значение, используя настройку CELERY_AMQP_TASK_RESULT_EXPIRES.
Если вам нужно удалить ВСЕ элементы в очереди (особенно если список длинный)
1) Сохраняет все элементы в файл
sudo rabbitmqctl list_queues -p /yourvhost name > queues.txt
не забудьте удалить первую и последнюю строки из 'queues.txt'
2) Используйте упомянутый код Python для выполнения задания.
from amqplib import client_0_8 as amqp
conn = amqp.Connection(host="127.0.0.1:5672", userid="guest", password="guest", virtual_host="/yourvhost", insist=False)
conn = conn.channel()
queues = None
with open('queues.txt', 'r') as f:
queues = f.readlines()
for q in queues:
if q:
#print 'deleting %s' % q
conn.queue_purge(q.strip())
print 'purged %d items' % len(queues)
