Django และ Celery: การประมวลผลงานแบบอะซิงโครนัสและคำถามสัมภาษณ์ 2026

Django และ Celery เป็นรากฐานของการประมวลผลงานแบบอะซิงโครนัสในแอปพลิเคชันเว็บ Python ด้วย Celery 5.6 และ Django 6.0 ที่ออกอัปเดตสำคัญในปี 2026 ความเข้าใจเกี่ยวกับคิวงานแบบกระจายยังคงเป็นทักษะสำคัญสำหรับการสัมภาษณ์ backend และระบบ production

วิธีที่ Celery ทำงานร่วมกับ Django

Celery ทำงานเป็น process อิสระที่ใช้ codebase ร่วมกับโปรเจกต์ Django การรวมระบบอาศัยสามองค์ประกอบ: แอปพลิเคชัน Django (producer), message broker (Redis หรือ RabbitMQ) และ Celery worker หนึ่งตัวขึ้นไป (consumer) เมื่อ Django view เรียก .delay() หรือ .apply_async() งานจะถูก serialize และส่งไปยังคิว broker Worker จะหยิบงานมาและดำเนินการฟังก์ชันใน process ของตัวเอง

การตั้งค่าเริ่มต้นด้วยโมดูล celery.py ที่ root ของโปรเจกต์ Django:

# myproject/celery.py
import os
from celery import Celery

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

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

การเรียก autodiscover_tasks() จะสแกนทุกแอป Django ที่ติดตั้งเพื่อค้นหาโมดูล tasks.py และลงทะเบียนฟังก์ชันที่ตกแต่งด้วย decorator ทั้งหมดเป็นงาน Celery โดยอัตโนมัติ

ใน settings.py broker และ result backend ถูกกำหนดค่า:

# myproject/settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'

Redis ทำหน้าที่สองอย่างทั้ง broker และ result backend ในการ deploy Django ส่วนใหญ่ RabbitMQ ยังคงเป็นตัวเลือกที่ดีกว่าสำหรับแอปพลิเคชันที่ต้องการ quorum queues หรือ routing ขั้นสูง ซึ่งเป็นฟีเจอร์ที่รองรับเต็มรูปแบบตั้งแต่ Celery 5.5

การเขียนและส่งงาน Celery

งาน Celery คือฟังก์ชัน Python ปกติที่ตกแต่งด้วย @shared_task Decorator shared_task หลีกเลี่ยงการ hardcode instance ของ Celery app ทำให้งานสามารถนำกลับมาใช้ใหม่ได้ในแอป Django ต่างๆ

# orders/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from orders.models import Order

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_order_confirmation(self, order_id: int) -> str:
    """Send confirmation email for a completed order."""
    try:
        order = Order.objects.select_related('user').get(id=order_id)
        send_mail(
            subject=f'Order #{order.id} Confirmed',
            message=f'Your order totaling {order.total} has been confirmed.',
            from_email='noreply@example.com',
            recipient_list=[order.user.email],
        )
        return f'Email sent for order {order_id}'
    except Order.DoesNotExist:
        return f'Order {order_id} not found'
    except Exception as exc:
        # Retry on transient failures (SMTP timeout, etc.)
        raise self.retry(exc=exc)

มีสามรูปแบบสำคัญที่ปรากฏที่นี่ ประการแรก bind=True ให้งานเข้าถึง self เพื่อเปิดใช้งาน retry ประการที่สอง ฟังก์ชันรับ integer order_id แทนที่จะเป็นออบเจกต์ ORM เนื่องจาก instance ของ model ไม่สามารถ serialize ได้อย่างน่าเชื่อถือข้ามขอบเขต process ประการที่สาม self.retry(exc=exc) จะส่งงานกลับเข้าคิวด้วย exponential backoff เมื่อเกิดความล้มเหลวชั่วคราว

การส่งจาก Django view:

# orders/views.py
from orders.tasks import send_order_confirmation

def complete_order(request, order_id):
    # ... process payment, update order status ...
    send_order_confirmation.delay(order_id)
    return redirect('order_success', order_id=order_id)

