Django 비동기 처리 및 웹 소켓 적용

문서 분석, 임베딩, VLM 캡션 생성과 같은 작업은 백엔드에서 상당한 시간이 걸린다. 이런 작업을 Django 뷰에서 동기적으로 처리하면 HTTP 요청이 끝날 때까지 브라우저가 응답을 기다리다가 타임아웃이 나거나, 사용자는 작업이 진행 중인지조차 알 수 없는 상태가 된다.

그래서 무거운 작업은 별도 워커에서 비동기로 돌리고, 처리가 끝나면 클라이언트에 완료 알림을 보내는 구조가 필요했다. 이를 위해 Celery + Redis 조합으로 비동기 처리를 구성하고, Django Channels 기반 웹 소켓으로 완료 알림 인터페이스를 구현하였다.

이 포스트에서 다루는 것

비동기 작업 큐(Celery + Redis), 웹 소켓 완료 알림(Django Channels), 그리고 관리자 화면의 처리 상태 UI까지를 한 흐름으로 정리한다.

Celery 및 Redis를 이용한 비동기 처리

비동기 처리의 핵심은 웹 요청을 받는 프로세스와 실제 작업을 수행하는 프로세스를 분리하는 것이다.

구성 요소	역할
Celery	Django 앱에 정의한 태스크를 워커가 받아 실행하는 분산 작업 큐. 웹에서 `task.delay(...)` 로 등록하면 워커가 큐에서 꺼내 실행한다.
Redis	메모리 기반 데이터 스토어. Celery 메시지 브로커로 태스크를 중계하고, 웹 소켓용 Channel Layer 백엔드로도 같은 인스턴스를 공유한다.

비동기 처리 핵심 구성 요소

전체 흐름은 다음과 같다.

[브라우저]
    |  (1) 파일 업로드 / 소스 수정 요청
    v
[Django Web]
    |  (2) Source 생성 또는 status=PENDING
    |  (3) document_parse_task.delay(...) 등 태스크 등록
    v
[Redis Broker] ----(4) 태스크 전달----> [Celery Worker]
                                              |
                              (5) Docling 분석 / 크롤링 / VLM / 임베딩
                                              |
                                              v
                                         [FAISS / MySQL]
                                              |
                              (6) status = COMPLETED | FAILED

파일 업로드 뷰에서는 파일을 저장하고 Source를 만든 뒤, 바로 Docling을 돌리지 않고 Celery 태스크만 넘긴다. HTTP 응답은 "업로드 완료" 수준으로 빠르게 끝나고, 무거운 분석은 워커가 이어받는다.

# 파일 저장 및 Source 등록 후
document_parse_task.delay(user_inst.pk, new_source.pk)

return JsonResponse(
    data={
        'message': f"파일 업로드 완료: {fm.file_name}",
        'source_id': new_source.pk
    },
    status=200,
)

태스크 함수는 @shared_task 로 정의한다. bind=True 를 주면 self.request.id 로 Celery 태스크 ID를 알 수 있어, 웹 소켓 알림에 함께 실을 수 있다.

@shared_task(bind=True)
def document_parse_task(self, user_id: int, source_id: int):
    source_inst = Source.objects.get(id=source_id)
    try:
        # Docling 문서 분석 → 이미지/컨텍스트 저장 → 청킹 → FAISS 업데이트
        ...
        source_inst.status = 'COMPLETED'
        source_inst.save()
    except Exception as e:
        source_inst.status = 'FAILED'
        source_inst.save()
        raise e

같은 방식으로 웹 크롤링(web_crawl_task), 소스 수정 및 VLM 캡션 반영(update_modify_task)도 워커에서 처리한다. 공통적으로 Source의 status 필드를 PENDING → COMPLETED / FAILED 로 갱신한다.

워커 실행 옵션: threads 풀

Docker Compose에서 Celery 워커는 다음과 같이 실행한다.

celery -A markit worker --loglevel=info --pool=threads --concurrency=2

CUDA Context 충돌

Celery 기본값인 prefork(멀티 프로세스)는 자식 프로세스가 부모의 자원을 복사하려 하면서 CUDA Context 충돌을 일으킬 수 있다. Docling처럼 GPU를 쓰는 작업이 워커에 포함되어 있으므로 --pool=threads 로 같은 프로세스 안에서 스레드로 나누는 편이 안전하다.

