백엔드/Redis

[Redis] Redis Stream 주요 명령어 알아보기

연유뿌린빙수 2025. 9. 8. 09:16

Redis는 다양한 자료구조를 저장할 수 있으며, 그중에서도 스트림(Stream) 기반 데이터를 지원한다.

이미지 출처 :   https://mattwestcott.org/blog/redis-streams-vs-kafka


Redis를 메시지 브로커로 활용하면 인메모리 기반의 빠른 비동기 메시지 처리가 가능하다. Kafka와 같은 대규모 메시지 브로커에 비해 구조는 단순하지만, Redis만의 장점이 존재한다.

 

Redis의 메시지 브로커로서의 특징은 다른 글에서 정리했으니,
오늘은 Redis Stream 기반 주요 명령어들에 대하여 알아보자.




1. XADD

가장 기본적인 발행이다(Publish)
Stream에 데이터를 발행한다.

XADD key ID field value [field value ...]
  • key : Stream의 이름
    -> 만약 Stream이 존재하지 않으면 이 때 새로 생성이 된다(그래서 구독이 아닌 발행의 측면에서 따로 생성하는 과정은 필요없음)
  • ID : 새 엔트리(Message)의 ID
      • : Redis에서 자동으로 생성해줌
    • 직접 지정도 가능
  • field value: Stream에 저장할 데이터로 필드-값 쌍의 내용들을 저장한다

예시

XADD streamname * ...

이 때 stream이 커지지 않도록 최대 길이 제한이 가능하다(trim)

XADD mystream MAXLEN 1000 * event click page home

-> 이렇게 MAXLEN의 옵션을 설정하면 mystream이라는 Stream에 최대 1000개까지 메시지 발행이 가능하다

그래서 일반적으로 MAXLEN의 경우, 자동으로 오래된 데이터를 삭제하도록 설정하기 위하여 추가해준다.



2. XREADGROUP

Redis는 Consumer group을 기반으로 구독을 한다.
그 그룹은 동일 스트림 키에 대하여 모든 내용을 공유한다.


XREADGROUP은 이를 기반으로 그룹 단위로 읽을 때의 명령어이다. ``` XREADGROUP GROUP group_name consumer_name [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...] ```
  • GROUP group_name consumer_name
    • group_name: 소비자 그룹 이름
    • consumer_name: 이 그룹 안의 특정 소비자(클라이언트) 이름
  • COUNT count : 한 번에 읽어올 엔트리 개수 제한
  • BLOCK milliseconds : 새 메시지가 없으면 기다릴 시간 (ms 단위, 0 = 무한 대기)
  • STREAMS key [key ...] id [id ...] : 읽을 스트림 키와 시작할 ID, 일반적으로 두 가지 모드로 사용됨
    • > : 그룹이 아직 소비하지 않은 새 메시지만 읽음
    • 특정 ID (0 또는 다른 값): 해당 소비자에게 할당된 미처리 메시지(pending)를 다시 읽음


3. XGROUP

소비자 그룹 만들 때 사용하는 명령어 XGROUP

XGROUP CREATE mystream mygroup $ MKSTREAM

이 명령어를 통하여 어플리케이션을 시작할 때 자동으로 구독할 수 있도록 명령어 설정이 가능하다.

  • $ : 지금 이후로 들어오는 새 메시지부터 소비
  • '0' : 미저치 메시지 처리(pending)


4. XTRIM / DEL

Redis Consumer가 현재 처리하지 못하고 있는 작업이 있다면 다음과 같이 뜰 것이다.
나의 경우 발행 이미지 자체가 잘못되어서 문제가 되었다.


2개의 메시지가 현재 작업에 남아있는데 계속 처리를 못해서 문제가 됐다.
그래서 새로 쿼리로 남아있는 pending 이미지들을 전부 삭제했다.

XTRIM key MAXLEN [~] count
XTRIM key MINID [~] id
  • MAXLEN count
    - 스트림 길이가 count를 초과하면 오래된 항목 삭제
    ~를 붙이면 근사치 모드 (성능 ↑, 조금 더 많이 남을 수 있음)
    TRIM mystream MAXLEN 1000
    -> mystream에 최근 1000개만 남김 (오래된 것을 삭제)
  • MINID id
    - 특정 ID 이전의 모든 항목 삭제
    XTRIM mystream MINID 1694174585690-0
    -> 해당 ID보다 작은 모든 엔트리 삭제