การเรียก .delay() จะ return ทันที ผู้ใช้เห็นหน้าสำเร็จในขณะที่งานอีเมลทำงานอยู่เบื้องหลัง

Task Routing และการจัดการคิว

ระบบ production ไม่ควรส่งงานทั้งหมดไปยังคิว default เดียว งานสร้างรายงานที่ช้าสามารถทำให้งานแจ้งเตือนที่เร็วขาดแคลนทรัพยากรหากใช้ worker pool เดียวกัน Celery แก้ปัญหานี้ด้วย task routing

# myproject/settings.py
CELERY_TASK_ROUTES = {
    'orders.tasks.send_order_confirmation': {'queue': 'notifications'},
    'reports.tasks.generate_monthly_report': {'queue': 'reports'},
    'images.tasks.resize_upload': {'queue': 'media'},
}

จากนั้น worker จะเริ่มต้นตามคิว:

# Start notification worker (fast tasks, high concurrency)
celery -A myproject worker -Q notifications -c 8 --loglevel=info

# Start report worker (slow tasks, limited concurrency)
celery -A myproject worker -Q reports -c 2 --loglevel=info

# Start media worker (CPU-bound, prefork pool)
celery -A myproject worker -Q media -c 4 -P prefork --loglevel=info

การแยกนี้ป้องกันการแย่งชิงทรัพยากร Worker แจ้งเตือนรองรับ throughput สูงด้วย 8 thread พร้อมกัน ในขณะที่ worker รายงานรันเพียง 2 งานพร้อมกันเพื่อหลีกเลี่ยงหน่วยความจำหมดจากการ query dataset ขนาดใหญ่

งานตามกำหนดเวลาด้วย Celery Beat

Celery Beat คือ scheduler ในตัวสำหรับงานที่ทำซ้ำ มันทำงานเป็น process แยกที่ส่งงานไปยัง broker ตามช่วงเวลาที่กำหนด

# myproject/settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'accounts.tasks.cleanup_expired_sessions',
        'schedule': crontab(hour=3, minute=0),  # Daily at 3 AM UTC
    },
    'sync-inventory': {
        'task': 'inventory.tasks.sync_external_inventory',
        'schedule': 300.0,  # Every 5 minutes
    },
    'generate-weekly-digest': {
        'task': 'notifications.tasks.send_weekly_digest',
        'schedule': crontab(hour=9, minute=0, day_of_week='monday'),
    },
}

Beat เริ่มต้นพร้อมกับ worker:

celery -A myproject beat --loglevel=info

ข้อผิดพลาดที่พบบ่อยใน production: การรัน Beat หลาย instance ทำให้เกิดการดำเนินงานซ้ำ ควรมีเพียง process Beat เดียวต่อ deployment เครื่องมืออย่าง django-celery-beat จัดเก็บตารางเวลาในฐานข้อมูล ทำให้สามารถแก้ไขได้ขณะรันผ่าน Django admin

การตรวจสอบและ Observability ด้วย Flower

Flower ให้ dashboard เว็บแบบ real-time สำหรับตรวจสอบ worker, งาน และคิวของ Celery Flower แสดงอัตราความสำเร็จของงาน เวลาดำเนินการ ความลึกของคิว และสถานะ worker

# Install and run Flower
pip install flower
celery -A myproject flower --port=5555

ใน production ตัวชี้วัดจาก Flower ควรถูกส่งไปยังระบบแจ้งเตือน ความลึกของคิวที่เพิ่มขึ้นพร้อมงานที่ค้างบ่งชี้ปัญหาด้านความจุ worker อัตรา retry ที่สูงในงานเฉพาะชี้ไปที่ความไม่เสถียรของบริการภายนอก

Celery 5.6 ยังแนะนำการปรับปรุง structured logging ที่ทำงานร่วมกับตัวรวบรวม JSON log เช่น Datadog หรือ ELK stack ให้ประสิทธิภาพการ debug ดีขึ้น 30% ตามบันทึกการเผยแพร่

Framework งานในตัวของ Django 6.0 เทียบกับ Celery

Django 6.0 เปิดตัว framework งานเบื้องหลังแบบ native ทำให้เกิดคำถามว่า Celery ยังจำเป็นหรือไม่ คำตอบขึ้นอยู่กับกรณีการใช้งาน

