레디스를 메시지 브로커로 사용하기 위한 정리

메시지 큐와 이벤트 스트림

메시지 큐와 이벤트 스트림의 차이

용어

메시지 큐에서는 주로 데이터를 생성하는 쪽을 생산자(producer)로, 데이터를 수신하는 쪽을 소비자(consumer)로 지칭한다.
이벤트 스트림에서는 데이터를 생성하는 쪽을 발행자(publisher)로, 데이터를 조회하는 쪽을 구독자(subscriber)로 지칭한다.

 

방향성

메시지 큐의 생산자는 소비자의 큐로 데이터를 직접 푸시한다.
2개의 서비스에 같은 메시지를 보내야 할 때 메시지 큐를 이용한다면 생산자는 2개의 각각 다른 메시지 큐에 각각 데이터를 푸시해야 한다.

반면 스트림을 이용한다면 생산자는 스트림의 특정 저장소에 하나의 메시지를 보낼 수 있고, 메시지를 읽어가고자 하는 소비자들은 스트림에서 같은 메시지를 풀(pull) 해갈 수 있기 때문에 메시지를 복제해서 저장하지 않아도 된다.

메시지 큐는 일대일(1:1) 상황에서 한 서비스가 다른 서비스에게 동작을 지시할 때 유용하게 사용될 수 있으며, 스트림은 다대다(n:n) 상황에서 유리함을 확인할 수 있다.

 

영속성

메시지 큐에서는 소비자가 데이터를 읽어갈 때, 큐에서 데이터를 삭제한다.
하지만 이벤트 스트림에서 구독자가 읽어간 데이터는 바로 삭제되지 않고, 저장소의 설정에 따라 특정 기간 동안 설정될 수 있다.

 

레디스를 메시지 브로커로 사용하기

레디스의 pub/sub에서 모든 데이터는 한 번 채널 전체에 전파된 뒤 삭제되는 일회성의 특징을 가지며, 메시지가 잘 전달됐는지 등의 정보는 보장하지 않는다.
따라서 완벽하게 메시지가 전달돼야 하는 상황에는 적합하지 않을 수 있지만 fire-and-forget 패턴이 필요한 알림 서비스에서는 유용하게 사용될 수 있다.

 

레디스의 pub/sub

레디스에서 pub/sub은 매우 가볍기 때문에 최소한의 메시지 전달 기능만 제공한다.
발행자는 메시지를 채널로 보낼 수 있을 뿐, 어떤 구독자가 메시지를 읽어가는지, 정상적으로 모든 구독자에게 메시지가 전달 됐는지 확인할 수 없다.
구독자 또한 메시지를 받을 수 있지만 해당 메시지가 언제 어떤 발행자에 의해 생성 됐는지 등의 메타 데이터는 알 수 없다.

한 번 전파된 데이터는 레디스에 저장되지 않으며 단순히 메시지의 통로 역할만 한다.
만약 특정 구독자에 장애가 생겨 메시지를 받지 못했다 하더라도 그 사실을 알 수 없기 때문에 정합성이 중요한 데이터를 전달하기에는 적합하지 않을 수 있다.
이럴 경우 애플리케이션 레벨에서 메시지의 송수신과 관련한 로직을 추가해야 할 수 있다.

 

커멘드

PUBLISH hello world

hello라는 채널에 world라는 메시지를 발행한다.

 

SUBSCRIBE hello
SUBSCRIBE mail-*

hello라는 채널을 구독한다.
mail-로 시작하는 채널을 구독한다.

 

레디스의 list를 메시지 큐로 사용하기

레디스의 자료 구조 중 하나인 list는 큐로 사용하기 적절한 자료 구조다.

 

list의 블로킹 기능

레디스를 이벤트 큐로 사용할 경우 블로킹 기능은 유용하게 사용될 수 있다.

 

폴링