--concurrency=2 는 동시에 처리할 작업 수이며, 관리자가 늘면 VRAM을 고려해 조정할 수 있다. 정리하면 Celery 워커 컨테이너 자체는 웹과 분리된 별도 프로세스이지만, 그 안에서 GPU 작업은 멀티 스레드로 돌리는 구조이다.

웹 소켓 구현

비동기로 작업을 넘기면 다음 문제가 생긴다. 처리가 언제 끝나는지 브라우저가 어떻게 알 것인가?

HTTP는 클라이언트가 요청하고 서버가 응답하는 1회성 통신이다. 폴링으로 상태를 계속 물어보는 방법도 있지만, 문서 분석처럼 수분 단위로 걸릴 수 있는 작업에는 비효율적이다. 그래서 클라이언트가 작업을 등록해 두고, 서버가 끝나면 알림을 밀어 주는 방식이 필요하다고 판단했고, 웹 소켓을 적용하기로 했다.

웹 소켓은 핸드셰이크로 브라우저-서버 간 연결을 연 뒤, 연결을 유지한 채 양방향 실시간 통신을 하는 프로토콜이다.

[HTTP]
  Client ---- Request ----> Server
  Client <--- Response ---- Server
  (연결 종료)

[WebSocket]
  Client <=== Handshake ===> Server
  Client <=== 상시 연결 ===> Server
  (WS / WSS, 서버가 먼저 메시지를 보낼 수 있음)

Django에서의 구성 요소

Django는 기본적으로 WSGI 기반이라 웹 소켓을 직접 다루기 어렵다. 프로젝트에서는 다음 조합으로 구동한다.

구성 요소	역할
Django Channels	웹 소켓 프로토콜을 Django에서 다룰 수 있게 해주는 확장 프레임워크
Daphne	ASGI 표준을 사용하는 비동기 서버
Redis	Channel Layer의 메시지 브로커. Celery 워커가 보낸 완료 메시지를 Consumer까지 중계한다

Django 웹 소켓 스택

ASGI 라우팅
HTTP와 WebSocket을 프로토콜별로 나눈다. WebSocket 요청은 AuthMiddlewareStack과 URLRouter를 거친다.
유저별 그룹 연결
/ws/task/{user_id}/ 로 접속한 클라이언트를 user_{id} 그룹에 가입시킨다.
워커에서 알림 전송
Celery 태스크 종료 시 group_send 로 상태 JSON을 보낸다.
Consumer가 브라우저로 전달
task_update 이벤트를 받아 WebSocket send 로 프론트에 푸시한다.

ASGI 진입점은 다음과 같다.

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(markit.routing.websocket_urlpatterns)
    ),
})

웹 소켓 URL은 사용자별로 그룹을 나누기 위해 user_id 를 경로에 넣었다.

websocket_urlpatterns = [
    re_path(r'ws/task/(?P<user_id>\w+)/$', consumers.TaskStatusConsumer.as_asgi()),
]

Consumer는 연결 시 해당 유저 그룹에 가입하고, Celery가 group_send 로 보낸 이벤트를 받아 브라우저로 JSON을 전달한다.

class TaskStatusConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.user_id = self.scope['url_route']['kwargs'].get('user_id', 'anonymous')
        self.group_name = f"user_{self.user_id}"
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        await self.channel_layer.group_discard(self.group_name, self.channel_name)

    async def task_update(self, event):
        await self.send(text_data=json.dumps({
            'source_id': event['source_id'],
            'task_id': event['task_id'],
            'status': event['status'],
        }))

Celery 태스크는 동기 코드이므로 async_to_sync 로 Channel Layer에 메시지를 보낸다. type: 'task_update' 는 Consumer의 task_update 메서드 이름과 대응한다.

cl = get_channel_layer()
group_name = f"user_{user_inst.pk}"

async_to_sync(cl.group_send)(
    group_name,
    {
        'type': 'task_update',
        'source_id': source_inst.pk,
        'task_id': self.request.id,
        'status': source_inst.status,
    }
)

워커 → Redis → Consumer → 클라이언트로 이어지는 알림 흐름은 아래와 같다.

[Celery Worker]
    |  작업 완료 후 group_send(status)
    v
[Redis Channel Layer]
    |  유저 그룹(user_{id})으로 전달
    v
[Django Consumer]
    |  WebSocket send(JSON)
    v
[브라우저]

문서 업로드부터 완료 알림까지를 한 장으로 그리면 다음과 같다.