Framework งานในตัวของ Django รองรับการดำเนินการเบื้องหลังแบบง่าย ได้แก่ การส่งอีเมล การล้าง cache การประมวลผลข้อมูลเบา โดยไม่ต้องใช้โครงสร้างพื้นฐาน broker แยกต่างหาก Task runner ถูกสร้างไว้ภายใน Django เอง

Celery ยังคงจำเป็นสำหรับ:

• การประมวลผลแบบกระจาย บนหลายเครื่อง
• คิวลำดับความสำคัญ และ task routing
• Rate limiting และนโยบาย retry ขั้นสูง
• การตั้งเวลาตามกำหนด (Celery Beat)
• การติดตามผลลัพธ์ ด้วย backend ที่กำหนดค่าได้
• Canvas workflows (chains, groups, chords) สำหรับการประสานงานที่ซับซ้อน

สำหรับแอปพลิเคชันที่ต้องการเพียงงานเบื้องหลังแบบ fire-and-forget โซลูชันในตัวของ Django 6.0 ลดความซับซ้อนในการดำเนินงาน สำหรับสิ่งใดก็ตามที่เกี่ยวข้องกับ worker แบบกระจาย การตั้งเวลา หรือการประกอบงาน Celery 5.6 คือตัวเลือกที่พิสูจน์แล้ว

คำถามสัมภาษณ์: Django และ Celery

การสัมภาษณ์ทางเทคนิคสำหรับตำแหน่ง backend มักทดสอบความรู้เกี่ยวกับ Celery ด้านล่างนี้คือคำถามที่นำมาจากสถานการณ์สัมภาษณ์ Django จริงที่บริษัทตั้งแต่ startup จนถึง FAANG

ถ: ทำไมงานควรรับ primary key แทนที่จะเป็น instance ของ model?

Instance ของ model Django มีการเชื่อมต่อฐานข้อมูล queryset และ relation ที่โหลดแบบ lazy ที่ไม่สามารถอยู่รอดจากการ serialization การส่ง order_id แทนที่จะเป็นออบเจกต์ Order ทำให้มั่นใจว่างานดึงข้อมูลใหม่จากฐานข้อมูล หลีกเลี่ยง state ที่ล้าสมัยและข้อผิดพลาดในการ serialization

ถ: self.retry() แตกต่างจากการส่งงานใหม่ด้วยตนเองอย่างไร?

self.retry() รักษาจำนวน retry ใช้ขีดจำกัด max_retries ที่กำหนด และใช้ exponential backoff โดยค่าเริ่มต้น การเรียก .delay() ด้วยตนเองอีกครั้งสร้างงานใหม่ทั้งหมดพร้อมตัวนับ retry ที่ถูกรีเซ็ต ซึ่งอาจนำไปสู่ลูป retry ไม่สิ้นสุดเมื่อเกิดความล้มเหลวแบบถาวร

ถ: เกิดอะไรขึ้นเมื่อ Celery worker ล่มกลางงาน?

พฤติกรรมขึ้นอยู่กับการตั้งค่า acks_late หาก acks_late=True broker จะส่งข้อความอีกครั้งไปยัง worker อื่นเนื่องจาก acknowledgment ไม่เคยถูกส่ง หากใช้ค่าเริ่มต้น acks_late=False ข้อความจะถูก acknowledge ก่อนดำเนินการ ดังนั้น crash หมายความว่างานหายไป ระบบ production ที่จัดการภาระงานสำคัญควรใช้ acks_late=True ร่วมกับการออกแบบงานแบบ idempotent

ถ: อธิบายความแตกต่างระหว่าง .delay(), .apply_async() และการเรียกงานโดยตรง

.delay(*args) คือ syntactic sugar สำหรับ .apply_async(args=args) .apply_async() รับตัวเลือกเพิ่มเติมเช่น countdown, eta, queue, priority และ expires การเรียกฟังก์ชันงานโดยตรง (โดยไม่ใช้ .delay()) จะดำเนินการแบบ synchronous ใน process ปัจจุบัน ซึ่งมีประโยชน์สำหรับการทดสอบแต่เสียจุดประสงค์ของการประมวลผล async

