Background Jobs (Celery, BullMQ)백그라운드 작업

클라이언트 → 웹 서버 → 작업 큐 (Redis/RabbitMQ)
                              ↓
                         워커 프로세스 (병렬 실행)
                              ↓
                         결과 저장 (DB/캐시)

# celery_app.py
from celery import Celery
from celery.utils.log import get_task_logger

app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

app.conf.update(
    task_serializer='json',
    result_expires=3600,
    worker_concurrency=4,
    task_acks_late=True,  # 처리 후 ACK (at-least-once)
)

logger = get_task_logger(__name__)

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email(self, user_id: int, template: str):
    try:
        user = User.objects.get(id=user_id)
        EmailService.send(user.email, template)
        logger.info(f"이메일 발송 완료: {user.email}")
    except Exception as exc:
        logger.error(f"이메일 발송 실패: {exc}")
        raise self.retry(exc=exc)

# 작업 등록
result = send_email.delay(user_id=123, template='welcome')
result.get(timeout=10)  # 결과 조회 (선택적)

# 지연 실행
from datetime import timedelta
send_email.apply_async(args=[456, 'reset'], countdown=300)  # 5분 후

# 주기적 작업 (Celery Beat)
app.conf.beat_schedule = {
    'daily-report': {
        'task': 'tasks.generate_report',
        'schedule': 86400.0,  # 24시간마다
    },
}

import { Queue, Worker, QueueEvents } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis({ host: 'localhost', port: 6379 });

// 큐 생성
const emailQueue = new Queue('email', {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: 100,  // 완료된 작업 100개만 유지
  },
});

// 작업 추가
await emailQueue.add('send-welcome', {
  userId: 123,
  template: 'welcome',
}, {
  delay: 5000,      // 5초 후 실행
  priority: 1,      // 우선순위
});

// 워커
const worker = new Worker('email', async (job) => {
  const { userId, template } = job.data;
  await sendEmail(userId, template);
  return { sent: true };
}, { connection, concurrency: 10 });

worker.on('completed', (job) => console.log(`완료: ${job.id}`));
worker.on('failed', (job, err) => console.error(`실패: ${job?.id}`, err));

// 이벤트 모니터링
const events = new QueueEvents('email', { connection });
events.on('progress', ({ jobId, data }) => {
  console.log(`진행: ${jobId} - ${data}%`);
});