Excluindo todas as tarefas pendentes no aipo / rabbitmq

199

Como posso excluir todas as tarefas pendentes sem saber o task_idpara cada tarefa?

nabizan
fonte

Respostas:

296

Dos documentos :

$ celery -A proj purge

ou

from proj.celery import app
app.control.purge()

(EDIT: Atualizado com o método atual.)

Philip Southam
fonte
56
Ou, do Django, para o aipo 3.0+: manage.py celery purge( celeryctlagora está obsoleto e desapareceu na versão 3.1).
Henrik Heimbuerger
3
Encontrei esta resposta procurando como fazer isso com um back-end de redis. O melhor método que encontrei foi o redis-cli KEYS "celery*" | xargs redis-cli DELque funcionou para mim. Isso eliminará todas as tarefas armazenadas no back-end do redis que você está usando.
Melignus
1
Como posso fazer isso no aipo 3.0?
luistm
2
Para mim, era simplesmente celery purge(dentro do ambiente virtual relevante). Opa! - há uma resposta com a mesma abaixo ... stackoverflow.com/a/20404976/1213425
Erve1879
Para o Celery 4.0+ em combinação com o Django, é novamente este comando, onde o argumento to -Aé o aplicativo Django onde celery.pyestá localizado.
Gitaarik
120

Para o aipo 3.0 ou superior:

$ celery purge

Para limpar uma fila específica:

$ celery -Q queue_name purge
ToonAlfrink
fonte
9
Se você receber erros de conexão, especifique o aplicativo, por exemplo celery -A proj purge.
Kamil Sindi
25

Para o aipo 2.xe 3.x:

Ao usar o trabalhador com o parâmetro -Q para definir filas, por exemplo

celery worker -Q queue1,queue2,queue3

então celery purgenão funcionará, porque você não pode passar os parâmetros da fila para ele. Ele excluirá apenas a fila padrão. A solução é iniciar seus trabalhadores com --purgeparâmetros como este:

celery worker -Q queue1,queue2,queue3 --purge

No entanto, isso executará o trabalhador.

Outra opção é usar o subcomando amqp do aipo

celery amqp queue.delete queue1
celery amqp queue.delete queue2
celery amqp queue.delete queue3
smido
fonte
Sim, é para versões mais antigas (2.xe talvez 3.x) do aipo. Não consigo editar a resposta
smido
9

Descobri que celery purgeisso não funciona para minha configuração de aipo mais complexa. Eu uso várias filas nomeadas para diferentes propósitos:

$ sudo rabbitmqctl list_queues -p celery name messages consumers
Listing queues ...  # Output sorted, whitespaced for readability
celery                                          0   2
[email protected]                      0   1
[email protected]                      0   1
apns                                            0   1
[email protected]                        0   1
analytics                                       1   1
[email protected]                   0   1
bcast.361093f1-de68-46c5-adff-d49ea8f164c0      0   1
bcast.a53632b0-c8b8-46d9-bd59-364afe9998c1      0   1
celeryev.c27b070d-b07e-4e37-9dca-dbb45d03fd54   0   1
celeryev.c66a9bed-84bd-40b0-8fe7-4e4d0c002866   0   1
celeryev.b490f71a-be1a-4cd8-ae17-06a713cc2a99   0   1
celeryev.9d023165-ab4a-42cb-86f8-90294b80bd1e   0   1

A primeira coluna é o nome da fila, a segunda é o número de mensagens aguardando na fila e a terceira é o número de ouvintes para essa fila. As filas são:

  • aipo - Fila para tarefas padrão e idempotentes do aipo
  • apns - Fila para tarefas do Apple Push Notification Service, não tão idempotente
  • analytics - Fila para análises noturnas de longa duração
  • * .pidbox - Fila para comandos de trabalhadores, como desligamento e redefinição, um por trabalhador (2 trabalhadores de aipo, um trabalhador de apns, um trabalhador de análise)
  • bcast. * - Filas de transmissão, para enviar mensagens a todos os trabalhadores que estão ouvindo uma fila (e não apenas o primeiro a pegá-la)
  • celeryev. * - Filas de eventos do aipo, para relatórios de análise de tarefas

A tarefa de análise é uma tarefa de força bruta que funcionou muito bem em pequenos conjuntos de dados, mas agora leva mais de 24 horas para processar. Ocasionalmente, algo sai errado e fica parado aguardando no banco de dados. Ele precisa ser reescrito, mas até então, quando ele fica preso, eu mato a tarefa, esvazio a fila e tento novamente. Detecto "obstrução" observando a contagem de mensagens da fila de análise, que deve ser 0 (análise concluída) ou 1 (aguardando a conclusão da análise da noite passada). 2 ou superior é ruim e recebo um email.