이벤트 기반 구조에서 시스템은 이벤트 루프를 돌며 신규로 처리할 이벤트가 있는지 체크한다.
이벤트 루프는 이벤트 큐에 새 이벤트가 있는지 체크하며, 새로운 이벤트가 없을 경우 정해진 시간 동안 대기한 뒤 다시 이벤트 큐에 데이터가 있는지 확인하는 과정을 반복한다.
이러한 작업을 폴링이라고 하며 폴링 프로세스가 진행되는 동안 애플리케이션과 큐의 리소스가 불필요하게 소모될 수 있다.
또한 이벤트 큐에 이벤트가 들어왔을 수 있지만, 폴링 인터벌 시간 동안 대기한 뒤 다시 확인하는 과정을 거치기 때문에 이벤트를 즉시 처리할 수 없다는 단점이 있다.

 

블로킹을 추가한 커맨드

BRPOP과 BLPOP은 각각 블로킹을 추가한 커맨드다.
해당 커멘드를 사용할 때 리스트에 데이터가 있으면 즉시 반환한다.
만약 데이터가 없다면 리스트에 데이터가 들어올 때까지 기다린 후에 들어온 값을 반환하거나, 클라이언트가 설정한 타임아웃시간만큼 대기한 후에 nil 값을 반환한다.

하나의 리스트에 대해 여러 클라이언트가 동시에 블로킹될 수 있으며, 리스트에 데이터가 입력되면 가장 먼저 요청을 보낸 클라이언트가 데이터를 가져간다.
그렇기에 BRPOP과 BLPOP는 2개의 데이터를 반환한다. 첫 번째는 팝된 리스트의 키 값을 반환하고, 두 번째에 반환된 데이터의 값을 반환한다.

 

Stream

데이터의 저장

메시지의 저장과 식별

레디스에서 하나의 stream 자료가 하나의 stream을 의미한다.
stream 역시 하나의 키에 연결된 자료 구조다.

카프카에서 스트림 데이터는 토픽이라는 개념에 저장된다. 토픽은 각각의 분리된 스트림을 뜻하며, 같은 데이터를 관리하는 하나의 그룹을 의미한다.

 

레디스 stream에 저장된 모든 데이터는 유니크한 ID를 가지며, 이 ID 값이 곧 시간을 의미하기 때문에 시간을 이용해 특정 데이터를 검색할 수 있다.

카프카에서 각 메시지는 0부터 시작해 증가하는 시퀀스 넘버로 식별할 수 있는데, 이때 시퀀스 넘버는 토픽 내의 파티션 안에서만 유니크하게 증가하기 때문에 토픽이 1개 이상의 파티션을 갖는다면 메시지는 하나의 토픽 내에서 유니크하게 식별되지 않는다.

 

스트림 생성과 데이터 입력

레디스에서는 따로 stream을 생성하는 과정은 필요하지 않으며, XADD 커맨드를 이용해 새로운 이름의 stream에 데이터를 저장하면 데이터의 저장과 동시에 stream 자료 구조가 생성된다.

카프카에서는 데이터를 저장하기 위해 토픽을 생성한 뒤, 프로듀서를 이용해 메시지를 보낼 수 있다.

 

XADD Push * userid 1000 ttl 3 body Hey

• Push: stream 자료구조
• {userid: 1000, body: Hey}: 데이터
• *: 자동으로 생성된 ID

만약 자동으로 생성되는 ID가 아니라 서비스에서 기존에 사용하던 ID를 이용해 메시지를 구분하고 싶으면 *이 아니라 직접 ID를 지정하면 된다.

 

데이터 조회

카프카와 레디스 stream에서 데이터를 저장하는 방식은 비교적 비슷하다.
하지만 데이터를 읽어가는 주체, 즉 소비자와 소비자 그룹이 동작하는 방식에서는 분명한 차이가 존재한다.