ถ: caching layer มีปฏิสัมพันธ์กับผลลัพธ์ของงาน Celery อย่างไร?

ผลลัพธ์ของงานที่เก็บใน Celery result backend สามารถ cache ได้โดยใช้ framework cache ของ Django View ตรวจสอบ cache ก่อน หากไม่มี จะส่งงานและเก็บ ID ของ AsyncResult request ถัดไปจะ poll result backend ผ่าน ID ที่ cache ไว้จนกว่างานจะเสร็จ แล้วจึง cache ผลลัพธ์สุดท้าย

รายการตรวจสอบการ Deploy สำหรับ Production

การ deploy Celery ร่วมกับ Django ต้องให้ความสำคัญกับข้อกังวลด้านการปฏิบัติงานหลายประการที่การสัมภาษณ์ก็มักถามเช่นกัน

# myproject/settings.py — Production configuration
CELERY_TASK_ALWAYS_EAGER = False  # Never True in production
CELERY_TASK_ACKS_LATE = True  # Redelivery on worker crash
CELERY_WORKER_PREFETCH_MULTIPLIER = 1  # Fair scheduling
CELERY_TASK_REJECT_ON_WORKER_LOST = True  # Reject on unexpected exit
CELERY_TASK_TIME_LIMIT = 300  # Hard kill after 5 minutes
CELERY_TASK_SOFT_TIME_LIMIT = 240  # SoftTimeLimitExceeded after 4 min
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000  # Prevent memory leaks
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True

การตั้งค่า WORKER_MAX_TASKS_PER_CHILD จะรีสตาร์ท worker process หลังจาก 1000 งาน ป้องกัน memory leak ซึ่งเป็นปัญหาที่ Celery 5.6 แก้ไขอย่างกว้างขวางด้วย patch memory leak

การจัดการ process ด้วย systemd หรือ supervisor ทำให้มั่นใจว่า worker จะรีสตาร์ทเมื่อเกิดความล้มเหลว unit systemd แบบง่าย:

# /etc/systemd/system/celery-worker.service
[Unit]
Description=Celery Worker
After=network.target redis.service

[Service]
Type=forking
User=django
Group=django
WorkingDirectory=/opt/myproject
ExecStart=/opt/myproject/venv/bin/celery -A myproject worker \
    --loglevel=info --concurrency=4 --pidfile=/var/run/celery/worker.pid
ExecStop=/bin/kill -s TERM $MAINPID
Restart=always

[Install]
WantedBy=multi-user.target

สรุป

• Celery 5.6.3 (มีนาคม 2026) นำ memory leak fix, structured logging, การรองรับ quorum queue และความเข้ากันได้กับ psycopg3 มาให้ ควรอัปเกรดจากเวอร์ชันเก่าเพื่อรับประโยชน์ทันที
• ส่ง primary key ให้งานเสมอ ไม่ใช่ออบเจกต์ ORM สิ่งนี้ป้องกันข้อผิดพลาดในการ serialization และข้อมูลที่ล้าสมัย
• กำหนดเส้นทางงานไปยังคิวเฉพาะตามลักษณะของภาระงาน การแจ้งเตือนที่เร็วและรายงานที่ช้าไม่ควรแข่งขันกันเพื่อ worker เดียวกัน
• ใช้ acks_late=True และการออกแบบงานแบบ idempotent สำหรับภาระงานสำคัญที่ต้องอยู่รอดจากการล่มของ worker
• ตั้งค่าทั้ง time_limit และ soft_time_limit ในทุกงานเพื่อป้องกัน worker ที่ค้างจากการใช้ทรัพยากรอย่างไม่จำกัด
• Framework งานในตัวของ Django 6.0 รองรับงานเบื้องหลังแบบง่าย Celery ยังคงจำเป็นสำหรับการประมวลผลแบบกระจาย การตั้งเวลา และ canvas workflows
• ตรวจสอบความลึกของคิวและอัตรา retry ด้วย Flower หรือการรวบรวม log แบบมีโครงสร้างเพื่อตรวจจับปัญหาด้านความจุก่อนที่จะส่งผลกระทบต่อผู้ใช้