모르는 상태로 하는 RSS&분석&RAG 프로젝트(5) Celery+Redis

오늘은 Celery와 Redis를 함께 사용해서 간단한 사용법을 알아볼 계획이다.

 

Celery와 Redis가 어떻게 동작하는지 어떤 시너지를 내는건지 아직 잘 모르기에 AI에게 간단한 테스트 과정을 시켜보았다.

 

먼저 이전에 간단하게 설명한것처럼

Celery는 Python에서 비동기작업을 할 수 있게 해주는 도구이며

Redis는 inmemory에서 작업할 수 있게 해주며 시간을 지정해두고 한번 저장한 것에 대해서는 빠르게 꺼내올 수 있는 도구이다.

 

Celery의 비동기와 Redis의 inmemory를 활용하여 작업하면 효율이 엄청나게 증가한다.

 

먼저 간단한 테스트 폴더를 만들라고 하니까 파일을 몇가지 만들었다. 

 

 📁 파일 구조

 

celery_redis/

├── celery_app.py       Celery 앱 설정 (Redis 연결)

├── tasks.py            작업 정의 (실제로 실행할 함수들)

├── easy_example1.py    예제 1: 기본 이해

├── easy_example2.py    예제 2: 비유로 이해

├── easy_example3.py    예제 3: 단계별 따라가기

├── easy_example4.py    예제 4: 여러 작업 동시 실행

└── requirements.txt    필요한 패키지

 

celery를 실행하는 파일과 함께 함수로 어떤 동작을 할건지 만드는 tasks파일, 샘플파일로 구성되어있으며

 

 🔗 파일들의 관계

 

celery_app.py (설정)

    ↑

    │ import

    │

tasks.py (작업 정의)

    ↑

    │ import

    │

easy_example1.py (실행)

 

으로 상속받고 정의하는 순서로 되어있다. 

 

즉. celery_app.py는 Celery 설정을 하고 

tasks.py는 어떤 작업을 할 것인지에 대한 작업을 정의

example.py들은 코드로 만들어놓은 작업을 tasks를 상속받아 실행한다.

 

먼저 Celery 설정을 하는 celery_app.py에는 

"""
Celery 앱 설정 파일
이 파일은 Celery 인스턴스를 생성하고 Redis를 브로커로 사용하도록 설정합니다.
"""
from celery import Celery

# Celery 앱 생성
# broker: 작업을 보내는 곳 (Redis)
# backend: 결과를 저장하는 곳 (Redis)
celery_app = Celery(
    'celery_redis_example',  # 프로젝트 이름
    broker='redis://localhost:6379/0',  # Redis 브로커 URL
    backend='redis://localhost:6379/0'  # Redis 백엔드 URL
)

# Celery 설정
celery_app.conf.update(
    task_serializer='json',  # 작업을 JSON 형식으로 직렬화
    accept_content=['json'],  # JSON만 허용
    result_serializer='json',  # 결과를 JSON 형식으로 직렬화
    timezone='Asia/Seoul',  # 시간대 설정
    enable_utc=True,  # UTC 사용
    # 작업 모듈 자동 import 설정 (Worker가 tasks.py를 자동으로 로드)
    include=['tasks'],  # tasks.py 모듈을 자동으로 import
)

이렇게 프로젝트 이름과 함께 broker와 backend를 설정하고 

celery에 관련된 설정을 해둔다.

 

tasks.py에는 

"""
Celery 작업 정의 파일
여기에 비동기로 실행할 작업들을 정의합니다.
"""
from celery_app import celery_app
import time


@celery_app.task(name='tasks.add')
def add(x, y):
    """
    간단한 덧셈 작업
    """
    print(f"덧셈 작업 실행: {x} + {y}")
    result = x + y
    print(f"결과: {result}")
    return result


@celery_app.task(name='tasks.multiply')
def multiply(x, y):
    """
    간단한 곱셈 작업
    """
    print(f"곱셈 작업 실행: {x} * {y}")
    result = x * y
    print(f"결과: {result}")
    return result


