Django Signals vs Celery Tasks 2026: 적절한 사용법과 면접 대비 가이드

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import UserProfile

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    """사용자 생성 시 프로필 자동 생성"""
    if created:
        UserProfile.objects.create(user=instance)

@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
    """사용자 저장 시 프로필도 함께 저장"""
    if hasattr(instance, 'profile'):
        instance.profile.save()

from django.dispatch import Signal

# 커스텀 시그널 정의
order_completed = Signal()
payment_processed = Signal()

# 시그널 발송
from .signals import order_completed

def complete_order(order):
    # 주문 완료 처리
    order.status = 'completed'
    order.save()
    
    # 시그널 발송
    order_completed.send(
        sender=order.__class__,
        order=order,
        user=order.user
    )

from .signals import order_completed
from .services import NotificationService, InventoryService

@receiver(order_completed)
def send_order_confirmation(sender, order, user, **kwargs):
    """주문 완료 알림 발송"""
    NotificationService.send_email(
        to=user.email,
        template='order_confirmation',
        context={'order': order}
    )

@receiver(order_completed)
def update_inventory(sender, order, **kwargs):
    """재고 업데이트"""
    for item in order.items.all():
        InventoryService.decrease_stock(
            product_id=item.product_id,
            quantity=item.quantity
        )

# celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Seoul'

# tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .services import ReportGenerator, ImageProcessor

@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id):
    """환영 이메일 비동기 발송"""
    try:
        from django.contrib.auth.models import User
        user = User.objects.get(id=user_id)
        send_mail(
            subject='환영합니다!',
            message=f'{user.username}님, 가입을 축하드립니다.',
            from_email='noreply@example.com',
            recipient_list=[user.email],
        )
    except User.DoesNotExist:
        pass
    except Exception as exc:
        self.retry(exc=exc, countdown=60)

@shared_task
def generate_monthly_report(month, year):
    """월간 리포트 생성"""
    report = ReportGenerator.create_monthly_report(month, year)
    return report.id

@shared_task(queue='image_processing')
def process_uploaded_image(image_id):
    """업로드된 이미지 처리"""
    ImageProcessor.resize_and_optimize(image_id)

# views.py
from .tasks import send_welcome_email, generate_monthly_report

def register_user(request):
    user = User.objects.create_user(**form.cleaned_data)
    # 태스크를 큐에 추가 (비동기 실행)
    send_welcome_email.delay(user.id)
    return redirect('registration_complete')

def request_report(request):
    # 태스크 스케줄링 (지연 실행)
    result = generate_monthly_report.apply_async(
        args=[request.POST['month'], request.POST['year']],
        countdown=300  # 5분 후 실행
    )
    return JsonResponse({'task_id': result.id})

# Signals: 동기 실행 (요청 내에서 완료)
@receiver(post_save, sender=Order)
def update_order_cache(sender, instance, **kwargs):
    """캐시 업데이트는 즉시 반영이 필요"""
    cache.set(f'order_{instance.id}', instance, timeout=3600)

# Celery: 비동기 실행 (요청 외부에서 처리)
@shared_task
def send_order_notification(order_id):
    """알림 발송은 지연되어도 무방"""
    order = Order.objects.get(id=order_id)
    NotificationService.notify_all_stakeholders(order)

# Signals에서 피해야 할 패턴
@receiver(post_save, sender=Product)
def bad_signal_handler(sender, instance, **kwargs):
    # 무거운 처리를 시그널에서 수행하면 응답 지연 발생
    generate_product_thumbnail(instance)  # 5초 소요
    sync_to_external_api(instance)  # 외부 API 호출
    send_notification_to_subscribers(instance)  # 대량 메일 발송

# 올바른 접근: Celery에 위임
@receiver(post_save, sender=Product)
def product_saved_handler(sender, instance, **kwargs):
    # 가벼운 처리만 시그널에서 실행
    cache.delete(f'product_{instance.id}')
    # 무거운 처리는 Celery에 위임
    process_product_async.delay(instance.id)

@shared_task
def process_product_async(product_id):
    product = Product.objects.get(id=product_id)
    generate_product_thumbnail(product)
    sync_to_external_api(product)
    send_notification_to_subscribers(product)

from django.db import transaction

# 문제: 시그널이 트랜잭션 완료 전에 실행됨
@receiver(post_save, sender=Order)
def notify_on_order_save(sender, instance, **kwargs):
    # 트랜잭션이 롤백되어도 알림이 발송될 가능성
    send_notification.delay(instance.id)

# 해결책: transaction.on_commit 사용
@receiver(post_save, sender=Order)
def notify_on_order_save(sender, instance, created, **kwargs):
    if created:
        transaction.on_commit(
            lambda: send_notification.delay(instance.id)
        )

# 1. 모델 간 자동 동기화
@receiver(post_save, sender=User)
def sync_user_data(sender, instance, **kwargs):
    UserSearchIndex.update_or_create(user=instance)

