Python 서버 프로그래밍: django-celery를 사용한 작업 대기열

from __future__ import absolute_import, unicode_literals

import os

from celery import Celery

# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'example.settings')

app = Celery('example')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

# Celery Configuration
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# app注册
INSTALLED_APPS = (
    ...
    'django_celery_results',
    'django_celery_beat',
    ...
)

from django.core.mail import send_mail
from celery import shared_task
from datetime import datetime


@shared_task
def send_email_task():
    subject = 'Celery Email Demo'
    message = 'This email is sent using celery!'
    from_email = 'demo@example.com'
    recipient_list = ['recipient@example.com']
    send_mail(subject, message, from_email, recipient_list)

    print('Email Sent Successfully')
    return None


@shared_task
def print_time():
    now = datetime.now()
    current_time = now.strftime("%H:%M:%S")
    print("Current Time =", current_time)

    return None

$ celery -A example worker --loglevel=info

from django.contrib import admin
from django_celery_beat.admin import PeriodicTaskAdmin, IntervalScheduleAdmin
from django_celery_results.models import TaskResult
from django_celery_results.admin import TaskResultAdmin
from core.tasks import send_email_task, print_time


class Tasks(admin.ModelAdmin):
    actions = ['send_email_task', 'print_time']

    def send_email_task(self, request, queryset):
        send_email_task.delay()

    send_email_task.short_description = "Send Email Task"

    def print_time(self, request, queryset):
        print_time.delay()

    print_time.short_description = "Print Current Time"


admin.site.unregister(TaskResult)
admin.site.register(TaskResult, TaskResultAdmin)
admin.site.register(IntervalSchedule, IntervalScheduleAdmin)
admin.site.register(PeriodicTask, PeriodicTaskAdmin)
admin.site.register(Tasks)