@celery_app.task(name='tasks.long_running_task')
def long_running_task(seconds):
    """
    시간이 오래 걸리는 작업 시뮬레이션
    실제로는 데이터 처리, 파일 다운로드, API 호출 등에 사용됩니다.
    """
    print(f"긴 작업 시작: {seconds}초 동안 대기합니다...")
    for i in range(seconds):
        time.sleep(1)
        print(f"진행 중... {i+1}/{seconds}초")
    print(f"긴 작업 완료!")
    return f"{seconds}초 작업이 완료되었습니다."


@celery_app.task(name='tasks.process_data')
def process_data(data_list):
    """
    데이터 처리 작업 예제
    리스트의 각 항목을 처리합니다.
    """
    print(f"데이터 처리 시작: {len(data_list)}개 항목")
    results = []
    for item in data_list:
        # 간단한 처리 예제 (실제로는 복잡한 로직이 들어갑니다)
        processed = item * 2
        results.append(processed)
        print(f"처리: {item} -> {processed}")
    print(f"데이터 처리 완료: {results}")
    return results

 

 

이 처럼 celery로 실행할 작업들을 정의하는 파일이다.

하지만 생각해보면 이걸 굳이 따로 작성해둘 필요가 있을까 싶지만,

Celery 작업은 일반 함수처럼 동작하지않고 worker가 나중에 실행하기에 worker가 어떤 함수를 실행할 지 알아야 하므로 따로 작성하는것이 맞고 tasks.py에 작업을 정의하면 worker가 자동으로 찾아서 실행하는 방식이다. 

 

이제 easy_example파일중에 1,2,4만 보면 될거같아서 작성하면

"""
예제 1: 가장 기본적인 이해
일반 함수 vs Celery 작업의 차이를 보여줍니다.
"""
from tasks import add

print("=" * 60)
print("예제 1: 일반 함수 vs Celery 작업")
print("=" * 60)
print()

# 방법 1: 일반 함수 (동기 방식)
print("【방법 1】 일반 함수 호출 (동기)")
print("프로그램이 여기서 멈춥니다...")
result1 = add(10, 20)  # 일반 함수처럼 호출 (Celery 사용 안 함)
print(f"결과: {result1}")
print("위의 함수가 끝나야 여기까지 올 수 있어요!")
print()

# 방법 2: Celery 작업 (비동기 방식)
print("【방법 2】 Celery 작업 호출 (비동기)")
print("프로그램이 멈추지 않아요!")
result2 = add.delay(10, 20)  # .delay()를 붙이면 Celery 작업!
print(f"작업 ID: {result2.id}")
print("바로 여기까지 왔어요! Worker가 처리하는 동안 기다리지 않았어요!")
print()

print("이제 결과를 가져올게요...")
final_result = result2.get()  # 결과를 가져올 때만 기다림
print(f"최종 결과: {final_result}")
print()

print("=" * 60)
print("차이점:")
print("- 일반 함수: 함수가 끝날 때까지 기다림 (블로킹)")
print("- Celery 작업: 즉시 다음 코드 실행 (논블로킹)")
print("=" * 60)

이걸 실행하기 위해선 먼저 celery를 실행 시키고 난뒤에 새로운 터미널로 이동해서 실행해야한다.

아래는 터미널 내용이다. 

$ python easy_example1.py
============================================================
예제 1: 일반 함수 vs Celery 작업
============================================================

【방법 1】 일반 함수 호출 (동기)
프로그램이 여기서 멈춥니다...
덧셈 작업 실행: 10 + 20
결과: 30
결과: 30
위의 함수가 끝나야 여기까지 올 수 있어요!

【방법 2】 Celery 작업 호출 (비동기)
프로그램이 멈추지 않아요!
작업 ID: a1ea3295-b23b-4286-8394-39764cd0cc5e
바로 여기까지 왔어요! Worker가 처리하는 동안 기다리지 않았어요!

이제 결과를 가져올게요...
최종 결과: 30

