Redis Stream 도입 사례

Data Product 팀은 montelena receiver와 post Processor로 수집한 view, event, impression 데이터를 적재·가공해 지표와 분석에 활용해 왔다.

유니크 active user를 식별하기 위해 session_id 히스토리를 저장했는데, **Big Smile Day(BDS)**나 UTC Push처럼 유입이 급증하는 시점에는 데이터 처리가 지연되는 문제가 생겼다. 부하를 만드는 로직을 분리해 별도 application으로 처리할 필요가 있었고, 기존 Redis를 그대로 활용할 수 있는 Redis Stream이 대안으로 선택됐다.

선택 이유는 분명했다. Kafka나 MQ를 새로 도입하면 개발 공수와 운영 리소스가 커지고, 행사 일정상 개발 기간도 짧았다. 이미 session_id 저장소로 Redis를 쓰고 있었기 때문에, 기존 로직을 크게 건드리지 않으면서 빠르게 히스토리를 적재할 수 있는 구조가 필요했고, consumer group을 통해 여러 pod 환경에서도 중복 없이 처리할 수 있는 Redis Stream이 맞았다.

Redis Pub/Sub과 비교하면 차이가 크다.

• Pub/Sub은 subscriber가 없거나 애플리케이션에 문제가 생기면 메시지가 휘발된다.
• 여러 subscriber가 있으면 동일 메시지가 모두에게 전달돼 중복 처리 위험이 있다.
• Redis Stream은 XADD, XREADGROUP, XACK, XPENDING, XCLAIM 같은 명령으로 마지막 수신 record id를 기준으로 메시지를 관리한다.
• consumer group을 지원해 하나의 stream을 여러 consumer가 병렬로 나눠 처리할 수 있고, ack와 pending 재처리도 가능하다.

구현은 단순한 구조였다. Publisher는 post Processor에서 가공한 session 객체를 ObjectRecord로 감싸 streamKey에 발행하고, opsForStream().add()가 반환하는 recordId를 고유 메시지 id로 사용했다. value 타입은 MapRecord도 지원하지만, 이 사례에서는 JSON 문자열 기반 ObjectRecord를 사용했다.

Consumer는 spring-boot-starter-data-redis를 추가한 뒤 StreamListener를 구현했다. afterPropertiesSet()에서 consumer group을 만들고 StreamMessageListenerContainer를 설정한 다음 ReadOffset.lastConsumed()로 구독을 시작했으며, 메시지를 처리한 뒤에는 acknowledge()로 처리 완료를 알렸다. 예외가 발생하면 recordId 기준으로 삭제 처리하는 방식도 함께 썼다.

운영에서 가장 중요하게 본 점도 있다. Redis Stream은 partition 개념이 없어 하나의 stream을 여러 consumer가 병렬 처리하는 구조이며, Redis cluster에서 메시지를 고르게 분산하려면 여러 stream을 노드별로 나누는 추가 설계가 필요하다. 그만큼 produce 순서가 엄격하게 보장되지 않는다는 점도 감안해야 했다.

또한 in-memory 저장소인 만큼 memory 관리가 핵심이다. XACK를 받지 못한 pending 메시지를 방치하면 Redis cluster 메모리가 계속 늘어나므로, 1분마다 스케줄러를 돌려 pending summary를 확인하고 재처리하는 로직을 넣어 memory full 이슈를 막았다.

적용 결과, Redis Stream 도입 후 UTC Push 발송 시 지연되던 트래픽 처리가 해소됐고, BDS 행사 기간에도 안정적으로 서비스를 운영할 수 있었다. 분산 환경에서 빠르게 실시간 데이터 처리를 분리해야 할 때 Redis Stream이 유효한 선택지라는 점을 보여준다.