레디스 stream에서는 데이터를 두 가지 방식으로 읽을 수 있다.
첫 번째는 카프카에서처럼 실시간으로 처리되는 데이터를 리스닝하는 것이고, 두 번째는 ID를 이용해 필요한 데이터를 검색하는 방식이다.

카프카에서 소비자는 특정 토픽을 실시간으로 리스닝하며, 새롭게 토픽에 저장되는 메시지를 전달받을 수 있다. 기본적으로는 리스닝을 시작한 시점부터 토픽에 새로 저장되는 메시지를 반환받도록 동작하며, --from-beginning 옵션을 이용하면 카프카에 저장돼 있는 모든 데이터를 처음부터 읽겠다는 것을 뜻한다. 소비자는 더 이상 토픽에서 읽어올 데이터가 없으면 새로운 이벤트가 토픽에 들어올 때까지 계속 토픽을 리스닝하면서 기다린다.

 

실시간 리스닝

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

XREAD 커맨드를 이용하면 실시간으로 stream에 저장되는 데이터를 읽어올 수 있다.

BLOCK 0은 더 이상 stream에서 가져올 데이터가 없더라도 연결을 끊지 말고 계속 stream을 리스닝하라는 의미다.

ID에 특수 ID인 $를 전달하면 커맨드를 실행한 이후 데이터부터 가져온다.
즉 $는 stream에 저장된 최대 ID를 의미하는 것이라 생각할 수 있다.

 

특정한 데이터 조회

XRANGE key start end [COUNT count]
XREVRANGE key start end [COUNT count]

XRANGE 커맨드를 이용하면 ID를 이용해 원하는 시간대의 데이터를 조회할 수 있다.

 

소비자와 소비자 그룹

같은 데이터를 여러 소비자에게 전달하는 것을 팬아웃이라 한다.
레디스 stream에서도 XREAD 커멘드를 여러 소비자가 수행한다면 팬아웃이 가능하다.

만약 같은 데이터를 여러 소비자가 나눠서 가져가기 위해서는 어떻게 해야 할까?
같은 역할을 하는 여러 개의 소비자를 이용해 메시지를 병렬 처리한다면 서비스의 처리 성능을 높일 수 있다.

레디스에선 stream을 이용해 이벤트 데이터를 처리하는 상황에서 이벤트의 처리 성능을 높이기 위해 여러 소비자를 이용해 한 번에 여러 이벤트를 병렬적으로 처리되도록 구성할 수 있다.
그리고 레디스의 stream은 데이터가 저장될 때마다 고유한 ID를 부여받아 순서대로 저장되기 때문에 소비자에게 데이터가 전달될 때, 그 순서는 항상 보장된다.

반면 카프카에서 유니크 키는 파티션 내에서만 보장되기 때문에 소비자가 여러 파티션에서 토픽을 읽어갈 때에는 데이터의 순서를 보장할 수 없다. 카프카에서 소비자가 토픽에서 데이터를 소비할 때는 파티션의 존재를 알지 못하고, 토픽 내의 전체 파티션에서 데이터를 읽어온다. 이때 여러 파티션 간 데이터의 정렬은 보장되지 않기 때문에 결국 소비자가 데이터를 읽어올 때에는 정렬이 보장되지 않는 데이터를 읽어오게 된다. 따라서 카프카에서 메시지 순서가 보장되도록 데이터를 처리하기 위해서는 소비자 그룹을 사용해야 한다.

 

소비자 그룹

레디스 stream은 메시지가 전달되는 순서를 신경쓰지 않아도 되기 때문이다.
그리고 소비자 그룹 내의 한 소비자는 다른 소비자가 아직 읽지 않은 데이터만 읽어간다.
레디스 stream에서 소비자 그룹은 stream의 상태를 나타내는 개념으로 간주된다.

카프카에서 소비자 그룹에 여러 소비자를 추가할 수 있으며, 이때 소비자는 토픽 내의 파티션과 일대일로 연결된다.

 

