сельдерей бить SpawnPoolWorker

У меня есть функция, которую Celery Beat вызывает каждые 5 секунд, и она добавляет что-то к глобальной переменной. Я ожидаю, что каждые 5 секунд моя функция добавляет элемент в глобальную переменную, но она делает это каждые 20 секунд.

Вот tasks.py:

# tasks.py
from celery import shared_task
from .celeryapp import app
from . import cfg

@shared_task
def update_a_global_list():
    try:
        if cfg.flag:
            cfg.init()
        l = ['first']
        cfg.my_global_var.append(l)
        print("my_global_var: " + str(cfg.my_global_var))
    except Exception as e:
        print(e)

Глобальная переменная находится в файле cfg.py:

# cfg.py
global my_global_var
global flag
flag = True


def init():
    global flag
    flag = False
    global my_global_var
    my_global_var = []
    print('Initialize Step')

Конфигурация celery проекта находится в celeryapp.py:

# celeryapp.py
from __future__ import absolute_import
from celery import Celery
from celery.schedules import crontab

import os

os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')

app = Celery('tasks',
             broker='amqp://shahab_user:pass1234@localhost:5672/shahab_vhost',
             backend='rpc://')

app.conf.beat_schedule = {
    'every-5-seconds': {
        'task': 'send_requests.tasks.update_a_global_list',
        'schedule': 5,
    },
}

Когда я запускаю команду:

сельдерей -A Tasks Worker -l INFO

в одном терминале и выполните команду:

сельдерей -A send_requests.celeryapp beat -l INFO

в другом терминале я вижу эти журналы в рабочем терминале:

