django MQ 시리즈 2편 - task queue (2) :Learn Django Celery with RabbitMQ

1. django MQ 시리즈 1편 - task queue (1) : Asynchronous Tasks With Django and Celery

2. django MQ 시리즈 2편 - task queue (2) : Learn Django Celery with RabbitMQ

3. django MQ 시리즈 3편 - task queue (3) : Custom Study Project

4. django MQ 시리즈 4편 - celery with redis for pub/sub

 

해당 포스팅은 django 기반의 기본적인 MQ(message queue)와 pub/sub MQ(message queue) 학습을 위해 작성된다.

관련 자료는 github 저장소에 저장되며 지속적으로 업데이트 된다.

https://github.com/bluebamus/django-mq-series

GitHub - bluebamus/django-mq-series: This repository covers mq and pus/sub learning projects for redis, rabbitmq, kafka, nats, e

해당 프로젝트는 다음 사이트를 기반으로 학습하였다.

https://www.youtube.com/watch?v=fBfzE0yk97k&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20 

* 윈도우에서 rabbitMQ를 설치하는 방법

https://afsdzvcx123.tistory.com/entry/RabbitMQ-RabbitMQ-%EC%84%A4%EC%B9%98

[RabbitMQ] RabbitMQ 설치

1. Erlang 다운 및 설치 : https://www.erlang.org/downloads 

2. rabbitMQ 다운 및 설치 : https://www.rabbitmq.com/install-windows.html

3. 윈도우 검색에서 rabbitMQ service - start / stop 실행

4. RabbitMQ Command Prompt 실행

5. RabbitMQ Server 관리 도구 설치 

rabbitmq-plugins enable rabbitmq_management

6. http://localhost:15672/ 접속

7. guest / guest 로 접속 (localhost로만 가능)

8. ip로 접속할 경우를 위해 계정 생성

rabbitmqctl.bat add_user {ID} {PASSWORD}

rabbitmqctl.bat set_user_tags {ID} administrator

* 관리 도구 설치시 에러 발생

.
.
Plugin configuration unchanged

참고 사이트 : https://velog.io/@toyou4203/%EC%8B%A4%EC%8B%9C%EA%B0%84-%EC%B1%84%ED%8C%85-%EA%B8%B0%EB%8A%A5

Spring Websocket & STOMP - 실시간 채팅 기능

1. RabbitMQ Command Prompt에 명령어 입력

rabbitmq-plugins.bat list

- 구동은 되나 근본적인 원인 파악은 안됨

---

* 해당 프로젝트는 django-celery-results, django-celery-beat 설치 없이 celery 기본 패키지만을 사용하여 학습을 수행함

  celery를 설치하면 기본으 amqp://guest:**@localhost:5672// 서버와 연결을 시도하기에 로컬에서 테스트로 사용시에는

  따로 설정이 필요 없다.

 1. Learn Django Celery with RabbitMQ - Install and create new celery instance, Run a simple task Part 1

https://www.youtube.com/watch?v=fBfzE0yk97k&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20 

github : https://github.com/veryacademy/YT-Django-Celery-Series-Intro-Install-Run-Task

GitHub - veryacademy/YT-Django-Celery-Series-Intro-Install-Run-Task

- celery 실행 방법 (윈도우는 gevent 등의 지원이 필요함)

pip install gevent

python -m celery -A django_celery worker -l info -P gevent

- 학습에서는 shell을 이용한 테스트를 진행함

- 좀더 편한 사용을 위해 shell_plus 설치

# 아래 둘 중 하나를 사용한다.
# 1번 shell_plus
pip install django-extensions
python manage.py shell_plus

# 2번 jupyter notebook
pip install "ipython[notebook]"
python manage.py shell_plus --notebook

# settings.py installed_apps에 'django_extensions' 추가

- 다음 라인별 명령을 입력하고 결과를 확인한다

>>> from app1.tasks import add
>>> add.delay(4,4)
<AsyncResult: f56ddea5-5457-4966-9715-e4985625efd9>

- celery의 로그를 확인한다.

[2023-03-06 01:09:39,048: INFO/MainProcess] Task app1.tasks.add[f56ddea5-5457-4966-9715-e4985625efd9] received
[2023-03-06 01:09:39,057: INFO/MainProcess] Task app1.tasks.add[f56ddea5-5457-4966-9715-e4985625efd9] succeeded in 0.01600000000325963s: 8

