[spring webflux] 이벤트 버스 활용하기

Intro

회사에서 진행하고 있던 프로젝트의 기능 중 하나가 유저의 추천인 코드를 업데이트 하는 것이었고 이 로직을 구현하면서 eventBus를 활용해보았습니다. 이 글은 eventBus를 만들고 활용한 예시에 대한 글이고 느낀점을 담았습니다.

EventBus란?

https://kwonnam.pe.kr/wiki/java/guava/eventbus
이글에 내가 작성하고 싶은 내용이 정확히 적혀있어서 가져왔다.

이글에는 아래와 같이 적혀있다.

개발을 하다보면 본질적인 비즈니스 로직(예: 사용자 가입. 사용자정보 검증 및 DB 저장)과 비본질적인 비즈니스에 대한 후처리 로직(예: 가입 축하 Email발송, 통계 서비스에 가입자 통지 등)이 강하게 결합(Tight Coupling)하는 경우가 발생한다.

실제 비즈니스 로직과 그 후처리 로직을 완전히 분리하여 비즈니스 로직을 간결하게 유지하며 코드 유지보수성을 높인다.

비즈니스 로직과 후처리 로직이 섞여 있으면 후처리 로직에서 발생하는 예외 등으로 인해 비즈니스 로직이 영향을 받을 수 있는데 Event Bus를 사용하면 서로간의 영향을 분리할 수 있다.

비즈니스에 직접적인 관련이 없는 코드를 추가할 일이 있을 때마다 비즈니스 코드 자체를 수정하는 것은 개방 폐쇄 원칙에 어긋나고 유지보수성도 떨어진다. 비본질적 코드는 본질적코드에서 알 필요가 없게 분리하는 것이 좋다.


비즈니스 코드의 마지막에서 Event를 발생시키고, 비본질적인 후처리 로직들을 Event Listener로 만들어 처리하도록 한다.

너무 길다.
내가 이해한바대로 간단히 말하자면 아래와 같다.

위에서 언급했던 대로 내가 만약 추천인 코드를 업데이트하는 코드를 작성한다면 함수에 추천인 코드를 업데이트 하는 로직만 넣고 다른 이벤트들에 대해서는 느슨하게 결합하는 것이다.

기존 코드

이해를 편하게 하기 위해 코드와 함께 작성해보겠다.

    suspend fun event1testEventBus() {
        println("이벤트1")
        println("이벤트1-1")
        println("이벤트1-2")
    }

위의 함수는 event1을 위한 함수이다.
event1은 특이하게도 event1이 발생하면 event1-1과 event1-2가 함께 발생한다.

이렇게되면 사실 event1에 대한 함수임에도 불구하고 event1-1과 event1-2에 대한 코드가 추가되어야한다.

eventBus는 이런게 싫어서 나왔다고 생각한다.

코드

google의 guava의 코드를 참고해서 만들었다.
guava를 쓰고싶었지만...코루틴 지원을 안한다ㅜ
suspend 함수 지원을 안해준다

// 타입을 지정하기 위해 generic 사용
class CoroutineEventBus<T> {

    private val subscribers = mutableListOf<EventSubscriber<T>>()
	
    // subscriber 등록
    fun register(subscriber: EventSubscriber<T>) {
        subscribers.add(subscriber)
    }
	
    // subscriber 등록 해제
    fun unregister(subscriber: EventSubscriber<T>) {
        subscribers.removeIf { it == subscriber }
    }
	
    // 이벤트 발행
    suspend fun post(event: T) = supervisorScope { // 상위 코루틴으로 예외 전파 방지
        subscribers.forEach {
            launch(Dispatchers.IO) {
                it.onEvent(event)
            }
        }
    }
}

subscriber의 구현체들은 아래 EventSubscriber 인터페이스를 구현한다.

interface EventSubscriber<out T> {
    suspend fun onEvent(topic: @UnsafeVariance T)
}

아래는 EventSubscriber인터페이스의 구현예시이다.
나는 테스트 삼아서 2개정도 만들어보았다.

@Component
class FirstSubscriber : EventSubscriber<String> {

    override suspend fun onEvent(topic: String) {
        println("$topic FirstSubscriber.onEvent()")
    }

}

@Component
class SecondSubscriber : EventSubscriber<String> {

    override suspend fun onEvent(topic: String) {
        println("$topic SecondSubscriber.onEvent()")
    }

}

그리고 bean으로 등록해서 사용하기 위해 아래와 같은 코드도 추가하였다.

@Configuration
class EventBusConfig(
    private val firstSubscriber: FirstSubscriber,
    private val secondSubscriber: SecondSubscriber,
) {

    @Bean
    fun printTestEventBus(): CoroutineEventBus<String> {
        return CoroutineEventBus<String>().apply {
            this.register(firstSubscriber)
            this.register(secondSubscriber)
        }
    }

}

최종

최종적으로 서비스에서는 아래와 같이 코드를 작성하면 된다.

@Service
class TestService(
    private val printTestEventBus: CoroutineEventBus<String>
) {

    suspend fun event1testEventBus() {
        println("이벤트1")
        printTestEventBus.post("이벤트메세지")
    }

}

그러면 event1에 대해 발생되는 이벤트들을 printTestEventBus.post("이벤트메세지") 이렇게 한줄로 마무리 할 수 있다.

실행 후

코드를 실행하면 아래와 같이 2개의 이벤트도 함께 실행되는 것을 볼 수 있다.

느낀점

직접 구현해보면서 느낀 장점은 우선 느슨한 결합을 할 수 있다는 것이다. 하나의 이벤트에 대해 연쇄적으로 발생되는 이벤트를 같은 코드에 녹일 필요가 없다.

하지만 이는 곳 디버깅의 어려움을 뜻한다.
아직은 하나의 이벤트에 대해 발생되는 이벤트들이 1,2개지만 만약에 더 많아진다면 어디서 에러가 발생하는지 찾아내기 어려울 것이다.

계속 사용해보고 활용해보면 어떤 장단점이 있는지 더 느껴지겠지만 지금 드는 생각은 이벤트 버스에 대한 subscriber들은 데이터 유실이 크게 영향이 없는? 그런 로직들에 대해 사용하면 좋을 듯하다. (ex 조회수 등)

그리고 내가 만든 이벤트 버스나 guava 이벤트 버스나 애플리케이션 이벤스 버스 전용인데 추후에 분산 시스템간에 메시지를 전달하는 아키텍처로 바뀌게 되면 카프카 같은 인프라를 사용해서 메시지 버스를 사용해야하는데 이때는 위에서 작성한 이벤트 버스 구조의
post, register, unregister 내부에 카프카 연동만 하면 될듯하다.