celery purge oferece para apagar tarefas de uma das filas de transmissão e não vejo uma opção para escolher uma fila nomeada diferente.

Aqui está o meu processo:

$ sudo /etc/init.d/celeryd stop  # Wait for analytics task to be last one, Ctrl-C
$ ps -ef | grep analytics  # Get the PID of the worker, not the root PID reported by celery
$ sudo kill <PID>
$ sudo /etc/init.d/celeryd stop  # Confim dead
$ python manage.py celery amqp queue.purge analytics
$ sudo rabbitmqctl list_queues -p celery name messages consumers  # Confirm messages is 0
$ sudo /etc/init.d/celeryd start
jwhitlock
fonte
Não é uma resposta, é? Muito informativo no entanto!
01
4
celeryctl purgenão funcionou com filas nomeadas. python manage.py celery amqp queue.purge <queue_name>fez. Eu acho que o contexto é útil para aqueles com configurações complexas, para que eles possam descobrir o que precisam fazer se celeryctl purgefalharem.
precisa saber é
Não consigo encontrar manage.pyno meu Aipo 3.1.17, o arquivo foi removido ou está sendo novo? Eu encontrei o que parece com a interface correspondente ( queue.purge) */bin/amqp.py, no entanto. Mas, depois de tentar correlacionar o conteúdo do arquivo com a documentação, devo admitir lamentavelmente que o Celery é lamentavelmente indocumentado e também um trabalho muito complicado, pelo menos julgando-o pelo seu código-fonte.
01
manage.pyé o script de gerenciamento do Django e manage.py celeryexecuta o aipo após carregar a configuração das configurações do Django. Eu não usei fora aipo de Django, mas o incluído celerycomando pode ser o que você está procurando: celery.readthedocs.org/en/latest/userguide/monitoring.html
jwhitlock
5

Em Aipo 3+

http://docs.celeryproject.org/en/3.1/faq.html#how-do-i-purge-all-waiting-tasks

CLI

Limpar fila nomeada:

 celery -A proj amqp queue.purge <queue name>

Limpar fila configurada

celery -A proj purge

Eu limpei as mensagens, mas ainda há mensagens na fila? Resposta: As tarefas são confirmadas (removidas da fila) assim que são realmente executadas. Depois que o trabalhador recebe uma tarefa, levará algum tempo até que seja realmente executado, especialmente se houver muitas tarefas aguardando execução. As mensagens que não são confirmadas são mantidas pelo trabalhador até que ele feche a conexão com o broker (servidor AMQP). Quando essa conexão é fechada (por exemplo, porque o trabalhador foi interrompido), as tarefas serão reenviadas pelo broker para o próximo trabalhador disponível (ou o mesmo trabalhador quando tiver sido reiniciado), para limpar adequadamente a fila de tarefas em espera que você precisa parar todos os trabalhadores e, em seguida, limpe as tarefas usando celery.control.purge ().

Portanto, para limpar toda a fila, os trabalhadores devem ser parados.

oneklc
fonte
5

Se você deseja remover todas as tarefas pendentes e também as ativas e reservadas para interromper completamente o aipo, é isso que funcionou para mim:

from proj.celery import app
from celery.task.control import inspect, revoke

# remove pending tasks
app.control.purge()

# remove active tasks
i = inspect()
jobs = i.active()
for hostname in jobs:
    tasks = jobs[hostname]
    for task in tasks:
        revoke(task['id'], terminate=True)

# remove reserved tasks
jobs = i.reserved()
for hostname in jobs:
    tasks = jobs[hostname]
    for task in tasks:
        revoke(task['id'], terminate=True)
kahlo
fonte
2

1. Para limpar adequadamente a fila de tarefas em espera, é necessário parar todos os trabalhadores ( http://celery.readthedocs.io/en/latest/faq.html#i-ve-purged-messages-but-there-are- mensagens estáticas deixadas na fila ):

$ sudo rabbitmqctl stop

ou (caso o RabbitMQ / message broker seja gerenciado pelo Supervisor):

$ sudo supervisorctl stop all

2. ... e limpe as tarefas de uma fila específica:

$ cd <source_dir>
$ celery amqp queue.purge <queue name>

3. Inicie o RabbitMQ:

$ sudo rabbitmqctl start

ou (caso o RabbitMQ seja gerenciado pelo Supervisor):

$ sudo supervisorctl start all
Ukr
fonte
2

comando aipo 4+ aipo limpeza para limpar todas as filas de tarefas configuradas

celery -A *APPNAME* purge

programaticamente:

from proj.celery import app
app.control.purge()

todas as tarefas pendentes serão eliminadas. Referencia: celerydoc

Roshan Bagdiya
fonte