============================================================
차이점:
- 일반 함수: 함수가 끝날 때까지 기다림 (블로킹)
- Celery 작업: 즉시 다음 코드 실행 (논블로킹)
============================================================

 

차이가 조금 있는것이 보인다. celery가 어떻게 동작하는지에 대한 아주 간단한 내용이다. 

.delay를 하지않으고 add만 있을경우에는 일반 함수랑 동일하게 위에서 아래로 그대로 이어져 내려올 뿐이다. 

하지만, delay를 걸게 되면 .get으로 가져올때까지 얌전히 잘 기다리는걸 볼 수 있다. 

 

두번째 예제로 보면 

"""
예제 2: 비유로 이해하기
레스토랑 주문 시스템으로 비유합니다.
"""
from tasks import add, multiply
import time

print("=" * 60)
print("예제 2: 레스토랑 비유")
print("=" * 60)
print()

print("🍽️  레스토랑에 왔어요!")
print()

# 주문 1: 파스타 주문
print("손님: '파스타 하나 주세요!'")
pasta_order = add.delay(5, 3)  # 주문서를 카운터에 제출
print(f"직원: '네, 주문번호는 {pasta_order.id}입니다. 잠시만 기다려주세요!'")
print("손님: '알겠어요! 그럼 다른 것도 주문할게요!'")
print()

# 주문 2: 피자 주문 (파스타가 나오기 전에!)
print("손님: '피자도 하나 주세요!'")
pizza_order = multiply.delay(4, 7)  # 또 다른 주문
print(f"직원: '네, 주문번호는 {pizza_order.id}입니다!'")
print()

# 피자는 빨리 나왔어요
print("피자가 먼저 나왔어요!")
pizza_result = pizza_order.get()
print(f"피자 주문 완료! 결과: {pizza_result}")
print()

# 파스타는 아직 조리 중...
print("파스타는 아직 조리 중이에요. 기다려볼게요...")
pasta_result = pasta_order.get()
print(f"파스타 주문 완료! 결과: {pasta_result}")
print()

print("=" * 60)
print("핵심:")
print("- 주문(.delay())은 즉시 받아들여짐")
print("- 여러 주문을 동시에 할 수 있음")
print("- 결과(.get())는 준비되면 받을 수 있음")
print("=" * 60)

 print가 많아서 복잡해 보이지만 전부 터미널에 순서대로 잘 되고있는지 확인할 수 있도록 만든것이고 중요한것은 

파스타 주문이 먼저 들어갔지만 결과는 피자가 더 빨리 나올수 있게도 할 수 있다는것이다. 

============================================================
예제 2: 레스토랑 비유
============================================================

🍽️  레스토랑에 왔어요!

손님: '파스타 하나 주세요!'
직원: '네, 주문번호는 2e022ae7-c1b7-4da9-988a-fef1fd04812a입니다. 잠시만 기다려주세요!'
손님: '알겠어요! 그럼 다른 것도 주문할게요!'

손님: '피자도 하나 주세요!'
직원: '네, 주문번호는 f88cd40e-8f58-4260-b1c3-d6a70d5c7929입니다!'      

피자가 먼저 나왔어요!
피자 주문 완료! 결과: 28

파스타는 아직 조리 중이에요. 기다려볼게요...
파스타 주문 완료! 결과: 8

============================================================
핵심:
- 주문(.delay())은 즉시 받아들여짐
- 여러 주문을 동시에 할 수 있음
- 결과(.get())는 준비되면 받을 수 있음
============================================================

위에 터미널을 살펴보면 각 주문번호인 .id를 넣어서 확인할 수 있게 된다면 .get으로 가져오게 되는 순서도 변경 할 수 있다는걸 알 수 있다. 

 

예제4를 살펴보면 celery의 worker의 좋은점을 더 직관적으로 알 수 있다. 

"""
예제 4: 여러 작업 동시에 하기
왜 Celery가 유용한지 보여줍니다.
"""
from tasks import add, multiply, long_running_task
import time

