Chạy tác vụ nền trong Django không cần Celery

Vì sao cần chạy tác vụ nền?

Ở phần 6, chúng ta đã xây dựng pipeline biến CV thành bộ câu hỏi phỏng vấn: phân tích, xác thực, cấu trúc hóa, tạo nội dung rồi lưu lại. Nhưng có một vấn đề: quy trình này khá chậm. Việc phân tích tệp cùng nhiều lần gọi AI tuần tự có thể mất vài giây, và bạn không thể bắt trình duyệt của người dùng chờ mãi trên một HTTP request duy nhất. Vì vậy, chúng ta cần chạy tác vụ nền trong Django, tách khỏi vòng đời của request.

Thông thường, lựa chọn được nhắc đến là Celery. Tuy nhiên, Celery đồng nghĩa với việc phải chạy một broker như Redis, thêm các process và triển khai hạ tầng thực tế—quá nhiều thứ đối với một MVP. Ứng dụng này chọn cách nhẹ hơn, không cần dịch vụ bổ sung: khởi chạy một worker process riêng cho mỗi tác vụ, rồi để frontend polling để kiểm tra tiến độ. Cơ chế này hoạt động như sau.

Một registry tác vụ nhỏ gọn

Trước tiên, chúng ta cần một cách đăng ký các job nền theo tên. Đây là một decorator cực kỳ đơn giản, lưu các function vào một dictionary để có thể tìm lại tác vụ bằng một chuỗi:

TASK_REGISTRY = {}


def register_task(name):
    def decorator(func):
        TASK_REGISTRY[name] = func
        return func
    return decorator


def get_task(name):
    return TASK_REGISTRY.get(name)

Giờ đây, function process_interview từ phần 6 chỉ cần thêm một decorator là có thể được tìm thấy bằng tên:

@register_task("process_interview")
def process_interview(task, payload):
    interview = Interview.objects.get(id=payload["interview_id"])
    # ... parse, validate, generate questions (from part 6)

Tại sao không gọi trực tiếp function mà cần registry? Vì worker chạy job nằm trong một process riêng và chỉ nhận được một chuỗi ("process_interview")—nó cần cách chuyển tên đó trở lại thành function. Registry đảm nhiệm việc tra cứu này. Nó cũng giúp tất cả tác vụ nền có thể được tìm thấy ở một nơi duy nhất.

Khởi chạy tác vụ

Khi người dùng tải lên CV, view chỉ thực hiện những việc tối thiểu một cách đồng bộ—tạo interview, tạo bản ghi tác vụ, khởi chạy worker—sau đó trả về ngay với mã 202 Accepted, báo cho client rằng “đã tiếp nhận, hãy tiếp tục kiểm tra”:

interview = Interview.objects.create(
    candidate_name=full_name,
    candidate_email=email,
    cv_file=resume_file,
    device=request.device,
    client_ip=client_ip,
)

task = BackgroundTask.objects.create(
    type="process_interview",
    payload={"interview_id": interview.id},
    name=f"Process Interview {interview.interview_uuid}",
)
start_background_task(task.id)

return Response({"data": {"id": str(interview.interview_uuid)}}, status=202)

Điểm quan trọng là 202, không phải 200. Đây là mã HTTP trung thực cho tình huống “đã tiếp nhận để xử lý nhưng chưa hoàn tất”. Phần việc nặng chưa chạy tại thời điểm đó—response được trả về chỉ sau vài mili giây, còn pipeline thực tế chạy ở nơi khác.

Khởi chạy worker process

Đây là phần giúp cơ chế hoạt động mà không cần Celery. start_background_task khởi chạy một process hoàn toàn riêng biệt bằng module subprocess của Python, rồi process này chạy một Django management command:

import subprocess
import sys

from core.models import BackgroundTask


