Как сделать запланированное задание celery, которое запускает несколько заданий с разными параметрами?

У меня есть задача сельдерея, такая как:

# Inside tasks.py

from .models import Animal

@shared_task
def process_animals():
    animals = Animal.ojbects.filter(age=5)

    for animal in animals: 
        utils.register_animal(animal)

У меня такой график: 

# Inside celery.py

from celery import Celery
from celery.schedules import crontab

app = Celery("core")

app.conf.beat_schedule = {
    "runs-every-1-min": {
        "task": "my_app.core.tasks.process_animals",
        "schedule": crontab(),
    },
}

Нет причин обрабатывать Animals по одному, они все независимы. Возможно ли "многопроцессорная" или "многозадачная" обработка этого списка?

Следующий код создаст поток для каждого животного и будет обрабатывать его независимо друг от друга.

#tasks.py
from .models import Animal
from threading import Thread

@shared_task
def process_animals():
    animals = Animal.ojbects.filter(age=5)
    for animal in animals:
        t = Thread(target=utils.register_animal, args=(animal,))
        t.start()

А если вы хотите сделать то же самое на одном ядре (если обработка меньше):

#tasks.py
from .models import Animal
from multiprocessing.dummy import Pool as ThreadPool
    
@shared_task
def process_animals():
    animals = Animal.ojbects.filter(age=5)
    pool = ThreadPool()

    results = pool.map(utils.register_animal, animals)

    pool.close()
    pool.join()
Вернуться на верх