# 일반 함수 버전 (비교용)
def normal_add(x, y):
    return x + y

def normal_multiply(x, y):
    return x * y

def normal_long_task(seconds):
    for i in range(seconds):
        time.sleep(1)
    return str(seconds) + "초 작업이 완료되었습니다."

print("=" * 60)
print("예제 4: 여러 작업 동시에 하기")
print("=" * 60)
print()

# 먼저 일반 함수로 측정
print("【일반 함수 방식】")
normal_start = time.time()
result1_normal = normal_add(10, 20)
result2_normal = normal_multiply(5, 6)
result3_normal = normal_long_task(5)
normal_total = time.time() - normal_start
print("일반 함수 총 시간: " + str(round(normal_total, 3)) + "초")
print("결과: " + str(result1_normal) + ", " + str(result2_normal) + ", " + str(result3_normal))
print()
print("=" * 60)
print()

print("📌 시나리오: 3가지 일을 해야 해요")
print("   1. 간단한 계산 (빠름)")
print("   2. 곱셈 계산 (빠름)")
print("   3. 긴 작업 (5초)")
print()

# 모든 작업을 한 번에 시작!
print("🚀 모든 작업을 한 번에 시작!")
start_time = time.time()
task1 = add.delay(10, 20)
task2 = multiply.delay(5, 6)
task3 = long_running_task.delay(5)
queue_time = time.time() - start_time

print("작업 1 ID: " + task1.id)
print("작업 2 ID: " + task2.id)
print("작업 3 ID: " + task3.id)
print("큐에 넣는 데 걸린 시간: " + str(round(queue_time, 1)) + "초")
print()

print("💡 핵심: 모든 작업이 백그라운드에서 동시에 실행되고 있어요!")
print()

# 빠른 작업들 먼저 결과 받기
print("빠른 작업 결과 받기:")
task1_start = time.time()
result1 = task1.get()
task1_time = time.time() - task1_start
print("  작업 1 결과: " + str(result1) + " (걸린 시간: " + str(round(task1_time, 3)) + "초)")

task2_start = time.time()
result2 = task2.get()
task2_time = time.time() - task2_start
print("  작업 2 결과: " + str(result2) + " (걸린 시간: " + str(round(task2_time, 3)) + "초)")
print()

print("긴 작업은 아직 진행 중...")
print("하지만 우리는 다른 일을 할 수 있어요!")
print()

# 긴 작업 결과 받기
print("긴 작업 결과 받기:")
task3_start = time.time()
result3 = task3.get()
task3_time = time.time() - task3_start
print("  작업 3 결과: " + str(result3) + " (걸린 시간: " + str(round(task3_time, 3)) + "초)")
print()

total_time = time.time() - start_time
print("=" * 60)
print("【Celery 실제 측정된 시간】")
print("  - 큐에 넣는 시간: " + str(round(queue_time, 3)) + "초")
print("  - 작업 1 처리 시간: " + str(round(task1_time, 3)) + "초")
print("  - 작업 2 처리 시간: " + str(round(task2_time, 3)) + "초")
print("  - 작업 3 처리 시간: " + str(round(task3_time, 3)) + "초")
print("  - 총 소요 시간: " + str(round(total_time, 3)) + "초")
print()
print("=" * 60)
print("비교 결과:")
print("=" * 60)
print("일반 함수: " + str(round(normal_total, 3)) + "초")
print("Celery: " + str(round(total_time, 3)) + "초")
print()
if total_time > normal_total:
    diff = total_time - normal_total
    print("⚠️ Celery가 " + str(round(diff, 3)) + "초 더 느립니다.")
    print()
    print("왜 느릴 수 있나요?")
    print("1. Redis 연결 오버헤드 (네트워크 통신)")
    print("2. 작업 직렬화/역직렬화 (JSON 변환)")
    print("3. Worker 프로세스 간 통신")
    print("4. 간단한 작업은 오버헤드가 더 큼")
    print()
    print("💡 하지만 중요한 차이점:")
    print("- 일반 함수: 프로그램이 " + str(round(normal_total, 3)) + "초 동안 완전히 멈춤")
    print("- Celery: 큐에 넣는 데 " + str(round(queue_time, 3)) + "초만 걸리고 바로 다음 코드 실행 가능!")
    print("  → 프로그램이 멈추지 않아서 다른 일을 할 수 있음!")