레디스 stream과 소비자 그룹은 독립적으로 동작할 수 있다.
소비자 그룹 1의 소비자가 a라는 메시지를 읽었다면 같은 그룹에서는 그 메시지를 다시 읽을 수는 없지만, 소비자 그룹 2 혹은 일반적인 다른 소비자에서는 해당 메시지를 읽을 수 있다.
하나의 소비자 그룹에서 여러 개의 stream을 리스닝하는 것도 가능하다.

ACK와 보류 리스트

여러 서비스가 메시지 브로커를 이용해 데이터를 처리할 때, 예상치 못한 장애로 인해 시스템이 종료됐을 경우 이를 인지하고 재처리할 수 있는 기능이 필요하다.
메시지 브로커는 각 소비자에게 어떤 메시지까지 전달됐고, 전달된 메시지의 처리 유무를 인지하고 있어야 한다.

레디스 stream에서는 소비자 그룹에 속한 소비자가 메시지를 읽어가면 각 소비자별로 읽어간 메시지에 대한 리스트를 새로 생성하며, 마지막으로 읽어간 데이터의 ID로 last_delivered_id 값을 업데이트한다.
레디스 stream은 소비자별로 보류 리스트(pending list)를 만들고, 어떤 소비자가 어떤 데이터를 읽어갔는지 인지하고 있다.
소비자가 stream에게 데이터가 처리됐다는 뜻의 ACK를 보내면 레디스 stream은 보류 리스트에서 ACK를 받은 메시지를 삭제한다.
즉, 보류 리스트를 이용해 소비자가 처리한 데이터를 파악할 수 있다.

카프카도 레디스 stream과 비슷하게 파티션별 오프셋을 관리한다. 카프카는 내부적으로 _consumer_offfset라는 토픽에 데이터를 기록하는데, 소비자가 지정된 토픽의 특정 파티션의 메시지를 읽으면 소비자 그룹, 토픽, 파티션 내용이 통합돼 저장된다. 소비자 그룹은 _consumer_offfset 토픽에 기록된 정보를 이용해 내부 소비자가 어디까지 읽었는지 추적할 수 있다. 카프카에서 오프셋은 소비자가 마지막으로 읽은 위치가 아니라 다음으로 읽어야 할 위치를 기록한다.

 

메시지의 재할당

만약 소비자 서버에 장애가 발생해 복구되지 않는다면, 해당 소비자가 처리하던 보류 중인 메시지들은 다른 소비자가 대신 처리해야 한다.
XCLAIM 커맨드를 이용하면 메시지의 소유권을 다른 소비자에게 할당할 수 있다.

XCLAIM <key> <group> <consumer> <min-idle-time> <ID-1> <ID-2> ... <ID-N>

XCLAIM 커맨드를 사용할 때에는 최소 대기 시간을 지정해야 한다.
이는 메시지가 보류 상태로 머무른 시간이 최소 대기 시간을 초과한 경우에만 소유권을 변경할 수 있도록 해서 같은 메시지가 2개의 다른 소비자에게 중복으로 할당되는 것을 방지할 수 있다.

 

메시지의 자동 재할당

소비자가 직접 보류했던 메시지 중 하나를 자동으로 가져와서 처리할 수 있도록 하는 XAUTOCLAIM 커맨드는 할당 대기 중인 다음 메시지의 ID를 반환하는 방식으로 동작하기 때문에 반복적 호출을 가능하게 한다.

 

메시지의 수동 재할당

만약 메시지에 문제가 있어 처리되지 못할 경우 메시지는 여러 소비자에게 할당 되기를 반복하면서 counter 값이 계속 증가하게 된다.
따라서 counter가 특정 값에 도달하면 이 메시지를 특수한 다른 stream으로 보내 관리자가 추후 처리할 수 있도록 하는 것이 현명할 수 있다.
보통 이런 메시지를 dead letter라 부른다.