- 로그는 received를 수행하고 최종적으로 완료까지의 수행 시간과 결과를 출력한다.

- 추가 학습 :

import datetime
from datetime import timedelta 
now = datetime.datetime.now()

add.delay(4,4) # 기본

add.apply_async((3,3),countdown=10) # 지금부터 10초 뒤 실행
add.apply_async((3,3),countdown=10) # 지금부터 10초 뒤 실행
add.apply_async((3,3),eta=now + timedelta(seconds=10)) # 지금부터 10초 뒤 실행
add.apply_async((3,3),countdown=60, expires=120) # 1분뒤에 실행하고 2분뒤에 파기
add.apply_async((3,3),expires=now + timedelta(days=2)) # 1분뒤에 실행하고 2분뒤에 파기

 2. Learn Django Celery - Sending Emails with Django and Celery - Part-2

https://www.youtube.com/watch?v=7dUxyIp5mMI&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20&index=4 

- github : https://github.com/veryacademy/YT-Django-Celery-Series-Intro-Install-Run-Task

GitHub - veryacademy/YT-Django-Celery-Series-Intro-Install-Run-Task

- debug를 위해 다음코드를 celery.py 제일 하단에 추가한다

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

 

content

- create a form / view

- build new celery task (send emails)

- configure email server - gmail configuration

- test stack

 

bugfix

1. 아래 코드는 소스에 있는 코드이다. 하지만 에러로 출력된다.

from celery.decorators import task

@task(name="send_review_email_task")

2. 아래 코드는 제대로 동작하는 코드이다

from config.celery import app

@app.task(name="send_review_email_task")

원인으로 원본 소스 코드가 아마도 폐기 되었기 때문이라 생각된다.

 

궁금한 항목 추가 학습

@app.task와 @shared_task의 차이점에 대해 궁금해서 검색을 해봤지만 정확한 답을 찾기 어려웠다.

그러다 잘 정리된 영문 페이지를 찾아 정리한다.

https://groups.google.com/g/celery-users/c/XiSDiNjBR6k/m/KUV0sS6hb8IJ

@app.task vs. @shared_task

1. app1 = Celery()을 사용해서 셀러리의 애플리케이션 인스턴스 생성한다. 그렇다면 인스턴스는 여러개 생성할 수 있다는 뜻이다.

2. app1 = Celery(), app2 = Celery() 이렇게 두 개의 인스턴스를 생성했다고 가정하자

3. @app1.task 데코레이터를 사용했다면, 이것은 어떤 인스턴스 설정값을 참조하게 될까?

4. celery를 테스트 하기 위한 코드가 아래와 같이 있다고 하자

assert app1.tasks[test.name]
assert app2.tasks[test.name]

5. 테스트 결과로 test는 항상 app1의 인스턴스를 참조하게 된다. 즉 app2.tasks는 실재로 app1.task의 설정으로 동작한다.

  - 수도 코드로 이해를 하기 때문에 설정이라는 항목의 범위가 어느 정도 인지는 테스트할 필요가 있다.

6. @shared_task의 경우는 생성된 celery 인스턴스의 구분 없이 의존성 없이 독립적으로 동작한다.

7. @shared_task는 라이브러리 혹은 재사용 가능한 앱에 유용하다.

 

 