sequenceDiagram
  actor User as 사용자
  participant Web as Django Web
  participant Redis as Redis
  participant Worker as Celery Worker
  participant Store as Ollama / FAISS

  User->>Web: 문서 업로드
  Web->>Redis: 태스크 등록
  Web-->>User: 업로드 응답
  Redis->>Worker: 태스크 전달
  Worker->>Store: 문서 분석 / 임베딩
  Store-->>Worker: 저장 완료
  Worker->>Redis: WS 완료 메시지
  Redis->>Web: Channel Layer 전달
  Web-->>User: 완료 알림 (WebSocket)

Redis의 이중 역할

Redis는 Celery 브로커로서 태스크 큐를 관리하고, Channel Layer로서 워커가 만든 완료 이벤트를 웹 소켓 Consumer로 넘겨 준다. 컨테이너를 하나로 유지하면서 비동기 작업과 실시간 알림을 동시에 지탱하는 구조이다.

처리 완료 인터페이스

백엔드에서 상태를 바꿔도 화면에 반영되지 않으면 사용자는 여전히 대기 화면만 보게 된다. 그래서 Source 모델에 상태를 두고, 템플릿과 프론트엔드가 이를 바로 읽도록 맞추었다.

상태	의미	UI
`PENDING`	대기 / 처리 중	스피너 표시, 검토/수정 버튼 비활성
`COMPLETED`	성공	성공 뱃지, 검토/수정 가능
`FAILED`	실패	실패 뱃지

Source.status 값과 화면 동작

문서·링크 목록 테이블의 상태 열은 PENDING일 때 스피너를 보여 주고, 완료/실패일 때 상태 뱃지를 표시한다. 처리가 끝나기 전에는 검토/수정 버튼을 비활성화해 중간 상태의 소스를 편집하지 못하게 했다.

{% if source.status == 'PENDING' %}
  <div class="spinner-border spinner-border-sm text-secondary" role="status"></div>
{% elif source.status == 'COMPLETED' %}
  <span class="kb-status-pill-success">{{ source.status }}</span>
{% elif source.status == 'FAILED' %}
  <span class="kb-status-pill-failed">{{ source.status }}</span>
{% endif %}

프론트엔드에서는 페이지 로드 시 사용자 ID로 웹 소켓에 연결한다. 완료 메시지가 오면 목록을 다시 불러 스피너가 상태 뱃지로 바뀌도록, 단순하게 페이지를 새로고침했다.

import TaskSocket from '../common/websocket.js';

document.addEventListener("DOMContentLoaded", function () {
  const user = JSON.parse(document.getElementById("context-user").textContent);
  const socket = new TaskSocket(user.id);
  socket.on((data) => {
    // data: { source_id, task_id, status }
    location.reload();
  });
});

TaskSocket 은 싱글톤으로 한 연결을 재사용하고, 연결이 끊기면 재접속을 시도한다. URL은 HTTPS 여부에 따라 ws / wss 를 고른다.

connect() {
  const protocol = window.location.protocol === 'https:' ? 'wss' : 'ws';
  const url = `${protocol}://${window.location.host}/ws/task/${this.userId}/`;
  this.socket = new WebSocket(url);

  this.socket.onmessage = (e) => {
    const data = JSON.parse(e.data);
    this.handleMessage(data);
  };

  this.socket.onclose = () => {
    setTimeout(() => this.connect(), this.reconnectInterval);
  };
}

업로드
관리자가 파일을 올리면 Source가 PENDING으로 등록되고 Celery 태스크가 시작된다.
대기 UI
목록에 스피너가 뜨고, 검토/수정은 비활성 상태로 남는다.
완료 알림
워커가 끝나면 웹 소켓으로 status가 전달되고 페이지가 새로고침된다.
후속 작업
COMPLETED 뱃지가 보이면 검토/수정으로 VLM 캡션 반영 등을 이어갈 수 있다.

결과적으로 관리자는 파일을 올리고 바로 다른 작업을 할 수 있고, 분석이 끝나면 목록의 스피너가 사라지고 COMPLETED 뱃지가 나타나며 검토/수정이 가능해진다. HTTP 요청-응답만으로는 어려웠던 "백그라운드 작업 + 실시간 완료 알림"을 Celery와 웹 소켓으로 나눈 것이 이 단계의 핵심이었다.

다음 포스트

응답 품질 개선