def start_background_task(task_id):
    task = BackgroundTask.objects.get(id=task_id)
    if task.status in [2, 3]:  # already running or done
        task.append_log("Task already running or completed", True)
        return task

    subprocess.Popen(
        [sys.executable, "manage.py", "background_worker", f"--task_id={task.id}"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    return task

Hai lựa chọn ở đây đều có chủ đích. start_new_session=True tách worker khỏi web process, nhờ đó worker vẫn tiếp tục chạy độc lập ngay cả khi HTTP response đã được trả về từ lâu. Output được chuyển tới DEVNULL vì đây là process chạy kiểu “fire-and-forget”—cơ chế “logging” thực sự nằm ở các trường trạng thái và log được ghi vào database, là những dữ liệu chúng ta có thể truy vấn sau này.

Tại sao dùng process riêng thay vì thread? Tính cô lập. Thread dùng chung bộ nhớ và process của web server—một tác vụ bị crash nghiêm trọng hoặc ngốn quá nhiều bộ nhớ có thể kéo sập toàn bộ server. Subprocess được cách ly: nếu nó dừng, web app của bạn thậm chí không nhận ra. Sự cô lập này đặc biệt giá trị khi công việc liên quan đến tệp lớn và các lần gọi AI khó dự đoán.

Worker command

Process được khởi chạy sẽ chạy một management command chỉ có nhiệm vụ tìm tác vụ, chạy tác vụ và ghi lại kết quả:

import os

from django.core.management.base import BaseCommand

from core.models import BackgroundTask
from core.registry import get_task
from tasks import *  # noqa — importing runs the @register_task decorators


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument("--task_id", type=str)

    def handle(self, *args, **options):
        task = BackgroundTask.objects.get(id=options["task_id"])
        task.status = 2  # running
        task.pid = os.getpid()
        task.save(update_fields=["status", "pid", "updated_at"])

        func = get_task(task.type)
        if not func:
            task.append_log("Task type not registered", commit=True)
            return

        try:
            func(task, task.payload)
            task.status = 3  # done
        except Exception as e:
            task.status = 4  # error
            task.append_log(str(e))
        task.save(update_fields=["status", "log", "updated_at"])

Hãy chú ý dòng from tasks import *—lệnh import này thực sự chạy các decorator @register_task và nạp dữ liệu vào registry, nhờ đó get_task có thể tìm thấy function. Từ đây, quy trình khá thẳng: đánh dấu tác vụ đang chạy, thực thi, rồi đánh dấu hoàn tất hoặc lỗi. Mọi thay đổi trạng thái đều được ghi vào database, và chính điều đó cho phép frontend theo dõi tiến độ từ bên ngoài.

Polling để kiểm tra trạng thái

Vì công việc diễn ra ngoài request, frontend cần một cách để hỏi “đã xong chưa?”. Đó chỉ là một endpoint đọc trạng thái hiện tại của interview:

class InterviewStatusAPIView(APIView):
    INTERVIEW_STATUS = {
        1: "Pending", 2: "Processing Resume", 3: "Preparing Interview",
        4: "In progress", 7: "Completed", 8: "Failed",
    }

    def get(self, request, *args, **kwargs):
        interview = Interview.objects.get(
            interview_uuid=kwargs["interview_id"],
            device=request.device,
        )
        return Response({
            "status": interview.status,
            "status_text": self.INTERVIEW_STATUS.get(interview.status),
        })

Frontend gọi endpoint này sau mỗi vài giây và cập nhật UI khi trạng thái chuyển từ “Processing Resume” sang “Preparing Interview”, rồi sẵn sàng. Đây là short polling—xét trên lý thuyết không phải lựa chọn thanh lịch nhất, nhưng với một job thỉnh thoảng mới chạy và chỉ chạy một lần như thế này, nó đơn giản, đáng tin cậy, không cần websocket hay hạ tầng bổ sung. Với MVP, cố ý chọn phương án nhàm chán nhưng không phụ thuộc là quyết định đúng trong nhiều trường hợp.

Phiên bản production sẽ cần tiến xa hơn

Đến đây, bạn đã có đầy đủ cơ chế: một registry, worker subprocess chạy kiểu fire-and-forget và polling trạng thái—đủ để chạy toàn bộ pipeline phỏng vấn ngoài vòng đời request, đồng thời đủ để hoàn thiện một ứng dụng thực sự hoạt động. Đây là cột mốc quan trọng: từ thời điểm này, quy trình phỏng vấn dạng văn bản đã chạy end-to-end.

Cách tiếp cận nhẹ này chưa cung cấp lớp vận hành nâng cao mà một hệ thống triển khai thực tế sau cùng sẽ cần: giới hạn lượng bộ nhớ worker có thể sử dụng, dừng hoặc kill một tác vụ bị treo, retry khi thất bại và tự động hoàn tất những interview bị đình trệ. Đây chính là các tính năng Celery cung cấp sẵn, và đánh đổi khi bỏ qua Celery là bạn phải tự xây dựng những phần thực sự cần dùng. Phiên bản task system đã được harden cho production—với giới hạn bộ nhớ, dừng tác vụ an toàn và khả năng phục hồi—có trong source hoàn chỉnh của bộ starter kit. Nếu không, hãy chuyển sang phần 8, nơi ứng dụng cuối cùng cũng “lên tiếng”: thêm tính năng nhập liệu bằng giọng nói để ứng viên trả lời thành tiếng, giống như một cuộc phỏng vấn thực tế.