Entenda o Celery: a biblioteca Python para filas de tarefas

Celery? Salsão? Vamos começar a falar de plantas agora na 4Linux?

Não, não. Outro Celery. A biblioteca do Python!

Aah, sim… É uma biblioteca do Python… É sobre jardinagem?

Olha, eu apreciaria se fosse! Jardinagem é um dos meus interesses, afinal de contas. 🙂

Mas vamos falar sobre Celery. O que é, para que serve, quando usar?

Um pouco sobre a ferramenta

Celery é uma ferramenta que nos possibilita criar filas de tarefas – do inglês task queues. E filas de tarefas nada mais são do que um mecanismo que nos permite distribuir tais tarefas entre threads ou processos diferentes. Ou seja, a execução dessas tarefas ocorre de forma independente do nosso programa e, dessa forma, podemos mandar o computador executar uma operação custosa pelo Celery em, por exemplo, um banco de dados, enquanto o nosso programa prossegue com o roteiro. Mais adiante, quando precisamos do resultado daquela operação, pedimos à task que o dê a nós.

Para quem já conhece ou faz uso de programação assíncrona, o Celery funciona como (mas não necessariamente é) uma abstração dela.

O Celery provê alta disponibilidade, sendo capaz de tentar reconexões em caso de falhas. Além disso, ele é flexível, podendo ser usado por si só ou estendido com serializadores, ferramentas de logging, variados mediadores (brokers) e agendadores. Também é possível integrar a ferramenta com frameworks web como Django, Flask, web2py, Pyramid e outros. Outra coisa que simplifica bastante o desenvolvimento com Celery é que ele não precisa de arquivos de configuração para ser usado: basta executar o worker passando a nossa aplicação e ele consegue autonomamente se configurar. Claro, não significa que setups mais complicados dispensem uma configuração básica.

Como funciona?

Para funcionar, o Celery precisa de pelo menos um mediador (broker) onde informações sobre as tarefas serão enfileiradas. As recomendações primárias presentes em seu site oficial são RabbitMQ e Redis, mas é possível usar até SQLite para desenvolvimento local. O Celery pode funcionar em modo multi-thread ou multi-process, e é possível subir múltiplos workers para lidar com múltiplas tarefas.

Mãos à obra!

Usaremos Docker para subir um container que executará o Redis e usá-lo como broker na nossa aplicação. Também usaremos um ambiente virtual do Python para instalar as nossas dependências. Comecemos por este.

Criei um diretório chamado app. Depois entre nele e crie os arquivos requirements.txt e main.py.

$ mkdir app
$ cd app
$ touch requirements.txt main.py

Em seguida, coloque as dependências redis e celery no arquivo requirements:

$ cat >> requirements.txt <<!
redis
celery
!

E agora crie e ative o ambiente virtual para instalar essas dependências:

$ python3 -m venv .venv
$ . .venv/bin/activate
$ pip install -r requirements.txt

O ambiente Python já está praticamente pronto. Agora suba o Redis via Docker.

$ docker run --name celery-app-redis -dp 6379:6379 redis:alpine

Com o Redis funcionando, você pode escrever o nosso código Python. Vamos abrir o arquivo main.py e criar nossa primeira task.

import celery  # A

REDIS_HOST = "redis://localhost:6379/0"  # B

# C
app = celery.Celery('main',
                    backend=REDIS_HOST,
                    broker=REDIS_HOST)

@app.task  # D
def add(x, y):
    return x + y

if __name__ == "__main__":
    print(add(1, 2))  # E

    # F
    task = add.delay(2, 3)
    print(task.get())

