сельдерей не работает с глобальными переменными

Question

from celery import Celery 

app = Celery('tasks', backend='amqp://guest@localhost//', broker='amqp://guest@localhost//') 

a_num = 0 

@app.task 
def addone(): 
    global a_num 
    a_num = a_num + 1 
    return a_num 

это код, который я использовал для проверки сельдерея. Я надеюсь, что каждый раз, когда я использую addone(), возвращаемое значение должно увеличиваться. Но это всегда 1 почему ???

Результатов

python 
>> from tasks import addone 
>> r = addone.delay() 
>> r.get() 
    1 
>> r = addone.delay() 
>> r.get() 
    1 
>> r = addone.delay() 
>> r.get() 
    1 

Answer 1

По умолчанию, когда рабочий запущен Celery запускает его с параллелизмом 4, а это означает, что 4 процесса начали обрабатывать запросы задачи. (Кроме того, это процесс, который контролирует другие процессы.) Я не знаю, какой алгоритм используется для назначения запросов задачи для процессов, запущенных для рабочего, но в конечном итоге, если вы выполните достаточно addone.delay().get(), вы увидите, что число больше 1 Что происходит, так это то, что каждый процесс (не каждый задание) получает свою собственную копию a_num. Когда я пробую это здесь, мое пятое исполнение addone.delay().get() возвращает 2.

Вы можете заставить число увеличивать каждый раз, запустив своего работника с помощью одного процесса для обработки запросов. (например, celery -A tasks worker -c1) Однако, если вы когда-нибудь перезагрузите своего работника, нумерация будет сброшена на 0. Более того, я бы не стал разрабатывать код, который работает только в том случае, если количество обработчиков запросов будет равным 1. Один день по дороге коллега решает, что несколько процессов должны обрабатывать запросы для задач, а затем перерывы. (Предупреждения в отношении большого жира в комментариях в коде могут делать только так.)

В конце дня такое состояние должно быть общим в кеше, например Redis, или базе данных, используемой в качестве кеша, которая будет работать для кода в вашем вопросе.

Однако в комментарии Вы писали:

Давайте посмотрим, я хочу использовать задачу отправить что-то. Вместо того чтобы подключаться каждый раз в задаче, я хочу поделиться глобальным соединением.

Хранение соединения в кеше не будет работать. Я бы решительно выступал за каждый процесс, который Сельдерей начинает использовать свою собственную связь, а не пытаться разделить ее между процессами. Соединение не обязательно должно быть открыто с каждым новым запросом задачи. Он открывается один раз на процесс, и затем каждый запрос задачи, обслуживаемый этим процессом, повторно использует соединение.

Во многих случаях попытка совместного использования одного и того же соединения между процессами (например, посредством обмена виртуальной памятью через fork) будет работать в любом случае. Соединения часто содержат состояние с ними (например, подключено ли соединение с базой данных в режиме автосохранения). Если две части кода ожидают, что соединение будет в разных состояниях, код будет работать непоследовательно.

Answer 2

Задача будет работать асинхронно так каждый раз, когда он начинает новый a_num задания будет установлен на 0. Они запускаются в виде отдельных экземпляров.

Если вы хотите работать со значениями, я предлагаю хранилище значений или базу данных какого-то рода.