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

이수진·2022년 8월 6일
celerydjango비동기
0
post-custom-banner

이번에는 Celery를 직접 설치하고 이용해보겠습니다!

로컬에서 바로 테스트 할 수 있는 방법을 소개해드리겠습니다.

1. 로컬 테스트용

저는 브로커로 RabbitMQ를 이용했는데요,
로컬에서 테스트하려면 RabbitMQ를 추가로 설치해주어야합니다.

맥에서 Rabbitmq를 미리 설치해주시고,
추가로 PATH 환경변수도 설정해주어야합니다.

.zshrc 에서

# rabbitmq
export PATH="$PATH:/usr/local/sbin"

✔️설치

$ pip install celery
$ pip install django-celery-results
$ pip install django-celery-beat

✔️파일 설정


# settings.py


# Application definition
INSTALLED_APPS = [
    ...
    
    # celery
    'django_celery_beat',
    'django_celery_results',
]


# Celery
CELERY_BROKER_URL = 'amqp://localhost'  # 로컬 테스트용
CELERY_RESULT_BACKEND = 'django-db'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

...

여기서 CELERY_BROKER_URL은 로컬 테스트를 위한 URL입니다. (배포용과는 다름에 주의하세요)

✔️celery.py 와 tasks.py 파일 정의하기

celery.py 는 프로젝트 디렉토리에,
task.py 는 앱 디렉토리 안에 위치해 있어야합니다.


# celery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from celery.schedules import crontab

# 기본 장고파일 설정
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '프로젝트명.settings')
app = Celery('프로젝트명')
app.config_from_object('django.conf:settings', namespace='CELERY')

#등록된 장고 앱 설정에서 task 불러오기
app.autodiscover_tasks()

# tasks.py

from __future__ import absolute_import, unicode_literals
from celery import shared_task

from dnd_7th_4_backend.celery import app


# test 용 함수
@shared_task
def printTime():
    print("Testtime: ", datetime.now())

✔️로컬에서 테스트해보기

먼저 터미널에서 rabbitmq-server를 실행해주어야합니다.

$ rabbitmq-server

다음 커맨드를 입력하여 rabbitmq-server를 실행시켜줍니다.

(base)  sujin@isujin-ui-MacBookPro  ~  rabbitmq-server
2022-08-13 11:01:17.081600+09:00 [info] <0.221.0> Feature flags: list of feature flags found:
2022-08-13 11:01:17.085556+09:00 [info] <0.221.0> Feature flags:   [x] classic_mirrored_queue_version
2022-08-13 11:01:17.085570+09:00 [info] <0.221.0> Feature flags:   [x] implicit_default_bindings
2022-08-13 11:01:17.085582+09:00 [info] <0.221.0> Feature flags:   [x] maintenance_mode_status
2022-08-13 11:01:17.085607+09:00 [info] <0.221.0> Feature flags:   [x] quorum_queue
2022-08-13 11:01:17.085614+09:00 [info] <0.221.0> Feature flags:   [x] stream_queue
2022-08-13 11:01:17.085622+09:00 [info] <0.221.0> Feature flags:   [x] user_limits
2022-08-13 11:01:17.085628+09:00 [info] <0.221.0> Feature flags:   [x] virtual_host_metadata
2022-08-13 11:01:17.085649+09:00 [info] <0.221.0> Feature flags: feature flag states written to disk: yes
2022-08-13 11:01:17.199884+09:00 [notice] <0.44.0> Application syslog exited with reason: stopped
2022-08-13 11:01:17.199928+09:00 [notice] <0.221.0> Logging: switching to configured handler(s); following messages may not be visible in this log output

  ##  ##      RabbitMQ 3.10.7
  ##  ##
  ##########  Copyright (c) 2007-2022 VMware, Inc. or its affiliates.
  ######  ##
  ##########  Licensed under the MPL 2.0. Website: https://rabbitmq.com

  Erlang:      25.0.3 [jit]
  TLS Library: OpenSSL - OpenSSL 1.1.1q  5 Jul 2022

  Doc guides:  https://rabbitmq.com/documentation.html
  Support:     https://rabbitmq.com/contact.html
  Tutorials:   https://rabbitmq.com/getstarted.html
  Monitoring:  https://rabbitmq.com/monitoring.html

  Logs: /opt/homebrew/var/log/rabbitmq/rabbit@localhost.log
        /opt/homebrew/var/log/rabbitmq/rabbit@localhost_upgrade.log
        <stdout>

  Config file(s): (none)

  Starting broker... completed with 7 plugins.

