[개발생각] 다양한 객체를 하나의 컨슈머로 발행하여 사용하고 싶을때

다른 형태의 객체를 같은 토픽으로 발행하고, 해당 타입에 따라, 별도로 기능을 동작시키고 싶을때,

objectMapper를 잘사용하면, 간단하게 구현을 할수가 있다.

 

아래와 같이 UserEvent객체는 각각,
UserEventTypeEnum.AttendanceDayOnce 일때는

   - UserEvent의 event필드는 AttendanceDayOnceEvent객체가 들어가고,

UserEventTypeEnum.UserRecommend 일때는

  - UserEvent의 event필드는 UserRecommendEvent객체가 들어간다.

 

그리고 UserEvent는 같은 토픽으로 이벤트 발행이 됬을때,

UserEvent를 RawUserEvent형태로 deserialize를 한다.

 

eventType를 판단하여,

각 이벤트에 맞게 event필드를 AttendanceDayOnceEvent, UserRecommendEvent등으로 변환한다.

이 변환할때는 objectMapper의 convertValue함수를 이용하면 편하게 변환을 할수있다.

 

예전에 convertValue를 사용하지않을때는,

objectMapper를 이용해서, serialize를 한번하고, 다시 deserialize를 하는 방법을 사용했다..

(물론, convertValue도 serialize와 deserialize를 내부적으로 수행하지만, 중간 단계없이 parser에서 바로 객체를 맵핑해주기에, 좀더 효율적으로 구동이 되고, 더 빠르다. 그리고 serialize, deserialize가 코드에 노출되는게 너무 꼴보기 싫어...ㅠ)

 

 

그래서 이런 패턴일때는 convertValue함수를 적극 활용하면, 좋다.

 

아래는 코드 샘플!!!

data class RawUserEvent(
    val eventType: UserEventTypeEnum,
    val event:Map<String, Any>,
)

data class UserEvent(
    val eventType: UserEventTypeEnum,
    val event: UserEventSpec
)

interface UserEventSpec {
    val userId:Long
}

data class AttendanceDayOnceEvent(
    override val userId: Long,
    val loginDtm: LocalDateTime,
) : UserEventSpec


data class UserRecommendEvent(
    override val userId: Long,
    val recommenderUserId: Long,
) : UserEventSpec

 

UserEvent객체를 objectMapper.writeValueAsString(userEvent) 해서,

kafka topic에 publish를 했다고 해보자. (AttendanceDayOnceEvent, UserRecommendEvent)

 

 

kafka consumer에서는 아래와 같이 구현을 했을때,

@KafkaListener(topics = ["test"])
fun consumeEvent(rawEvent: String) {
        with(objectMapper) {
            val rawUserEvent = readValue(rawEvent, RawUserEvent::class.java)
            when (rawUserEvent.eventType) {
                UserEventTypeEnum.AttendanceDayOnce -> {
                    val event  = convertValue<AttendanceDayOnceEvent>(rawUserEvent.event)
                    usernManageUseCase.attendanceDayOnce(event)
                }
                UserEventTypeEnum.UserRecommend -> {
                    val event  = convertValue<UserRecommendEvent>(rawUserEvent.event)
                    usernManageUseCase.userRecommend(event)
                }
                else -> {}
            }
        }
    }

 

이렇게 구현하면, 하나의 토픽에서 각각의 서로다른 메세지 객체를 유연하게 사용할수있어서, 기능별로 한곳으로 모아서 처리를 할수있어서, 코드를 간결하게 관리할수있다.

 

위 코드 샘플은, 프로젝트에서 유저가 어떤 행위를 했을때 , 해당 행위에 대한 이벤트를 비동기로 발생시키고,

그 이벤트에 따른 처리를 리스너에서, 종류별로 수행을 하도록 하는 부분이 있는데,

위와 같은 형태로 구현을 한것을 좀바꿔서 샘플을 넣은것이다.

 

이벤트 버스형태로 동작하도록 만든 비즈니스 로직들은 조금만 관리가 허술해져도, 상당히 복잡하고, 스파게티처럼 꼬이게 된다.

이 꼬이게 되는 문제중에 가장 크리티컬한게, 재귀호출이다.

 

리스너에서 수행한 service안에서 다시 해당 이벤트를 발생하고, 해당 이벤트로 다시 리스너가 동작하고, 다시 이벤트가 발생하는 무한 재귀 이벤트 발생 문제를 찾아내기가 힘들다.

 

따라서 미리미리 잘 관리를 해야하는데, 그런 측면에서 위와같이 동작하도록 코드를 유지하는게 우선은 선제적으로 중요하다.

 

그리고 이벤트 발생은 서비스에 넣어두는것보다는, 아래와 같이..

 

usernManageUseCase.attendanceDayOnce(event) {

       publisher(UserEventTypeEnum.AttendanceDayOnce)

}

 

이런식으로 람다블럭을 받아서 실행하도록 변경하고, 해당 람다블록안에서 이벤트를 발행하도록 변경을 하면,

 

Listener등에서는 그냥 usernManageUseCase.attendanceDayOnce(event) 이렇게만 사용하면,

이벤트 발행코드없이 사용할수있기에, 재귀발행을 막을수있다.

 

아니면, 좀더 방어적으로 한다면,

Aspect를 만들어서, 호출된 부모 메소드에 @KafkaListener 어노테이션이 있는경우, publisher발행함수가 무력화되게도 만들수도 있을것 같다. (근데..만들기 귀찮어..)

이게 더 안전하긴하지만, 오버엔지니어링이라고 볼수도 있긴하다...

 

이런 케이스가 많지는 않아서..

그냥 람다방식으로 잘 정리해서 사용하면, 큰문제는 없을듯하다.

 

이벤트방식을 사용할때는 재귀적으로 이벤트가 발생하는 문제는 꼭 고려를 해야한다... 무한 셀프 공격을 하는 코드를 만들수있다...ㅠ

(내일 한번더 점검해봐야지...)