[Django + RabbitMQ + Celery + Redis] Celery를 활용한 분산 비동기 작업 처리
MVStudio Github
6주간 부트캠프에서 진행했던 프로젝트 MVStudio에 Celery를 활용하여 분산 비동기처리를 한 경험을 포스팅하려고 한다.
Celery란 무엇일까?
Celery 공식문서에서 메인페이지에 적혀있는 글을 가져와보았다.
Celery - Distributed Task Queue
Celery is a simple, flexible, and reliable distributed system to process vast amounts of messages, while providing operations with the tools required to maintain such a system.
"셀러리는 방대한 양의 메시지를 처리할 수 있는 간단하고, 유연하며, 안정적인 분산 시스템이며, 이러한 시스템을 유지하는 데 필요한 도구를 운영에 제공합니다."
한마디로 방대한 양의 메시지를 분산 처리 할 수 있는 분산 작업 대기열이다.
작업 대기열이란 무엇일까?
작업 대기열은 작업을 여러 스레드 또는 머신에 분산하는 메커니즘으로 사용된다. 큐를 생각하면 쉽다. 작업 큐로 작업을 수행하라는 메세지가 순차적으로 들어오게 되고 먼저 들어온 메세지가 먼저 나가게 된다.
이때 Message를 제공하는 역할을 Producer, Message Queue를 Broker, Message를 가져가는 역할을 Consumer라고 칭한다.
Producer가 메시지를 메시지 큐에 넣게 되면 Consumer가 먼저 들어온 작업을 순차적으로 가져가게 된다. 이때 각각의 작업은 여러개의 Consumer를 통해 작업이 처리될 수 있다. 이것을 분산 작업 처리가 된다고 한다.
이때 Consumer로써 Celery Worker들이 동작한다고 생각하면 된다. 여기서 Worker는 작업을 처리하는 하나의 객체로 생각하면 쉽다. 그리고 Celery는 Broker로 크게 RabbitMQ와 Redis를 사용한다.
RabbitMQ와 Redis의 차이점을 확인해보고 각각 어떤 상황에 사용하는지 생각해보자.
1. 메시지 처리 방식
RabbitMQ는 Advanced Message Queuing Protocol(AMQP)을 사용하여 복잡한 라우팅 로직을 지원한다. RabbitMQ는 모든 Consumer가 메시지를 받았는지 확인하는 작업을 수행하며 확인 받지 못하면 여러 번 재시도를 하여 메시지 전송을 보장한다.
반면, Redis는 단순히 메시지를 푸시할 뿐, 메시지 전송을 보장하지 않는다.
2. 속도
Redis는 주로 메모리에서 메시지를 처리하므로 RabbitMQ보다 훨씬 빠르다. 반대로 RabbitMQ는 메시지를 디스크에 저장하는 데 시간이 더 걸리므로 평균적인 메시지 교환속도가 더 느리다. Redis는 초당 최대 수천만 개의 메시지를 전송할 수 있는 반면, RabbitMQ는 초당 최대 수만 개의 메시지만 처리할 수 있다.
그럼 각각 어떤 상황에서 사용이 될까?
Redis는 비교적 단순한 짧은 캐싱이 요구되는 애플리케이션에서 더 뛰어난 성능을 제공한다. 반면 RabbitMQ는 복잡한 라우팅 기능이 필요하고 메시지 전송이 보장되어야 할때 사용된다.
이제 우리 프로젝트 상황에서 한번 보도록 하자.
나는 Producer로 Django 백엔드 서버가 있고 Broker로써 RabbitMQ, Consumer로 Celery를 사용하였다. 왜 이렇게 사용했는지를 설명하기 전에 우리 프로젝트에서의 프로세스를 간략하게 설명해보겠다.
우리 프로젝트는 MVStudio라는 프로젝트이다. 사용자가 원하는 주제를 선택하면 AI를 통해 뮤직비디오를 생성하고 사용자에게 제공해주는 서비스이다. 뮤직비디오 제작 과정은 다음과 같다.
- 사용자가 주제 및 가사를 선택한다.
- 주제 및 가사를 토대로 SunoAI를 사용하여 음악을 생성한다. 동시에 가사를 토대로 Text to Video 모델인 RunwayML의 Gen3 모델을 통해 8개의 비디오를 제작한다.
- 최종적으로 완성된 음악과 비디오를 합쳐서 뮤직비디오를 완성한다.
크게 3가지의 프로세스로 확인할 수 있다.
음악을 생성, 비디오를 생성하는 작업은 AI를 통해 처리되며 긴 작업 시간을 갖게된다. 최종적으로 뮤직비디오 제작 또한 영상 편집이다 보니 긴 작업시간을 갖는다.
이런 무거운 작업을 Django 백엔드 서버에서 온전히 부담하게 되면 다른 작업을 수행할 수 없고 사용자 또한 상당히 많은 시간을 기다리게 된다.
이러한 문제점을 Celery를 통해 해결 할 수 있다.
첫번째로 Django 백엔드 서버에서 동기적으로 작업이 실행되면 다른 작업이 수행 될 수 없지만 비동기로 실행되면 다른 작업이 수행 될 수 있다. 이를 Celery에게 작업을 맡기면 해결 할 수 있다.
두번째로 많은 시간 소모이다. 우리 프로젝트에서는 8개의 비디오 제작, 1개의 음악 제작이 이루어지게 된다. 이 작업들은 각각 병렬로 처리되어도 상관없는 작업이다. Celery를 통해 9개의 작업을 병렬로 처리하게 되면 상당히 많은 시간이 단축 될 수 있다.
분산 작업 처리
그럼 Celery는 어떤 방식으로 분산 작업 처리가 진행이 될까?
총 2가지의 방법이 있다.
- 멀티프로세싱
- 멀티스레딩
멀티프로세싱
먼저 멀티프로세싱 방식은 celery가 실행하는 프로세스의 수를 늘리는 방법이다. 즉, 여러개의 celery worker를 실행하여 각 프로세스별로 테스크를 실행하는 것이다.(하나의 celery worker에서 --pool=prefork 옵션을 통해서도 멀티프로세싱이 가능하다.)
이 방식은 CPU-bound 작업, 즉, CPU 자원을 많이 사용하는 작업에 유용하다.
고려해야할 점은 프로세스의 수가 많아지게되면 메모리 및 CPU의 사용량이 상대적으로 증가하므로 요구사항을 적절히 고려하여 동시성을 설정해야 한다.
멀티스레딩
celery는 2개의 스레드 기반 celery pool을 제공한다. (eventlet, gevent)
Green Thread 기반의 동시성 옵션으로, 어플리케이션 레벨에서 스케줄링하는 가벼운 스레드를 의미한다. 이런 방식은 I/O bound 작업, 특히 network I/O bound 작업에 적합하다. 이 옵션을 사용하게 되면, 한 worker 프로세스 내에서 수백, 수천개의 작업을 동시에 처리할 수 있다.
다음 명령어를 통해 사용할 수 있다.
celery -A {프로젝트 이름} worker -l info --pool=gevent --concurrency={숫자}나는 이 2가지 방법을 모두 사용하였다.
먼저 음악 제작, 비디오 제작, 뮤직비디오 제작 3가지의 작업은 각각 독립적인 작업이므로 각 작업별로 worker를 지정해주었다. 특히, 뮤직비디오 제작은 CPU 자원을 많이 소모하므로 독립적인 프로세스에서 진행되는 것이 옳다고 생각하였다.
비디오 제작은 RunwayML Gen3 모델서버에 API 요청을 하여서 폴링 방식을 통해 작업이 성공적으로 수행됐는지 확인하는 작업이 주로 이루어져 있다. 이 작업은 대부분 네트워크 대기(응답을 기다리는 시간)에 사용되므로 Network I/O bound 작업에 가깝다고 생각하였다. 따라서 총 8개의 비디오 제작 작업을 병렬적으로 처리할 때 멀티스레딩 방식이 옳다고 생각하였다.
지금까지의 상황을 간단히 그려보았다.
문제점 발생
여기서 여러가지 문제점을 발견 할 수 있었다.
첫번째로, 작업 순서의 비일관성이다.
비디오 제작과 음악 제작이 이루어진 후에 뮤직비디오 제작이 이루어져야 한다. 하지만 현재 그림상으로는 순서를 보장 할 수 없다. 그래서 사용하게 된 것은 celery의 group과 chord이다.
Group 이란?
말 그대로 하나의 집합을 만들어 실행시키는 작업으로 Task를 병렬로 실행 할 수 있다. 8개의 비디오를 제작하는 과정은 병렬로 실행 될 수 있는 작업이므로 group 함수를 사용하여 묶어주었고 결과는 리스트로 반환된다.
chord 란?
사전적 의미로는 화음이라고 하며, 그룹의 모든 작업이 완료된 후에만 실행되는 작업이다. 음악 제작과 비디오 제작이 전부 완료된 후에 뮤직비디오 제작 작업에 들어가야 하므로 chord 함수를 사용하였다. 이때, 주의해야 할 점이 있다. 결과값을 무조건 반환받아야하며 Result Backend가 필요하다. (여기서 Result Backend는 결과값을 저장해두는 공간이며 나는 Redis를 사용하였다.)
나의 경우 음악 제작 작업이 완료된 후 제작된 음악과 비디오 제작 작업이 완료된 후 제작된 8개의 비디오가 전부 뮤직비디오 제작에 필요하였다. 따라서 chord 함수를 사용하여 음악 제작과 비디오 제작의 결과를 Result Backend에 저장하였고 전부 작업이 완료되고 저장되면 뮤직비디오 제작 작업이 들어가도록 하였다.
두번째 문제점은 작업 간의 간섭 발생과 사용자 요청의 불공정한 처리이다.
하나의 큐를 사용하면 여러 Task가 동일한 큐에 쌓이게 되는데 특정 Task가 오래걸리거나 자원을 많이 사용하는 경우 다른 Task들이 대기 상태로 지연될 수 있다. 또한 여러 사용자가 동시에 작업을 요청하는 경우, 각 사용자간에 작업이 뒤섞이며 먼저 요청한 사용자가 뒤늦게 요청이 처리 될 수 있는 상황이 발생할 수 있다. 이러한 문제점을 해결하기 위해서 각각의 작업별로 Queue를 할당해주었고 RabbitMQ의 Exchange를 활용하였다. (내가 RabbitMQ를 Broker로 사용한 이유이기도 하다.)
RabbitMQ의 Exchange란?
모든 메시지는 Queue로 직접 전달되지 않고, 반드시 Exchange에서 먼저 받는다. 그리고 Exchange Type과 Binding 규칙에 따라 적절한 Queue로 전달된다.
메시지 전달방식에는 크게 4가지의 방식이 있는데 나는 Direct Exchange 방식을 사용하였다. Direct Exchange 방식은 메시지에 포함된 routing key를 기반으로 Queue에 메시지를 전달하는 방식이다. 이를 통해 각각의 작업별로 큐를 할당해 줄 수 있었다.
최종적으로 다음과 같은 구조가 완성 될 수 있었다.
결론
이 사진은 Celery Monitoring Tool인 Flower를 통해 확인한 각 Task 실행 모습이다. 최종적으로 평균 50분이 걸리는 기능을 약 5분이내로 (약10배) 단축하는데 성공하였다.
내가 6주간 프로젝트를 하면서 가장 많이 고민하고 시간을 들였던 부분이었다. 그 이유는 우리 서비스의 메인 기능이기도 하며 가장 어려운 기능이기도 했기 때문이다.
그리고 내가 여기서도 가장 신경 쓴 부분은 바로 근거이다. 모든 기술의 사용은 합당한 근거가 있어야 한다고 생각한다. 내가 사용한 기술들을 하나하나 이해하고 효율적으로 사용하기 위해 노력을 많이 했다.
시간을 많이 소모한 만큼 배울 수 있는게 가장 많았고 성장 할 수 있었던 것 같다. 기술을 하나하나 알아가고 제대로 사용하면서 결과 또한 좋아지니까 정말 재밌는 경험이었다.
혹시나 설명이 부족했을 수도 있으므로 이해를 돕기 위해 최종발표 준비할 때 제작되었던 영상자료를 공유하면서 포스팅을 마치도록 하겠다.