실행하면 다음과 같습니다.

그 다음에는 python 로컬 터미널 두 개를 열어줍니다.

$ celery -A 프로젝트명 worker -l info

를 입력합니다.

(venv) (base)  sujin@isujin-ui-MacBookPro  ~/Django/dnd_7th_4_backend   main/ssssujini99 ± >R>  celery -A dnd_7th_4_backend worker -l info

-------------- celery@isujin-ui-MacBookPro.local v5.2.7 (dawn-chorus)
--- ***** ----- 
-- ******* ---- macOS-12.1-arm64-arm-64bit 2022-08-13 11:10:42
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         dnd_7th_4_backend:0x1068d3c40
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     
- *** --- * --- .> concurrency: 10 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
-------------- [queues]
               .> celery           exchange=celery(direct) key=celery
               

[tasks]
 . main.tasks.printTime

[2022-08-13 11:10:42,478: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2022-08-13 11:10:42,482: INFO/MainProcess] mingle: searching for neighbors
[2022-08-13 11:10:43,506: INFO/MainProcess] mingle: all alone
[2022-08-13 11:10:43,522: WARNING/MainProcess] /Users/sujin/Django/dnd_7th_4_backend/venv/lib/python3.8/site-packages/celery/fixups/django.py:203: UserWarning: Using settings.DEBUG leads to a memory
           leak, never use this setting in production environments!
 warnings.warn('''Using settings.DEBUG leads to a memory

[2022-08-13 11:10:43,522: INFO/MainProcess] celery@isujin-ui-MacBookPro.local ready.

celery 세팅이 잘 되었다면, 이렇게 마지막에 준비가 되었다고 뜹니다.

그러면 바로 다음 로컬 창에서는 이 tasks.py에 있는 함수를 실행시켜줍니다.

(venv) (base)  sujin@isujin-ui-MacBookPro  ~/Django/dnd_7th_4_backend   master ±  python manage.py shell
Python 3.8.9 (default, Oct 26 2021, 07:25:53) 
[Clang 13.0.0 (clang-1300.0.29.30)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> from main.tasks import printName
>>> printName.delay("sujin")
<AsyncResult: 48c5d9d3-c0a6-459d-929c-ba8f8ba4c014>
>>> printName.delay("celery")
<AsyncResult: e237b492-83e5-4038-bab5-b7a6e4bdd9d4>
>>>

위와 같이 파이썬 쉘을 실행하여 task를 import 하여 실행하면 worker로 job을 보내게 됩니다.

바로 worker server를 확인하면 다음과 같습니다.

[2022-08-07 05:13:27,146: INFO/MainProcess] Task main.tasks.printName[48c5d9d3-c0a6-459d-929c-ba8f8ba4c014] received
[2022-08-07 05:13:27,147: WARNING/ForkPoolWorker-7] my name is sujin
[2022-08-07 05:13:27,161: INFO/ForkPoolWorker-7] Task main.tasks.printName[48c5d9d3-c0a6-459d-929c-ba8f8ba4c014] succeeded in 0.014397875000000226s: None
[2022-08-07 05:13:42,329: INFO/MainProcess] Task main.tasks.printName[e237b492-83e5-4038-bab5-b7a6e4bdd9d4] received
[2022-08-07 05:13:42,331: WARNING/ForkPoolWorker-7] my name is celery
[2022-08-07 05:13:42,340: INFO/ForkPoolWorker-7] Task main.tasks.printName[e237b492-83e5-4038-bab5-b7a6e4bdd9d4] succeeded in 0.009288417000000493s: None

✔️ celery.py에 task로 등록하기

이번에는, 직접 celery.py에 등록한 후에 실행시킨 결과를 확인해보겠습니다.
celery.py 파일 가장 밑에 수행시킬 task를 등록합니다.

# celery.py
...

# task 함수 주기 설정
app.conf.beat_schedule = {
    'printTime': {  # 스케쥴링 이름
        'task' : 'main.tasks.printTime', # 수행할 task 설정
        'schedule': crontab(),  # 인자 없으면 매 분마다 실행
    }
}

이렇게 직접 등록시켜 printTime 함수를 매 분마다 실행시킨 결과를 확인해보겠습니다.

✔️ 결과 확인하기

아까와 같이 rabbitmq-server와 celery-worker를 키고,
celery-beat까지 함께 켜주어야합니다.

다음 명령어를 입력해줍니다.

$ celery -A 프로젝트명 beat --loglevel=info

입력하면 다음과 같습니다.

(venv) (base)  sujin@isujin-ui-MacBookPro  ~/Django/dnd_7th_4_backend   main/ssssujini99 ± >R>  celery -A dnd_7th_4_backend  beat --loglevel=info
celery beat v5.2.7 (dawn-chorus) is starting.
__    -    ... __   -        _
LocalTime -> 2022-08-13 11:09:09
Configuration ->
   . broker -> amqp://guest:**@localhost:5672//
   . loader -> celery.loaders.app.AppLoader
   . scheduler -> celery.beat.PersistentScheduler
   . db -> celerybeat-schedule
   . logfile -> [stderr]@%INFO
   . maxinterval -> 5.00 minutes (300s)
[2022-08-13 11:09:09,185: INFO/MainProcess] beat: Starting...
[2022-08-13 11:09:09,201: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:10:00,006: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:11:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:12:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:13:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:14:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:15:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:16:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:17:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:18:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)
[2022-08-13 11:19:00,000: INFO/MainProcess] Scheduler: Sending due task printTime (main.tasks.printTime)

다음과 같이 매 분마다 스케쥴러가 task(printTime)를 보냄을 알 수 있고

celery-worker을 켠 터미널을 확인해보면

[2022-08-13 11:10:43,522: INFO/MainProcess] celery@isujin-ui-MacBookPro.local ready.
[2022-08-13 11:11:00,010: INFO/MainProcess] Task main.tasks.printTime[db00b343-62aa-4050-a4e8-900fc936fdf8] received
[2022-08-13 11:11:00,016: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:11:00,017: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:11:00,017: WARNING/ForkPoolWorker-7] 2022-08-13 11:11:00.014020
[2022-08-13 11:11:00,035: INFO/ForkPoolWorker-7] Task main.tasks.printTime[db00b343-62aa-4050-a4e8-900fc936fdf8] succeeded in 0.021716749999999507s: None
[2022-08-13 11:12:00,001: INFO/MainProcess] Task main.tasks.printTime[f7b38630-3c38-4f48-9986-90a288d8f613] received
[2022-08-13 11:12:00,002: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:12:00,002: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:12:00,003: WARNING/ForkPoolWorker-7] 2022-08-13 11:12:00.002442
[2022-08-13 11:12:00,007: INFO/ForkPoolWorker-7] Task main.tasks.printTime[f7b38630-3c38-4f48-9986-90a288d8f613] succeeded in 0.005130540999999766s: None
[2022-08-13 11:13:00,002: INFO/MainProcess] Task main.tasks.printTime[454b514e-ad10-4522-9008-296611a85faf] received
[2022-08-13 11:13:00,004: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:13:00,005: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:13:00,005: WARNING/ForkPoolWorker-7] 2022-08-13 11:13:00.003522
[2022-08-13 11:13:00,010: INFO/ForkPoolWorker-7] Task main.tasks.printTime[454b514e-ad10-4522-9008-296611a85faf] succeeded in 0.007554249999998319s: None
[2022-08-13 11:14:00,005: INFO/MainProcess] Task main.tasks.printTime[d0eedb6d-84e5-4e90-a09e-c6b6f0319f18] received
[2022-08-13 11:14:00,007: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:14:00,008: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:14:00,009: WARNING/ForkPoolWorker-7] 2022-08-13 11:14:00.006894
[2022-08-13 11:14:00,016: INFO/ForkPoolWorker-7] Task main.tasks.printTime[d0eedb6d-84e5-4e90-a09e-c6b6f0319f18] succeeded in 0.01001204199999961s: None
[2022-08-13 11:15:00,002: INFO/MainProcess] Task main.tasks.printTime[c4b228b1-f983-450f-aa66-21cf5efeea12] received
[2022-08-13 11:15:00,004: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:15:00,004: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:15:00,004: WARNING/ForkPoolWorker-7] 2022-08-13 11:15:00.003708
[2022-08-13 11:15:00,011: INFO/ForkPoolWorker-7] Task main.tasks.printTime[c4b228b1-f983-450f-aa66-21cf5efeea12] succeeded in 0.008307541999954537s: None
[2022-08-13 11:16:00,010: INFO/MainProcess] Task main.tasks.printTime[742604f8-9ec0-49c4-95c0-75ee406daa53] received
[2022-08-13 11:16:00,016: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:16:00,017: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:16:00,017: WARNING/ForkPoolWorker-7] 2022-08-13 11:16:00.013669
[2022-08-13 11:16:00,031: INFO/ForkPoolWorker-7] Task main.tasks.printTime[742604f8-9ec0-49c4-95c0-75ee406daa53] succeeded in 0.018162916999983736s: None
[2022-08-13 11:17:00,010: INFO/MainProcess] Task main.tasks.printTime[1b65e82a-b7b6-49df-9d47-7e7433caf0d7] received
[2022-08-13 11:17:00,016: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:17:00,017: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:17:00,018: WARNING/ForkPoolWorker-7] 2022-08-13 11:17:00.013599
[2022-08-13 11:17:00,025: INFO/ForkPoolWorker-7] Task main.tasks.printTime[1b65e82a-b7b6-49df-9d47-7e7433caf0d7] succeeded in 0.012287082999989707s: None
[2022-08-13 11:18:00,005: INFO/MainProcess] Task main.tasks.printTime[5c7c9e0a-aa51-4f1b-ac98-b1cc9bf14b56] received
[2022-08-13 11:18:00,006: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:18:00,007: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:18:00,007: WARNING/ForkPoolWorker-7] 2022-08-13 11:18:00.006561
[2022-08-13 11:18:00,014: INFO/ForkPoolWorker-7] Task main.tasks.printTime[5c7c9e0a-aa51-4f1b-ac98-b1cc9bf14b56] succeeded in 0.00807187499998463s: None
[2022-08-13 11:19:00,003: INFO/MainProcess] Task main.tasks.printTime[a2a91596-9264-4341-9e47-6036569aab73] received
[2022-08-13 11:19:00,003: WARNING/ForkPoolWorker-7] Testtime: 
[2022-08-13 11:19:00,003: WARNING/ForkPoolWorker-7]  
[2022-08-13 11:19:00,003: WARNING/ForkPoolWorker-7] 2022-08-13 11:19:00.003563
[2022-08-13 11:19:00,007: INFO/ForkPoolWorker-7] Task main.tasks.printTime[a2a91596-9264-4341-9e47-6036569aab73] succeeded in 0.004472082999996019s: None

이렇게 작업이 수행됐음을 확인할 수 있습니다!

다음번에는 직접 docker로 배포한 환경에서 테스트를 해보겠습니다

profile
이수진
꾸준히, 열심히, 그리고 잘하자
이전 포스트

[Django + Celery + RabbitMQ] 분산 비동기 작업 수행을 위한 Celery

다음 포스트

[EC2 + Docker + Django + Celery + RabbitMQ] EC2 + Docker 환경에서 Celery 세팅 및 테스트

post-custom-banner

0개의 댓글