(Celery) 트랜잭션이 커밋된 이후에 Celery Task 전송하기

트랜잭션 안에서 Celery Task가 호출되는 경우, 트랜잭션이 커밋되기 전에 Task가 실행될 수 있다.

이 때, Task가 커밋되지 않은 데이터를 참조하는 경우, 오류가 발생할 수 있다. 

위와 같은 상황의 예시와 이를 회피하는 방법에 대해서 정리해보고자 한다.

 

문제상황

아래의 코드는 회원가입을 처리하는 가상의 코드이다.

from django.db import transaction

from coupon.util import create_welcome_coupon
from mileage.util import create_welcome_mileage

@transaction.atomic()
def create_user(user_data):
    user = User.objects.create(**user_data)
    send_email_welcome_join.delay(user.pk)
    create_welcome_coupon(user.pk)
    create_welcome_mileage(user.pk)

 

이 때, 트랜잭션이 실행되는 도중에 이메일 발송 태스크가 먼저 실행되는 경우, 이메일 발송 태스크에서는 회원 데이터를 조회할 수 없다. 

 

Celery에서는 회원가입 로직과 다른 DB session에서 데이터를 조회하기 때문이다. 

 

해결방법

django transaction의 on_commit를 활용하면, commit이 성공한 이후에 셀러리 함수가 호출되도록 할 수 있다. 

from django.db import transaction

from coupon.util import create_welcome_coupon
from mileage.util import create_welcome_mileage

@transaction.atomic()
def create_user(user_data):
    user = User.objects.create(**user_data)
    transaction.on_commit(lambda: send_email_welcome_join.delay(user.pk))
    create_welcome_coupon(user.pk)
    create_welcome_mileage(user.pk)

 

Celery Task  함수를 on_commit 함수로 감싸주면 된다.

 

참고

Database transactions | Django documentation | Django

Tasks — Celery 4.3.0 documentation