No ponto A temos a instrução de importação do celery; não conseguimos usá-lo antes disso. No ponto B definimos a nossa constante REDIS_HOST, que só aponta para a nossa instância do Redis rodando via Docker. Nós o usamos como tanto backend quanto broker – o broker é o que enfileira as tarefas e o backend é onde os resultados ficam salvos – no ponto C, que é onde instanciamos um objeto Celery. Passamos "main" como primeiro argumento para que o Celery consiga contextualizar as tarefas usando esse nome como namespace. No ponto D definimos a nossa função add como uma task do Celery usando o decorator app.task. (Quem conhece Flask já deve ter reconhecido a similaridade com a criação de rotas.) No ponto E podemos ver como é possível chamar a função normalmente ao apenas escrever os parênteses e passar os argumentos. Finalmente, no ponto F, chamamos o método .delay passando os argumentos da função e atribuindo o resultado, que é um AsyncResult do Celery. Podemos realizar outras operações depois de chamar .delay e, quando quisermos o resultado da tarefa, chamamos .get para obtê-lo. Vamos ver funcionando?

Primeiro, vamos revisar a a estrutura do projeto de teste:

$ tree
app
├── main.py
└── requirements.txt

Ou seja, estamos dentro da pasta app temos os arquivos main.py e requirements.txt, que editamos e usamos anteriormente. Também estamos com o ambiente virtual ativado. Neste ponto, precisamos digitar no terminal:

$ celery -A main worker -l INFO

Em que:

• celery é o executável que rodará nosso processo que cuidará das tasks;
• -A main é a opção que definimos para dizer ao Celery que a -Aplicação se encontra no módulo main;
• worker é o subcomando que diz ao Celery para executar uma única instância de worker, que é uma entidade responsável por monitorar a fila de tarefas e executar quaisquer novas tarefas que nosso programa colocar nela;
• -l INFO é a opção do worker que define o nível de log para INFO (mensagens informacionais).

Ao executar, o Celery  mostra algumas informações de inicialização e diz que está pronto para receber as tasks:

[tasks]
  . main.add

[2021-08-03 15:53:11,670: INFO/MainProcess] Connected to redis://localhost:6379/0
[2021-08-03 15:53:11,677: INFO/MainProcess] mingle: searching for neighbors
[2021-08-03 15:53:12,693: INFO/MainProcess] mingle: all alone
[2021-08-03 15:53:12,703: INFO/MainProcess] celery@4rchlinux ready.

Em outro terminal, também com o ambiente virtual ativado, apenas executamos o nosso programa:

$ python main.py 
3
5

E a saída do Celery:

[2021-08-03 16:01:27,060: INFO/MainProcess] Task main.add[ID_DA_TAREFA] received
[2021-08-03 16:01:27,065: INFO/ForkPoolWorker-4] Task main.add[ID_DA_TAREFA] succeeded in 0.0038968220001152076s: 5

Perceba que, onde está escrito ID_DA_TAREFA, deve haver um ID único no formato UUID que o Celery criou.

Vemos que a primeira saída do nosso programa é 3. Esta é a saída da chamada a add(1, 2) no ponto E do nosso programa. Depois, no ponto F, quando chamamos add.delay(2, 3), o nosso worker do Celery recebe a tarefa e prossegue a executá-la, imprimindo no terminal o resultado. Esse resultado é o mesmo que conseguimos quando chamamos task.get() no nosso programa.

Como mencionamos anteriormente, é possível realizarmos outras operações enquanto as nossas tarefas são executadas pelo Celery. Vamos colocar um “custo” de 5 segundos na nossa função:

from time import sleep

# resto do código aqui...

@app.task
def add(x, y):
    sleep(5)
    return x + y

if __name__ == "__main__":
    task = add.delay(2, 3)
    print("fazendo outras coisas enquanto a tarefa é executada")
    print("custo de 2 segundos")
    sleep(2)
    print("chamando get:")
    print(task.get())

Nota: é necessário reiniciar o worker do Celery para que as alterações sejam aplicadas.

Executamos e vemos que, de fato, o Celery possibilita continuar com o fluxo normal do programa enquanto ele se encarrega de executar as tarefas no background.

A partir disso, é só deixar a imaginação fluir e pensar nos cenários onde é possível ou necessário usar a ferramenta.

Quer saber mais sobre ela? Acessa a documentação oficial! Lá eles explicam de tudo. 🙂

Líder em Treinamento e serviços de Consultoria, Suporte e Implantação para o mundo open source. Conheça nossas soluções:

CURSOSCONSULTORIA