else:
    print("✅ Celery가 더 빠릅니다!")
print()
print("=" * 60)
print("결론:")
print("간단한 작업은 일반 함수가 더 빠를 수 있지만,")
print("Celery의 진짜 장점은 '프로그램이 멈추지 않는다'는 것입니다!")
print("=" * 60)

이 코드는 실제로는 일반함수가 더 빠르게 동작한다. 

하지만 celery를 사용하면 좋은점은 병렬식동작 보단 비동기에 있다. 

일반 함수같은경우 한번 실행되면 끝나기 전까진 다른 동작을 할 수 없게끔 되어있는데 celery는 이걸 무시한채 다른 동작도 할 수 있게 만들어준다.

============================================================
예제 4: 여러 작업 동시에 하기
============================================================

【일반 함수 방식】
일반 함수 총 시간: 5.003초
결과: 30, 30, 5초 작업이 완료되었습니다.

============================================================

📌 시나리오: 3가지 일을 해야 해요
   1. 간단한 계산 (빠름)
   2. 곱셈 계산 (빠름)
   3. 긴 작업 (5초)

🚀 모든 작업을 한 번에 시작!
작업 1 ID: 93dcd963-d684-44bd-a745-e6e3235fdc92
작업 2 ID: 9fe8f43a-bb19-42c1-af3c-5c31db43427d
작업 3 ID: f12d2f3b-ea07-456f-a791-b9e395b93b43
큐에 넣는 데 걸린 시간: 6.3초

💡 핵심: 모든 작업이 백그라운드에서 동시에 실행되고 있어요!

빠른 작업 결과 받기:
  작업 1 결과: 30 (걸린 시간: 2.055초)
  작업 2 결과: 30 (걸린 시간: 0.0초)

긴 작업은 아직 진행 중...
하지만 우리는 다른 일을 할 수 있어요!

긴 작업 결과 받기:
  작업 3 결과: 5초 작업이 완료되었습니다. (걸린 시간: 2.96초)

============================================================
【Celery 실제 측정된 시간】
  - 큐에 넣는 시간: 6.297초
  - 작업 1 처리 시간: 2.055초
  - 작업 2 처리 시간: 0.0초
  - 작업 3 처리 시간: 2.96초
  - 총 소요 시간: 11.314초

============================================================
비교 결과:
============================================================
일반 함수: 5.003초
Celery: 11.314초

⚠️ Celery가 6.311초 더 느립니다.

왜 느릴 수 있나요?
1. Redis 연결 오버헤드 (네트워크 통신)
2. 작업 직렬화/역직렬화 (JSON 변환)
3. Worker 프로세스 간 통신
4. 간단한 작업은 오버헤드가 더 큼

💡 하지만 중요한 차이점:
- 일반 함수: 프로그램이 5.003초 동안 완전히 멈춤
- Celery: 큐에 넣는 데 6.297초만 걸리고 바로 다음 코드 실행 가능!       
  → 프로그램이 멈추지 않아서 다른 일을 할 수 있음!

============================================================
결론:
간단한 작업은 일반 함수가 더 빠를 수 있지만,
Celery의 진짜 장점은 '프로그램이 멈추지 않는다'는 것입니다!
============================================================

 

오늘은 이처럼 celery+redis를 활용해서 간단하게 테스트 및 활용방법에 대해서 알아보았다. 

내일은 RSS를 살펴볼 계획이다. RSS를 살펴본 이후에는 실제 프로젝트에 적용해서 

celery+redis까지 넣어 기사 및 핵심뉴스를 비동기로 수집하고 짧은 시간에 많은 기사를 긁어모을수 있도록 해볼 예정이다.