[2020-10-07 16:39:48,545: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[c24168e6-49df-4188-be2e-8ea05a563f2a]
[2020-10-07 16:39:48,545: WARNING/SpawnPoolWorker-1] Initialize Step                    <===
[2020-10-07 16:39:48,561: WARNING/SpawnPoolWorker-1] my_global_var: [['first']]         <===
[2020-10-07 16:39:48,670: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_a_global_list[c24168e6-49df-4188-be2e-8ea05a563f2a] succeeded in 0.125s :None
[2020-10-07 16:39:53,440: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[ce0ec733-be6d-4640-950f-a2f47ecf1693]
[2020-10-07 16:39:53,440: WARNING/SpawnPoolWorker-2] Initialize Step                    <===
[2020-10-07 16:39:53,440: WARNING/SpawnPoolWorker-2] my_global_var: [['first']]         <===
[2020-10-07 16:39:53,547: INFO/SpawnPoolWorker-2] Task 
send_requests.tasks.update_a_global_list[ce0ec733-be6d-4640-950f-a2f47ecf1693] succeeded in 0.0940000000409782s: None
[2020-10-07 16:39:58,450: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[fae64a0a-8132-4d88-bcd1-cc59d7e73794]
[2020-10-07 16:39:58,453: WARNING/SpawnPoolWorker-3] Initialize Step                    <===
[2020-10-07 16:39:58,454: WARNING/SpawnPoolWorker-3] my_global_var: [['first']]         <===
[2020-10-07 16:39:58,532: INFO/SpawnPoolWorker-3] Task 
send_requests.tasks.update_a_global_list[fae64a0a-8132-4d88-bcd1-cc59d7e73794] succeeded in 0.0779999999795109s: None
[2020-10-07 16:40:03,453: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[86d92236-aaf4-4c62-81d6-2679eed287b2]
[2020-10-07 16:40:03,453: WARNING/SpawnPoolWorker-4] Initialize Step                    <===
[2020-10-07 16:40:03,453: WARNING/SpawnPoolWorker-4] my_global_var: [['first']]         <===
[2020-10-07 16:40:03,533: INFO/SpawnPoolWorker-4] Task 
send_requests.tasks.update_a_global_list[86d92236-aaf4-4c62-81d6-2679eed287b2] succeeded in 0.0779999999795109s: None
[2020-10-07 16:40:08,467: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[1c9fb4e4-a364-4079-98fa-f6deeb3ad638]
[2020-10-07 16:40:08,469: WARNING/SpawnPoolWorker-1] my_global_var: [['first'], ['first']]         <===
[2020-10-07 16:40:08,472: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_a_global_list[1c9fb4e4-a364-4079-98fa-f6deeb3ad638] succeeded in 0.015999999945051968s: None
[2020-10-07 16:40:13,463: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[654d9da4-2c13-46c3-88fe-7b4e73edc8d0]
[2020-10-07 16:40:13,465: WARNING/SpawnPoolWorker-2] my_global_var: [['first'], ['first']]         <===
[2020-10-07 16:40:13,468: INFO/SpawnPoolWorker-2] Task 
send_requests.tasks.update_a_global_list[654d9da4-2c13-46c3-88fe-7b4e73edc8d0] succeeded in 0.0s:None
[2020-10-07 16:40:18,465: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[8d777ade-870b-4133-9f26-8467e4ca4ba7]
[2020-10-07 16:40:18,467: WARNING/SpawnPoolWorker-3] my_global_var: [['first'], ['first']]         <===
[2020-10-07 16:40:18,470: INFO/SpawnPoolWorker-3] Task 
send_requests.tasks.update_a_global_list[8d777ade-870b-4133-9f26-8467e4ca4ba7] succeeded in 0.0s:None
[2020-10-07 16:40:23,470: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[dc8bfa49-0f3e-48bc-9c4d-33705048ffce]
[2020-10-07 16:40:23,473: WARNING/SpawnPoolWorker-4] my_global_var: [['first'], ['first']]         <===
[2020-10-07 16:40:23,476: INFO/SpawnPoolWorker-4] Task 
send_requests.tasks.update_a_global_list[dc8bfa49-0f3e-48bc-9c4d-33705048ffce] succeeded in 0.0s:None
[2020-10-07 16:40:28,465: INFO/MainProcess] Received task: 
send_requests.tasks.update_a_global_list[bad53432-03cb-4843-b71b-105c8d83c971]
[2020-10-07 16:40:28,467: WARNING/SpawnPoolWorker-1] my_global_var: [['first'], ['first'], ['first']]     <===
[2020-10-07 16:40:28,470: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_a_global_list[bad53432-03cb-4843-b71b-105c8d83c971] succeeded in 0.0s:None

Почему Initialize Step выполняется более одного раза?

Почему у меня разные SpawnPoolWorker и они показывают не то, что я ожидал?

Спасибо за вашу помощь.

[EDIT]: Согласно ответу @DejanLekic, я также использовал кеш Django. Но я получил те же результаты.

На этот раз я написал свою программу таким образом:

@shared_task
def update_global_list():
    try:
        test = []
        l = ['first']
        cached_object = cache.get('my_global_var')
        if cached_object is None:
            cache.set('my_global_var', test)
            cached_object = cache.get('my_global_var')
        cached_object.append(l)
        cache.set('my_global_var', cached_object)
        print("my_global_var : " + str(cache.get('my_global_var')))
    except Exception as e:
        print(e)

И я получил такие результаты:

[2020-10-09 14:11:45,903: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[93264e6e-3cd0-4068-ac9e-98f366cdfd51]
[2020-10-09 14:11:45,907: WARNING/SpawnPoolWorker-1] my_global_var : 
[['first']]
[2020-10-09 14:11:45,963: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_global_list[93264e6e-3cd0-4068-ac9e-98f366cdfd51] 
succeeded in 0.0470000000204891s: None
[2020-10-09 14:11:50,835: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[0673a0a8-8b79-4325-a9aa-015061f76166]
[2020-10-09 14:11:50,838: WARNING/SpawnPoolWorker-2] my_global_var : 
[['first']]
[2020-10-09 14:11:50,887: INFO/SpawnPoolWorker-2] Task 
send_requests.tasks.update_global_list[0673a0a8-8b79-4325-a9aa-015061f76166] 
succeeded in 0.0470000000204891s: None
[2020-10-09 14:11:55,830: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[2c58b9b3-ae37-4b8a-b28f-8e4dce2952aa]
[2020-10-09 14:11:55,834: WARNING/SpawnPoolWorker-3] my_global_var : 
[['first']]
[2020-10-09 14:11:55,884: INFO/SpawnPoolWorker-3] Task 
send_requests.tasks.update_global_list[2c58b9b3-ae37-4b8a-b28f-8e4dce2952aa] 
succeeded in 0.0470000000204891s: None
[2020-10-09 14:12:00,829: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[2fc7a1a9-27e7-41dc-936d-a017d2a283bc]
[2020-10-09 14:12:00,833: WARNING/SpawnPoolWorker-4] my_global_var : 
[['first']]
[2020-10-09 14:12:00,951: INFO/SpawnPoolWorker-4] Task 
send_requests.tasks.update_global_list[2fc7a1a9-27e7-41dc-936d-a017d2a283bc] 
succeeded in 0.125s:None
[2020-10-09 14:12:05,838: INFO/MainProcess] Received task: 
send_requests.tasks.update_global_list[d382a795-8041-4331-a9ca-d4d74b2c8982]
[2020-10-09 14:12:05,840: WARNING/SpawnPoolWorker-1] my_global_var : 
[['first'], ['first']]
[2020-10-09 14:12:05,842: INFO/SpawnPoolWorker-1] Task 
send_requests.tasks.update_global_list[d382a795-8041-4331-a9ca-d4d74b2c8982] 
succeeded in 0.015999999945051968s: None

Похоже, что разные рабочие процессы не синхронизированы друг с другом. Я совершенно сбит с толку. Как я могу их синхронизировать?


python celery celerybeat
person shahab    schedule 07.10.2020    source источник
comment
Кажется, SpawnPoolWorkers работают параллельно. как я могу делать свою работу в серии?   -  person shahab    schedule 07.10.2020
comment
@DejanLekic ты прав. Но мне нужна глобальная переменная для хранения некоторых данных и использования их в разных функциях. Я мог бы использовать базу данных, но она много раз обращается к базе данных и неэффективна.   -  person shahab    schedule 07.10.2020

Ответы (1)


arrow_upward
1
arrow_downward

Использование глобальных переменных в распределенной среде просто вызывает проблемы... Это может работать, если вы используете один рабочий процесс Celery и потоки в качестве типа параллелизма. - Типичным решением для этого является использование сервера кэширования (Redis, memcached или аналогичный).

Почему? Все рабочие процессы будут иметь собственную версию my_global_var, поэтому при выполнении задачи, добавляющей к ней что-то, она изменит my_global_var внутри этого рабочего процесса...

person DejanLekic    schedule 07.10.2020
comment
Похоже, что разные рабочие процессы не синхронизированы друг с другом. Как я могу их синхронизировать? - person shahab; 09.10.2020
comment
Ответ прост - у вас нет. Вместо этого глобальные состояния, которыми вы должны поделиться со всеми воркерами в вашем кластере Celery, должны находиться на кэширующем сервере, таком как Redis или Memcached, как я уже упоминал. - person DejanLekic; 09.10.2020
comment
Фактически. Я работаю над проектом Django со списком объектов. Эти объекты имеют атрибут с именем status. «Статус» динамически меняется, и я отслеживаю изменения по ритму сельдерея. Я мог бы использовать запрос к базе данных, чтобы сказать, что если статус изменился, сохранить его в базе данных. Но он попадает в базу данных так много раз. По вашему совету лучше использовать кэширование. Но я неправильно реализую это? - person shahab; 09.10.2020
comment
Я отредактировал свой вопрос. Я также использую кеширование Django. Есть ли разница между кэшированием Django и Redis или Memcached? - person shahab; 09.10.2020