app.tast(name=''") 항목에 대해 사용하는 예시가 없었다.

해당 항목은 상태를 주기적으로 확인하고 모니터링할 경우 식별할 수 있는 식별자이다.

즉 셀러리 작업 이름이라 생각하면 된다.

 

* 고려 사항

더 이상 google의 stmp 기능이 지원되지 않는다. 

때문에 내가 테스트한 코드에서는 naver의 stmp 기능을 사용했다.

 

테스트 결과

 3. Learn Django Celery - Deploy Redis on a Heroku Service for use in local development

https://www.youtube.com/watch?v=fvYo6LBZUh8&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20&index=5 

해당 클래는 heroku에서 지원하는 redis를 사용하여 celery 작업을 수행한다.

 

part2의 프로젝트에서 settings.py의 변수를 추가한다.

#rabbitMQ
CELERY_BROKER_URL = "amqp://guest:guest@localhost:5672//"

#redis
CELERY_BROKER_URL = "redis://localhost:6379"

localhost 부분이  실재 ip가 되면 된다. rabbitMQ는 계정 정보와 vhost의 정보가 추가될 수 있다.

 4. Learn Django Celery - Intro to scheduling Tasks - Part 3

https://www.youtube.com/watch?v=kWxYPq7Sc8A&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20&index=6 

github : https://github.com/veryacademy/YT-Django-Celery-Series-Part3-Schedule-Tasks

GitHub - veryacademy/YT-Django-Celery-Series-Part3-Schedule-Tasks

content

- install - getting started with Flower (monitoring)

- install django-celery-beat (scheduler)

- explore how to configure scheduled tasks

 

note

1. flower 설치

pip install flower

2. flower 실행

celery -A config flower --port=5555

3. 모니터링 사이트 접속 : http://localhost:5555

4. 모니터링 탭이 안보

   - v1이 되면서 모니터링 탭은 폐기되었다.

   - 사용하고자 한다면 v.0.9.7을 사용하면 된다 하지만 Celery v5와 호환되지 않는다

   - 모니터링 탭이 폐기된 대신 Prometheus와 통합되었다.

   - 해당 정보는 아래 페이지에서 확인할 수 있다.

https://stackoverflow.com/questions/69930545/python-celery-flower-no-monitor-tab-no-real-time-grohics

Python celery flower no monitor tab (no real-time grohics)

   -  Prometheus와 통합은 아래 페이지에서 확인할 수 있다.

https://github.com/mher/flower/blob/master/docs/prometheus-integration.rst

GitHub - mher/flower: Real-time monitor and web admin for Celery distributed task queue

crontab() execute time

- crontab(minute=0, hour=0) Execute daily at midnight.
- crontab(minute=0, hour='*/3') Execute every three hours: midnight, 3am, 6am, 9am, noon, 3pm, 6pm, 9pm.
- crontab(minute=0, hour='0,3,6,9,12,15,18,21') Same as previous.
- crontab(minute='*/15') Execute every 15 minutes.
- crontab(day_of_week='sunday') Execute every minute (!) at Sundays.
- crontab(minute='*', hour='*', day_of_week='sun') Same as previous.
- crontab(minute='*/10', hour='3,17,22', day_of_week='thu,fri') Execute every ten minutes, but only between 3-4 am, 5-6 pm, and 10-11 pm on Thursdays or Fridays.

 

crontab(minute=0, hour='*/2,*/3') Execute every even hour, and every hour divisible by three. This means: at every hour except: 1am, 5am, 7am, 11am, 1pm, 5pm, 7pm, 11pm

 

crontab(minute=0, hour='*/5') Execute hour divisible by 5. This means that it is triggered at 3pm, not 5pm (since 3pm equals the 24-hour clock value of “15”, which is divisible by 5).

 

- crontab(minute=0, hour='*/3,8-17') Execute every hour divisible by 3, and every hour during office hours (8am-5pm).
- crontab(0, 0, day_of_month='2') Execute on the second day of every month.
- crontab(0, 0, day_of_month='2-30/2') Execute on every even numbered day.
- crontab(0, 0, day_of_month='1-7,15-21') Execute on the first and third weeks of the month.
- crontab(0, 0, day_of_month='11', month_of_year='5') Execute on the eleventh of May every year.
- crontab(0, 0, month_of_year='*/3') Execute every day on the first month of every quarter.

 

note

- celery-beat의 스케줄은 admin 페이지에서 설정 가능하고 settings.py에서도 설정 가능하다.

- 기본적인 방법은 settings.py에 스케줄을 미리 선언하는 것이다.

CELERY_BEAT_SCHEDULE = {
    "scheduled_task": {
        "task": "task1.tasks.add", #작업 이름
        "schedule": 5.0, #초단위
        "args": (10, 10), #아규먼트
    },
    "database": {
        "task": "task3.tasks.bkup",
        "schedule": 5.0,
    },
}

- celery-beat 실행

celery -A config beat -l INFO

- flower의 tasks 탭을 눌러 실시간 모니터링을 확인하면 5초마다 20이란 결과가 출력되는 것을 확인할 수 있다.

- admin 페이지를 통해 설정을 할 수도 있다. 하지만 충돌을 방지하기 위해 settings.py에 설언된 스케줄은 삭제해야한다.

- celery-beat 설치 후 migrate를 하게 되면 admin 페이지에 메뉴가 생성된다.

- periodic task에 작업을 새로 생성하여 아래 항목들에 대해서 입력을 해준다.

- 결과로 40이 출력되는 것을 확인할 수 있다.

 5. Learn Django Celery - Result Back-ends - Part 4

https://www.youtube.com/watch?v=jeE-k5dG8ug&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20&index=7 

 

content

- django-celery- results-backend package

- django cache framework

   - saving results to django cache database

 

note

- django-celery-results 

   - 설치 : pip install django-celery-results

   - 설정 

#settings.py
INSTALLED_APPS = [
.
.
"django_celery_results",
]

CELERY_RESULT_BACKEND = "django-db"

   - migrate 실행

   - admin page에서 확인

* CELERY_CACHE_BACKEND는 setting에서 cache를 직접 설정할 수 있는 방법들이 나온 이후 더 이상 사용되지 않는다고 메뉴얼에 나와있음

   - 선언해도 캐시 데이터가 누적되는게 확인 안됨

   - 예상으로 기존 CACHES를 사용하던 방법으로 따로 고려하지 않아도 될것 같음

 6. Docker - Django, Celery & Redis Docker Compose setup

https://www.youtube.com/watch?v=oBQxFn1CDno&list=PLOLrQ9Pn6caz-6WpcBYxV84g9gwptoN20&index=7 

github : https://github.com/veryacademy/YT-Django-Docker-Compose-Celery-Redis-PostgreSQL

GitHub - veryacademy/YT-Django-Docker-Compose-Celery-Redis-PostgreSQL

note

- docker 파일에 아래와 같은 선언이 있다. 이는 출력 버퍼를 사용 여부를 결정한다. 

  1은 True의 의미이다. 콘솔에서는 -u 옵션을 사용하면 바로 출력된다.

ENV PYTHONUNBUFFERED=1

- docker-compose에서 environment를 사용하여 외부 환경변수에 값을 선언하고 django에서 이를 참조하는 방식을 사용한다. 이를 위해 settings.py에서는 다음과 같은 코드가 요구된다.

# 왼쪽 변수명, 오른쪽 default
CELERY_BROKER_URL = os.environ.get("CELERY_BROKER", "redis://redis:6379/0")
CELERY_RESULT_BACKEND = os.environ.get("CELERY_BROKER", "redis://redis:6379/0")

 

reference : 

- official tutorial site 

https://www.rabbitmq.com/getstarted.html

RabbitMQ Tutorials — RabbitMQ

- [Django] Celery + RabbitMQ 세팅 [출처] [Django] Celery + RabbitMQ 세팅|작성자 레챠

https://blog.naver.com/PostView.naver?blogId=lastingchild&logNo=222348184974&parentCategoryNo=&categoryNo=&viewDate=&isShowPopularPosts=false&from=postView 

[Django] Celery + RabbitMQ 세팅

- [Django + Celery + RabbitMQ] 분산 비동기 작업 수행을 위한 Celery 로컬에서 테스트해보기

https://velog.io/@ssssujini99/Django-Celery-RabbitMQ-%EB%B6%84%EC%82%B0-%EB%B9%84%EB%8F%99%EA%B8%B0-%EC%9E%91%EC%97%85-%EC%88%98%ED%96%89%EC%9D%84-%EC%9C%84%ED%95%9C-Celery-%EB%A1%9C%EC%BB%AC%EC%97%90%EC%84%9C-%ED%85%8C%EC%8A%A4%ED%8A%B8%ED%95%B4%EB%B3%B4%EA%B8%B0

[Django + Celery + RabbitMQ] 분산 비동기 작업 수행을 위한 Celery 로컬에서 테스트해보기

- Python Celery & RabbitMQ Tutorial

https://kimdoky.github.io/tech/2019/01/23/celery-rabbitmq-tuto/

Python Celery & RabbitMQ Tutorial

- Django Celery - 단점, Task & subTask & Signature 비동기 작업 다루기 with network I/O

https://velog.io/@qlgks1/Django-Celery-%EB%8B%A8%EC%A0%90-Task-subTask-Signature-%EB%B9%84%EB%8F%99%EA%B8%B0-%EC%9E%91%EC%97%85-%EB%8B%A4%EB%A3%A8%EA%B8%B0-with-network-IO

Django Celery - 단점, Task & subTask & Signature 비동기 작업 다루기 with network I/O