Redis Stream의 구조적 특징(Message Queue)

1. Redis Stream 개요

소개

Redis 5.0에서 주요한 기능으로 등장했다. ( 18년 10월)

Message Queue (MQ)로 사용할 수 있는 스택 중 하나이다. 

 

https://cloud.google.com/solutions/event-driven-architecture-pubsub?hl=ko

 

Redis Stream MQ는 이런 특징을 가진다. (Rabbit MQ, Active MQ도 마찬가지)

- 중복 없이 동시처리 가능

- 재처리 가능

 

이 특징들은 Redis Stream 구조에서 기인한다.

1. 중복 없이 동시처리 가능한 이유는 consumer group이 single stream을 두고 메시지를 각 consumer가 하나씩 빼와서 처리하기 때문이다.

2. 재처리가 가능한 이유는 메시지마다 id를 두고 관리하며, ACK 명령을 통해 메시지 처리완료를 지원하고, 만약 ACK를 받지 못한 메시지는 PENDING 처리하기 때문이다. 이 두가지 이야기는 이 글에서 조금 더 자세히 설명할 것이다.

 

구조상 유의점

  하나의 stream을 여러 consumer가 병렬 처리하는 구조로 동작한다. consumer group당 single stream이 존재한다. 전통적인 대기열과 같은 구조이다. 이 구조의 특징은 처리 순서를 보장할 수 없다는 것이다. 왜냐하면 각 consumer가 스트림에서 가져온 데이터를 처리완료하는 시점은 충분히 다를 수 있기 때문이다. 다시말해 stream에 순차적으로 쌓인 데이터가 병렬처리한다는 개념으로 보면 된다.

 

The Redis consumer group archetecture (https://mattwestcott.org/blog/redis-streams-vs-kafka)

 

 

In-Memory 특징상 유의점

stream의 '메시지'는 크게 DELIVERED, PENDING, ACK 상태를 가지고 있다.

 

- DELIVERED: 메시지가 stream에 전달된 상태

- PENDING: 메시지가 consumer에게 전달된 상태 (consumer가 stream에서 읽은 상태)

- ACK: 메시지가 consumer에 의해 처리완료된 상태 (consumer가 stream에 ACK를 보내야만 함)

 

  앞서 말한 내용을 기억해보자. stream은 pub/sub과는 달리 수신여부를 저장할 수 있다고 했다. 따라서 메시지가 유실되지 않도록 처리완료되지 않은 메시지를 보관하는 리스트 자료구조인 PEL을 가지고 있다. (PEL: Pending Entry List)

만약 XACK를 받지 못해서 PENDING으로 남은 메시지가 점점 쌓이면 메모리 이슈를 발생시킨다. 따라서 stream을 사용할 때는 pending 상태의 메시지가 쌓이지 않도록 신경써야 한다. RedisTemplate을 사용할 경우 redisTemplate.opsForStream().ackowledge(consumerGroupName, message); 같은 코드를 사용해야 한다는 것이다.

흔히 schedule을 등록하여 pending 메시지를 처리해준다.

 

 

 

Pub/Sub과의 비교

Redis stream과 pub/sub 모두 메시지 통신을 처리하는 이벤트 브로커로서 사용될 수 있다. stream과 pub/sub 모두 Event Driven Architecture, EDA 구조다. 이벤트 메시지를 기다리고 있는 구조이다. 실시간 데이터(이벤트) 처리, 비동기 시스템에 고려할 수 있다. 이벤트 브로커의 구조가 왜 stream이든 pub/sub이든 특별하게 구성되어야 할 필요가 있었을까?  서버-클라이언트가 직접 통신할 때의 결합도를 낮춰서, 시스템간의 영향을 줄여야 했기 때문이다. 예를 들어, 이벤트 브로커를 두면 subscriber가 늘어나더라도 publisher는 전혀 영향을 받지 않는다.

 

 

 하지만 메시지 중복여부와 수신확인 기능을 고려하여 적절한 스택을 선택해야 할 것이다.

 

Pub/Sub의 경우 publisher가 메시지를 발행했을 때 sub이 존재하지 않거나 다른 오류가 나면 수신여부와 상관없이 메시지가 휘발된다. 수신 여부를 확인하고 싶다면 Pub/Sub은 부적절할 수 있다. 또한 데이터가 중복될 필요가 없는 경우도 pub/sub을 피해야 한다. 동일 channel의 sub들에게 동일 메시지를 브로드캐스트하는 특징이 있기 때문이다. (아래 구조 참고)

https://cloud.google.com/solutions/event-driven-architecture-pubsub?hl=ko

 

 

 

Stream의 경우 수신 여부를 알 수 있다. 마지막으로 수신한 record id를 저장하여 메시지 수신을 관리할 수 있다. 예를 들어 XACK 명령어를 사용해 메시지 처리 여부를 확인할 수 있고, Pending Entries List가 존재하여 일정시간 처리되지 못한 메시지를 재처리할 수도 있다. 또한 stream 구조상 consumer는 single stream을 공유하여 하나씩 가져간다. 메시지를 중복 처리하지 않는 상황일 경우 적절하다.

 

 

 

 

---

2. Consumer 구현

RedisStreamConsumer는 implements StreamListener 해서 onMessage()를 구현해야 한다.

 

onMessage

여기에는 consumer에게 전달된 메시지들을 처리하는 코드를 작성하게 된다.

 

메시지 처리는 비즈니스에 적절하게 하면 될 테지만, 앞서 언급한 것처럼 consumer가 메시지를 처리완료했다는 ACK 명령이 포함되어야 하고, 이를 통해 consumer에게 간 메시지가 pending 상태에 있지 않도록 해야 한다. (메모리 이슈!)

 

Interface StreamListener<K,V extends Record<K,?>>

K - Stream key and Stream field type.

V - Stream value type.

참고: 파라미터 message 타입은 ObjectRecord, MapRecord 등이 종종 활용됨

 

 

 

.

.

 

Ref.

https://techblog.lycorp.co.jp/ko/building-a-messaging-queuing-system-with-redis-streams

https://dev.gmarket.com/113

https://mattwestcott.org/blog/redis-streams-vs-kafka

https://dlwnsdud205.tistory.com/369