# 2. 캐시 무효화
@receiver(post_save, sender=Article)
@receiver(post_delete, sender=Article)
def invalidate_article_cache(sender, instance, **kwargs):
    cache.delete_pattern(f'article_{instance.id}*')
    cache.delete('article_list')

# 3. 감사 로그 기록
@receiver(pre_save, sender=SensitiveData)
def log_data_change(sender, instance, **kwargs):
    if instance.pk:
        old = SensitiveData.objects.get(pk=instance.pk)
        AuditLog.objects.create(
            model='SensitiveData',
            object_id=instance.pk,
            changes=get_changes(old, instance)
        )

# 1. 외부 API 연동
@shared_task(bind=True, autoretry_for=(RequestException,), retry_backoff=True)
def sync_to_crm(self, customer_id):
    customer = Customer.objects.get(id=customer_id)
    CRMClient().update_customer(customer.to_crm_format())

# 2. 배치 처리
@shared_task
def process_daily_analytics():
    yesterday = date.today() - timedelta(days=1)
    events = Event.objects.filter(date=yesterday)
    analytics = AnalyticsProcessor.process(events)
    DailyReport.objects.create(date=yesterday, data=analytics)

# 3. 정기 실행 태스크
from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-expired-sessions': {
        'task': 'myapp.tasks.cleanup_sessions',
        'schedule': crontab(hour=3, minute=0),
    },
    'generate-weekly-digest': {
        'task': 'myapp.tasks.send_weekly_digest',
        'schedule': crontab(day_of_week=1, hour=9, minute=0),
    },
}

# 답변에 포함해야 할 핵심 포인트:

# 1. 디버깅의 어려움
# 시그널은 암시적 호출이므로 코드 추적이 어려움

# 2. 순환 참조 위험
@receiver(post_save, sender=ModelA)
def update_model_b(sender, instance, **kwargs):
    instance.model_b.save()  # 이 동작이 ModelA 업데이트를 유발할 가능성

# 3. 테스트 복잡성 증가
from django.db.models.signals import post_save
from django.test import TestCase
from unittest.mock import patch

class TestWithoutSignals(TestCase):
    def test_without_signal(self):
        # 시그널 비활성화 후 테스트
        post_save.disconnect(create_user_profile, sender=User)
        try:
            user = User.objects.create(username='test')
            # 프로필이 생성되지 않음
        finally:
            post_save.connect(create_user_profile, sender=User)

# 멱등성을 보장하는 태스크 구현
from celery import shared_task
from django.db import transaction

@shared_task(bind=True)
def process_payment(self, payment_id):
    with transaction.atomic():
        payment = Payment.objects.select_for_update().get(id=payment_id)
        
        # 이미 처리된 경우 스킵
        if payment.status == 'processed':
            return {'status': 'already_processed'}
        
        # 멱등키로 중복 실행 방지
        idempotency_key = f'payment_{payment_id}'
        if cache.get(idempotency_key):
            return {'status': 'duplicate_request'}
        
        cache.set(idempotency_key, True, timeout=3600)
        
        # 실제 처리 수행
        result = PaymentGateway.charge(payment)
        payment.status = 'processed'
        payment.save()
        
        return {'status': 'success', 'result': result}

# 권장 패턴: 시그널에서 Celery 태스크 호출
from django.db import transaction

@receiver(post_save, sender=Order)
def handle_order_created(sender, instance, created, **kwargs):
    if created:
        # 트랜잭션 커밋 후 태스크 실행
        transaction.on_commit(lambda: [
            send_order_confirmation.delay(instance.id),
            update_inventory.delay(instance.id),
            notify_warehouse.delay(instance.id),
        ])

# 안티패턴: Celery 태스크 내에서 시그널 다량 발생
@shared_task
def bad_batch_update(ids):
    for id in ids:
        # 각 save마다 시그널 발생으로 성능 저하
        obj = MyModel.objects.get(id=id)
        obj.status = 'updated'
        obj.save()  # 시그널 발생

# 개선된 버전: bulk_update로 시그널 회피
@shared_task
def good_batch_update(ids):
    MyModel.objects.filter(id__in=ids).update(status='updated')
    # 필요시 명시적으로 시그널 한 번만 발송
    batch_updated.send(sender=MyModel, ids=ids)

from celery import shared_task
from celery.exceptions import MaxRetriesExceededError

@shared_task(
    bind=True,
    max_retries=5,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True
)
def resilient_external_call(self, data):
    try:
        response = ExternalService.call(data)
        return response
    except RateLimitError as exc:
        # 커스텀 재시도 간격 설정
        raise self.retry(exc=exc, countdown=exc.retry_after)
    except FatalError as exc:
        # 재시도 없이 즉시 실패
        raise exc
    except MaxRetriesExceededError:
        # 최대 재시도 횟수 초과 시 처리
        FailedTask.objects.create(
            task_name='resilient_external_call',
            data=data,
            error='Max retries exceeded'
        )
        raise