[Redis] Stream

 

 

 

실시간 데이터 처리, 메세지 큐, 이벤트 스트리밍을 구현할 때 사용된다. 

Redis Core 내부에 포함되어있어 RediSearch 처럼 별도의 모듈로 설치되지 않아도 사용할 수 있다.

 

Message Producer / Redis / Worker 로 구성된 시스템에서 서는 Redis가 메세지 큐 역할을 수행해 생산자가 생산한 메세지를 큐에 저장하고 소비자가 큐에서 메세지를 가져가 처리할 수 있도록 도와준다. 

 

 

스트림에 데이터를 추가할 때는 XADD 명령어를 사용한다.

 

 

 

스트림 명과 고유 ID를 지정한 후 key - value 쌍을 입력받는다.

고유 ID를 * 로 지정할 경우 Redis는 내부에서 Unix Timestamp 기반 ID를 생성하는데, ID가 겹치는 경우 -0, -1 처럼 옵션이 함께 붙어 구분된다.

 

 

저장된 메세지를 읽을 때는 XREAD 명령어를 사용하는데, 저장된 메세지를 읽더라도 스트림에서 데이터를 삭제하지 않으니 여러 소비자가 스트림에 저장된 데이터를 동시에 읽을 수 있다.

 

 

 

 

BLOCK 옵션을 지정하면 스트림에서 데이터를 비동기 방식으로 읽어올 수 있다.

대상 스트림에 목표 데이터가 있으면 데이터를 반환하지만, 데이터가 없는 경우 BLOCK 옵션으로 지정한 시간만큼 대기한다.

 

대기 중 조건에 맞는 데이터가 스트림에 입력되면 해당 데이터를 반환한다.

이벤트가 발생할 때 마다 소비자가 해당 이벤트를 읽고 처리할 수 있어 채팅 시스템 등 실시간 시스템 구축에 사용된다.

 

XREAD로 데이터를 읽을 때 스트림에서 데이터가 삭제되지 않아 여러 사람이 한 번에 같은 스트림의 데이터에 접근할 수 있지만, 메세지가 중복 처리될 수 있어 스트림의 데이터를 병렬로 처리할 때 주의해야 한다.

 

여러 메세지 브로커처럼 Redis도 Consumer Group을 정의하고 여러 소비자가 협력해 하나의 스트림에서 데이터를 병렬로 처리할 수 있는 기능을 제공한다.

 

Consumer Group 내에 속한 소비자들은 각 소비자들끼리 동일한 메세지를 처리하지 않도록 관리되고, 처리 도중 오류가 발생한 메세지는 다른 소비자가 다시 처리할 수 있도록 자동화되어있다.

 

 

 

 

XGROUP CREATE 명령어로 Consumer Group을 설정할 수 있다.

대상 스트림과 그룹명을 지정하고 0과 $ 옵션으로 모든 데이터를 읽을지, 현재 시점부터의 데이터만 읽을지를 설정한다.

 

그룹에 소비자를 추가할 때는 XGROUP CREATECONSUMER 명령어를 사용한다.

 

일반적으로 스트림의 소비자는 그룹에서 소비자가 XREADGROUP 명령어로 데이터를 읽을 때 자동으로 생성되니 CREATECONSUMER 명령어를 사용하지 않아도 되지만, 명시적으로 소비자를 추가하거나 미리 소비자를 추가해 두는 등 전처리 작업을 수행할 때 사용하자.

 

소비자가 메세지를 처리했으면 해당 메세지가 다른 소비자에게 다시 할당됨을 방지하기 위해 Consumer Group에게 해당 메세지를 처리했다고 알려 줘야 한다.

XACK 명령어를 사용해 메세지를 처리했음을 알려준다.

 

처리 중 오류가 발생한 메세지나 아직 처리되지 않은 메세지는 XPENDING 명령어로 일괄 조회할 수 있다.

 

 

Kafka, RabbitMQ 등 오픈소스 진영에는 유명한 메세지 브로커가 있는데..

설정이 복잡하다.

 

이전에 Redis를 캐시 서버로 사용하고 있었다면 추가적인 메세지 브로커 인프라를 도입하지 않고 그대로 스트림을 사용할 수 있으니, 간단하게 사용할 경량 메세지 큐가 필요한 경우 스트림을 사용하자.