В настоящее время я изучаю Django и Celery. Итак, мне нужно добавить функцию в проект Django, и я использую для этого Celery: мне нужно постоянно проверять список на моем Redis сервере.
Для этого я использую Celery, но не знаю, как это правильно реализовать, и выбрал while True:.
После запуска моей задачи с celery -A <my_project> worker -l INFO убить задачу невозможно (не работает CTRL + C), или я не нашел как можно. Я прихожу к выводу, что это не лучший способ делать то, что я хочу.
И еще: я знаю, что расширение django-celery-beat существует, но считаю «странным» открывать другой терминал для запуска этого расширения. Мне уже нужно запустить свой сервер Django, затем мой рабочий (с сельдереем) на двух разных терминалах, мне нужно открыть третий, чтобы запустить запланированное задание, я не знаю, действительно ли это странно, но... . Не знаю, мне кажется это странным, но я не знаю, так ли это на самом деле.
Вот моя (уникальная) задача в app моем проекте django:
from celery import shared_task
import redis, time
@shared_task
def watcher():
client = redis.Redis('localhost', port=6379, decode_responses=True)
queue = 'queue'
lock_key = 'watcher_key'
try:
client.delete(queue)
if client.get(lock_key):
print('The matchmaking is already launch.')
return
client.set(lock_key, 'true')
while True:
in_queue = client.lrange(queue, 0, -1)
print(f'\'{queue}\': {in_queue}')
time.sleep(1)
# don't even know what condition I can put for breaking the loop
except Exception as e:
print(f'ERROR REDIS: {e}')
return
finally:
client.delete(lock_key)
Проблема: я не знаю/не нашел, как я могу постоянно проверять список на моем Redis сервере с помощью ХОРОШЕГО метода. Если django-celery-beat — единственное решение, я воспользуюсь им.
Итак, прошу вашей помощи!
Спасибо всем заранее!






Это верно, на самом деле мы не хотим запускать команду для рабочего процесса сельдерея. Теоретически вы могли бы запустить его в контейнере Docker или Kubernetes, но вы, вероятно, не захотите делать ничего подобного.
Лучший способ запустить работника — настроить расписание, используя администрирование Django по умолчанию. Вам необходимо обязательно перенести модели django-celery-beat, чтобы вы могли настраивать различные типы задач в администрировании.
Чтобы настроить администрацию по умолчанию, следуйте инструкциям на странице 2 руководства . и при необходимости обратитесь к сайту администратора Django документации.
После настройки вы сможете настраивать свои задачи на основе всевозможных событий. По таймерам, определенному времени или даже восходу солнца.
Чтобы убедиться, что функции, которые вы настраиваете в celery.py, можно использовать в администрировании по умолчанию, обязательно добавьте декоратор @shared_task. Вот пример:
# Runs once everyday to check if there are any subscriptions to be updated
# SecureSign.celery.task_check_free_trial
from celery import shared_task
@shared_task(expires=45)
def task_check_free_trial():
print('Checking Free Trial...')
morningCheckFreeTrial()
Чтобы настроить периодическую задачу, войдите в систему и добавьте свою задачу там, где отображается Модель периодических задач, и добавьте желаемый период времени для повторного запуска задачи или выберите «одноразовую задачу». При вводе имени задачи сельдерея обращайтесь к нему следующим образом:
DjangoMainAppNameOrFolderName.CeleryFile.Function
Мне нравится комментировать это прямо над функцией, чтобы вы знали, что вводить в администрирование при настройке задач. Как видите, мое имя:
SecureSign.celery.task_check_free_trial
Удачи, дайте мне знать, если у вас возникнут вопросы!
Привет, не знаю, почему ты не видишь ссылку. По сути, администратора Django по умолчанию можно найти по адресу yoursite.com/admin. Следуйте руководству по Django и настройте свои учетные данные. Это дает вам интерактивный способ создания моделей, которые были перенесены в вашу базу данных MySQL, и это очень удобно. В приложении django_celery_beat есть файл models.py, который необходимо перенести, после чего вы сможете использовать администратора для настройки таких моделей, как PeriodicTask. Ознакомьтесь с разделом руководства «Знакомство с администратором Django», который находится после создания и активации моделей. docs.djangoproject.com/en/5.0/intro/tutorial02
Хорошо, наконец-то я делаю то, что вы мне предлагаете, и это тоже работает! Извините за время, которое я трачу на ответ. Так что большое спасибо за вашу помощь.
@Nimpô Нет проблем, дайте мне знать, если у вас возникнут какие-либо проблемы, и, пожалуйста, проголосуйте, если это вам помогло (:
Привет ! Спасибо за ответ. Я не знаю, что ты хочешь мне объяснить. Я не вижу связи с администратором по умолчанию, я уже администратор по умолчанию в своем проекте. Я даже не знаю, как задача вызывается постоянно. Можете ли вы объяснить больше, что вы хотите мне объяснить, пожалуйста? Я новичок в Джанго, ахах 😅