Как настроить Celery с Django, чтобы работники правильно получали задания? - PullRequest
Новая
       13

Как настроить Celery с Django, чтобы работники правильно получали задания?

0 голосов
/ 11 декабря 2019

Я установил RabbitMQ (служба systemd) и настроил Celery в своем проекте Django. Я запустил работников Celery, но в настоящее время моя функция задачи (определенная в моем tasks.py в моем приложении) запущена, но я не вижу никаких «полученных» сообщений в рабочем процессе.

Djangoмакет проекта 

my_project
│
├── my_project
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   ├── wsgi.py
│   ├── __init__.py
│   ├── __pycache__
│   ├── static
│   └── templates
├── my_app
│   ├── admin.py
│   ├── apps.py
│   ├── forms.py
│   ├── __init__.py
│   ├── models.py
│   ├── tasks.py
│   ├── urls.py
│   ├── views.py
│   ├── migrations
│   ├── __pycache__
│   └── templates
└── venv

settings.py 

# /foo/django/my_project/my_project/settings.py

# ...
INSTALLED_APPS = [
    #...
    'myapp",
}    

# ...
CELERY_BROKER_URL = 'amqp://localhost'

__ init __. py 

# /foo/django/my_project/my_project/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

celery.py 

# /foo/django/my_project/my_project/celery.py

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE','my_project.settings')

app = Celery('my_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

tasks.py 

# /foo/django/my_project/my_app/tasks.py

import string

from celery import shared_task
from django.utils.crypto import get_random_string

@shared_task
def my_task(n):
    for c in range(n):
        result = get_random_string(20)
        print('Celery Task [' + str(c + 1) + '/' + str(n) + '] : ' + result)
    return '{} results'.format(n)

views.py 

# /foo/django/my_project/my_app/views.py    

from .tasks import my_task

def task_create(request):
    title = 'Task creation'

    if request.method == 'POST':    
        form = TaskCreateForm(request.POST)
        if form.is_valid():
            my_num = form.cleaned_data.get('my_num')
            # Call to Celery task below:
            my_task(my_num)
            redirect_url_valid = "/tasks/"
            return redirect(redirect_url_valid)
    else:
        form = TaskCreateForm()

    return render(request,'task_create_form.html',{
        'title': title,
        'task_create_form': form
    })

RabbitMQ

Демон rabbitmq-server работает как системная служба со статусом «активный (работает)».

/var/log/rabbitmq/rabbit@xxxxx-sasl.log

=INFO REPORT==== XX-Xxx-2019::xx:xx:05 === accepting AMQP connection <0.2445.0> (127.0.0.1:36890 -> 127.0.0.1:5672)

==> /var/log/rabbitmq/rabbit@xxxxx-sasl.log <==

Рабочие процессы Celery

Запускается из моего венва, запускается вручную (celery -A my_project worker -l debug): 

 -------------- celery@xxxxx.tld v4.3.0 (rhubarb)
---- **** ----- 
--- * ***  * -- Linux-3.10.0-1062.1.1.el7.x86_64-x86_64-with-centos-7.7.1908-Core 2019-12-11 14:51:56
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         my_project:0x7f6e295bb8d0
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 2 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

[xxxx-xx-xx xx:xx:28,768: INFO/MainProcess] celery@xxxxx.tld ready.
[xxxx-xx-xx xx:xx:28,768: DEBUG/MainProcess] basic.qos: prefetch_count->8
[xxxx-xx-xx xx:xx:28,819: DEBUG/MainProcess] heartbeat_tick : for connection d6d0471ec2904f60bece04b2aa38fb0a
[xxxx-xx-xx xx:xx:28,819: DEBUG/MainProcess] heartbeat_tick : Prev sent/recv: 28/145, now - 28/205, monotonic - 18540.910102267, last_heartbeat_sent - 18460.861329738, heartbeat int. - 120.0 for connection d6d0471ec2904f60bece04b2aa38fb0a

Рабочие процессы Celery (ps -ef | grep celery): 

myuser     5321  3506  0 xx:xx pts/3    00:00:01 /foo/django/my_project/venv/bin/python3 /foo/django/my_project/venv/bin/celery -A my_project worker -l debug
myuser     5326  5321  0 xx:xx pts/3    00:00:00 /foo/django/my_project/venv/bin/python3 /foo/django/my_project/venv/bin/celery -A my_project worker -l debug
myuser     5327  5321  0 xx:xx pts/3    00:00:00 /foo/django/my_project/venv/bin/python3 /foo/django/my_project/venv/bin/celery -A my_project worker -l debug

Сервер Django 

[xxxx-xx-xx xx:xx:24] "GET /tasks/create/ HTTP/1.1" 200 27374
[xxxx-xx-xx xx:xx:24] "GET /static/css/common/main.css HTTP/1.1" 304 0
Celery Task [1/3] : kpwkvuir7XTcW8ZezAYT
Celery Task [2/3] : K968U1Xsj40icvgoJ2lL
Celery Task [3/3] : dXYyIa8F32rfttcd94B0
[xxxx-xx-xx xx:xx:32] "POST /tasks/create/ HTTP/1.1" 302 0

Моя функция задачи запускается после проверки формы (см. Вывод сервера Django выше) и правильно печатает ее результат, но я не вижу никаких сообщений «Получено» в работнике Celeryв чем проблема?

...