Project : RainMind 개발일지 - 4 [이벤트 처리 이슈]

Spring에서 redis enqueue / dequeue를 위해 발행했던 event, FastAPI에서는 해당 기능이 존재하지 않아 직접 event_publisher.py를 이용해야 했었다.

Spring에서의 Event 처리, 그리고 FastAPI에서 이를 흉내내어 구현했던 방법까지 정리해두면 좋을 것 같아 정리한다.

Spring - Event

Spring에서의 Event란 본디 Bean 간의 소식 주고받음을 위해 설계되었다. 자극과 반응이 있으니 통신이라고 불러야 하나… 어쨌든 이러한 기능이 많이 확장되어 지금은 트랜잭션 단계에서 event 발행 또한 가능해지게 되었다.

기본적으로 이벤트를 발행(publisher) 하고, 이벤트를 수신하여 소비(구독이라고 하기도 한다)하는 구조를 사용한다.

Spring에서 본디 Bean을 염두에 두고 만들었다 보니 애플리케이션이 시작될때 (ApplicationContext가 띄워질때), 또는 중지될때 이를 이벤트로 발행하였다. 구체적으로 Context Start / Stop / Close Event가 있고, HTTP 요청 관련한 Event들도 있고… 여러 가지 존재하지만 지금 중요한 것은 “사용자 정의 이벤트를 발행했다”는 것이다.

ScheduleService.kt 코드에서, ApplicationEventPublisher를 주입받아 publishEvent() 함수를 사용했다. 또한 NotifyQueueService.kt 에서 내가 발행한 사용자 정의 이벤트들을 소비하는 Event Listener들을 @TransactionalEventListener를 통해 지정했다.

이러한 이벤트들을 트랜잭션 경계에서 발생한 이벤트 라고 하여 ‘Transaction-Bound Events’라고 하는데, 트랜잭션 상태(commit, rollback)에 따라 동작을 지정할 수 있다.

Spring은 수많은 기능을 모아놓은 프레임워크이며 우리는 그 중에서 Web 관련 기능들만 쏙 빼먹고 있는 것이기 때문에 원체 다양한 기능이 많다.

FastAPI - Event

FastAPI에서는 외부 라이브러리 기능을 가져오는 등의 노력을 하지 않는 이상 Spring처럼 어노테이션 하나로 트랜잭션 상태와 연동시켜 이벤트를 처리하는 기능은 없는 걸로 보인다.

FastAPI는 웹 프레임워크라 그런지 웹 요청 처리에 집중하는 듯한 그런 느낌이다.

Spring의 이벤트 발행 기능을 흉내내기 위해서는, 크게 nested function call, 직접 구현, 외부 라이브러리 사용 이렇게 방법을 생각할 수 있을 것 같다.

• Nested Function Call: 말 그대로 함수 안에서 또 함수를 부른다. 예를 들어, 현재 내 코드에서는 DB 작업 이후 Redis를 조작하여 알림을 삽입하므로, DB 작업하는 함수 로직 맨 아래에 알림 삽입 로직을 담당하는 함수를 내부에서 호출할 수 있을 것이다. 그러나 이 방법의 경우, Redis와 DB에 장애가 발생하면 데이터 정합성을 보장할 방법이 없기 때문에 사용하지 않는다.

• 직접 구현: Event_publisher를 사용하여, 죽지 않는 (while문 무한루프) worker를 구현했다. Outbox pattern을 가져왔으므로, DB에서 PENDING Outbox를 조회해야 하므로 DB 세션 열어주고, 1초마다 DB를 조회해서 Redis에 삽입한다. Outbox table은 트랜잭션으로 보호받으므로 항상 신뢰할 수 있기 때문이다.

• 외부 라이브러리 사용: fastapi-events가 있긴 하던데, 어차피 event를 발행해도 장애로 인해 Redis에 삽입 안될 경우를 대비하여 일정 시간마다 Outbox 조회를 담당하는 publisher를 구현해야 하므로 직접 구현 방식을 선택했다.

FastAPI에서는 직접적으로 Event handle을 지원하는 기능이 없어, Spring의 AFTER_COMMIT 방식을 그대로 흉내내어 Outbox와 Publisher로 역할을 분해했다.

FastAPI는 구현된 어노테이션이 없어 불편했지만, 프레임워크 설계 의도에서 차이가 있었고, 또 데이터가 언제 어떤 상태에서 어떻게 진입하여 처리되는지 명확하게 표현할 수 있었다.