기본적인 명령어는 위와 같다. 적용해보면 다음과 같다.


127.0.0.1:6379> XTRIM <STREAM KEY> MAXLEN 0
(integer) 2

가령 예를 들어서 현재 내가 구독중인 스트림 키의 이름이 "stream.key.1"이라고 가정해보자. 그러면 명령어로

127.0.0.1:6379> XTRIM stream.key.1 MAXLEN 0
(integer) 2

이렇게 입력해주면 다음 2개의 메시지가 삭제됨을 알 수 있다.

만약에 처음부터 없애버리고 싶다면,

DEL stream.key.1

이렇게 스트림 키 자체를 입력해주면 사라진다

 

5. XACK

Consumer Group과 함께 사용하는 명령어이다.

XREADGROUP으로 "읽은 메시지"는 PENDING LIST에 기록되는데 XACK는 해당 메시지를 처리완료 상태라고 Redis Stream에 알리는 역할이다.


XACK key groupname ID [ID ...]
  • key → 스트림 이름
  • groupname → 소비자 그룹 이름
  • ID [ID ...] → 완료 처리할 메시지 ID(하나 이상 가능)

ex) 실제 사용

XACK mystream mygroup 1694174585690-0

-> mystream 스트림에서 mygroup 그룹이 1694174585690-0 메시지를 처리 완료했다고 표시



XCLAIM

XCLAIM은 Consumer Group 환경에서 특정 메시지의 소유권을 다른 소비자에게 넘겨주는 명령어이다.
즉, 메시지를 재처리해주는 명령어이다.


다른 소비자가 그 메시지를 가져와서 처리할 수 있게 해준다.

XCLAIM key group consumer min-idle-time id [id ...]
       [IDLE ms] [TIME ms-unix-time] [RETRYCOUNT count]
       [FORCE] [JUSTID]
  • key → 스트림 이름
  • group → 그룹 이름
  • consumer → 새로 메시지를 가져갈 소비자 이름
  • min-idle-time → 메시지가 최소한 이 시간(ms) 동안 pending 상태였을 때만 소유권 이전 가능
  • id [id ...] → 재할당할 메시지 ID 목록

그래서 특정 작업을 완료하지 못하였을 때 XCLAIM을 통해서 재시도를 하는 등의 횟수에 대하여 추가해주면 좋다.

 
 

AUTOXCLAIM

이러한 XCLAIM을 기반으로 생성된 것이 AUTOXCLAIM이다.

XAUTOCLAIM은 pending 상태에서 오래 방치된 메시지들을 자동으로 다른 소비자에게 재할당하는 명령어다.
즉, XCLAIM과 달리 직접 메시지 ID를 지정할 필요 없이, idle 시간이 오래된 메시지를 한꺼번에 가져올 수 있다는 점에서 'AUTO'가 붙은 것이다.

XAUTOCLAIM key group consumer min-idle-time start [COUNT count] [JUSTID]

기존의 XCLAIM과 틀은 비슷하지만 차이가 존재한다.

  • start → 검색 시작할 ID, 일반적으로 "0"을 넣으면 pending 전체를 대상으로 검색
  • COUNT count → 최대 몇 개까지 가져올지 제한 (기본값 100)
  • JUSTID → 메시지 ID만 반환 (데이터는 안 가져옴)
 
 
차이점 정리
 

메시지 지정 직접 ID 지정 오래된 메시지를 자동 검색
대상 메시지 특정 메시지 조건에 맞는 여러 메시지
페이징 없음 있음 (next-start)
사용 편의성 수동 관리 필요 자동화, 운영 편리

 

이러한 펜딩에 대한 관리가 있어야 인메모리 기반의 Redis에서 메모리의 과부하를 줄일 수 있다.
인메모리에 계속 pending 내용이 쌓이면 성능에 문제가 생길 수 있으므로 오래된 데이터는 자동으로 삭제하는 등의 관리와, 처리하지 못한 작업에 대한 재처리 작업 등이 필요하다.