Redis Stream은 Redis 버전 5.0에 도입된 새로운 추상 데이터 유형입니다.
개념적으로 Redis Stream은 항목을 추가할 수 있는 목록입니다. 각 항목에는 고유한 식별자가 있습니다. 기본적으로 ID는 자동으로 생성되며 타임스탬프를 포함합니다. 따라서 Unix "tail -f" 명령이 로그 파일을 읽고 새 데이터를 기다리는 동안 정지되는 것처럼 시간이 지남에 따라 레코드 범위를 쿼리하거나 스트림에 도착하는 새 데이터를 받을 수 있습니다. 많은 "tail -f" 프로세스가 서로 충돌하지 않고 동시에 파일을 읽을 수 있는 것처럼 여러 클라이언트가 동시에 스레드를 수신할 수 있습니다.

새로운 데이터 유형의 모든 이점을 이해하기 위해 Redis Stream의 기능을 부분적으로 복제하는 기존 Redis 구조를 간략하게 살펴보겠습니다.

레디스 PUB/SUB

Redis Pub/Sub는 키-값 저장소에 이미 내장된 간단한 메시징 시스템입니다. 그러나 단순성에는 비용이 발생합니다.

• 어떤 이유로 게시자가 실패하면 모든 구독자를 잃습니다.
• 게시자는 모든 구독자의 정확한 주소를 알아야 합니다.
• 데이터가 처리되는 것보다 빠르게 게시되면 게시자는 구독자에게 작업을 과부하시킬 수 있습니다.
• 메시지는 메시지가 전달된 구독자 수와 이 메시지를 얼마나 빨리 처리할 수 있었는지에 관계없이 게시 직후 게시자의 버퍼에서 삭제됩니다.
• 모든 가입자는 동시에 메시지를 받게 됩니다. 구독자들은 동일한 메시지를 처리하는 순서에 대해 어떻게든 서로 동의해야 합니다.
• 구독자가 메시지를 성공적으로 처리했는지 확인하는 기본 제공 메커니즘은 없습니다. 구독자가 메시지를 받고 처리 중에 충돌이 발생하면 게시자는 이에 대해 알 수 없습니다.

레디스 목록

Redis List는 읽기 명령 차단을 지원하는 데이터 구조입니다. 목록의 처음이나 끝에서 메시지를 추가하고 읽을 수 있습니다. 이 구조를 기반으로 분산 시스템에 적합한 스택 또는 대기열을 만들 수 있으며 대부분의 경우 이것으로 충분합니다. Redis Pub/Sub와의 주요 차이점:

• 메시지는 한 클라이언트에게 전달됩니다. 첫 번째 읽기 차단된 클라이언트가 먼저 데이터를 수신합니다.
• Clint는 각 메시지에 대한 읽기 작업을 직접 시작해야 합니다. 목록은 클라이언트에 대해 아무것도 모릅니다.
• 메시지는 누군가가 읽거나 명시적으로 삭제할 때까지 저장됩니다. 데이터를 디스크로 플러시하도록 Redis 서버를 구성하면 시스템의 안정성이 크게 향상됩니다.

스트림 소개

스트림에 항목 추가

팀 XADD 스트림에 새 항목을 추가합니다. 레코드는 단순한 문자열이 아니며 하나 이상의 키-값 쌍으로 구성됩니다. 따라서 각 항목은 이미 구조화되어 있으며 CSV 파일의 구조와 유사합니다.

> XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0

위의 예에서는 이름(키)이 "mystream"인 스트림에 "sensor-id" 및 "온도" 값을 각각 "1234" 및 "19.8"로 추가합니다. 두 번째 인수로 명령은 항목에 할당될 식별자를 사용합니다. 이 식별자는 스트림의 각 항목을 고유하게 식별합니다. 그러나 이 경우 Redis가 새 ID를 생성하도록 하기 위해 *를 전달했습니다. 각각의 새로운 ID가 증가합니다. 따라서 각각의 새 항목은 이전 항목과 관련하여 더 높은 식별자를 갖게 됩니다.

식별자 형식

명령으로 반환된 항목 ID XADD, 두 부분으로 구성됩니다.

{millisecondsTime}-{sequenceNumber}

밀리초시간 — Unix 시간(밀리초)(Redis 서버 시간). 그러나 현재 시간이 이전 녹음 시간과 같거나 이전 녹음 시간보다 작을 경우에는 이전 녹음의 타임스탬프가 사용됩니다. 따라서 서버 시간이 과거로 돌아가더라도 새 식별자는 여전히 증분 속성을 유지합니다.

시퀀스 번호 동일한 밀리초에 생성된 레코드에 사용됩니다. 시퀀스 번호 이전 항목에 비해 1씩 증가합니다. 왜냐하면 시퀀스 번호 크기가 64비트인 경우 실제로는 XNUMX밀리초 내에 생성할 수 있는 레코드 수에 제한이 있어서는 안 됩니다.

이러한 식별자의 형식은 언뜻 보면 이상해 보일 수 있습니다. 신뢰할 수 없는 독자는 왜 시간이 식별자의 일부인지 궁금해할 수 있습니다. 그 이유는 Redis 스트림이 ID별로 범위 쿼리를 지원하기 때문입니다. 식별자는 기록이 생성된 시간과 연결되어 있으므로 시간 범위를 쿼리할 수 있습니다. 명령을 볼 때 구체적인 예를 살펴보겠습니다. 엑스레인지.

어떤 이유로 사용자가 외부 시스템과 관련된 자신의 식별자를 지정해야 하는 경우 이를 명령에 전달할 수 있습니다. XADD 아래와 같이 * 대신:

> XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2

이 경우 ID 증분을 직접 모니터링해야 한다는 점에 유의하세요. 이 예에서 최소 식별자는 "0-1"이므로 명령은 "0-1"보다 작거나 같은 다른 식별자를 허용하지 않습니다.

> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item

스트림당 레코드 수

간단히 명령을 사용하여 스트림의 레코드 수를 얻을 수 있습니다 엑스렌. 예를 들어 이 명령은 다음 값을 반환합니다.

> XLEN somestream
(integer) 2

범위 쿼리 - XRANGE 및 XREVRANGE

범위별로 데이터를 요청하려면 범위의 시작과 끝이라는 두 개의 식별자를 지정해야 합니다. 반환된 범위에는 경계를 포함한 모든 요소가 포함됩니다. 또한 두 개의 특수 식별자 "-"와 "+"가 있는데, 각각 스트림에서 가장 작은(첫 번째 레코드) 식별자와 가장 큰(마지막 레코드) 식별자를 의미합니다. 아래 예에서는 모든 스트림 항목을 나열합니다.

> XRANGE mystream - +
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"
2) 1) 1518951482479-0
   2) 1) "sensor-id"
      2) "9999"
      3) "temperature"
      4) "18.2"

반환된 각 레코드는 식별자와 키-값 쌍 목록이라는 두 요소의 배열입니다. 우리는 이미 레코드 식별자가 시간과 관련되어 있다고 말했습니다. 따라서 특정 기간의 범위를 요청할 수 있습니다. 그러나 요청에 전체 식별자가 아닌 Unix 시간만 지정할 수 있으며 관련 부분은 생략됩니다. 시퀀스 번호. 식별자에서 생략된 부분은 범위 시작 부분에서는 자동으로 XNUMX으로 설정되고 범위 끝 부분에서는 가능한 최대값으로 설정됩니다. 다음은 XNUMX밀리초 범위를 요청하는 방법의 예입니다.

> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"

이 범위에는 항목이 하나만 있지만 실제 데이터 세트에서는 반환되는 결과가 엄청날 수 있습니다. 이러한 이유로 엑스레인지 COUNT 옵션을 지원합니다. 수량을 지정하면 처음 N개의 레코드를 간단히 얻을 수 있습니다. 다음 N개의 레코드(페이지 매김)를 가져와야 하는 경우 마지막으로 받은 ID를 사용하여 늘릴 수 있습니다. 시퀀스 번호 하나씩 다시 물어보세요. 다음 예에서 이를 살펴보겠습니다. 우리는 10개의 요소를 추가하기 시작합니다. XADD (mystream이 이미 10개의 요소로 채워져 있다고 가정). 명령당 2개의 요소를 가져오는 반복을 시작하려면 전체 범위에서 시작하지만 COUNT는 2입니다.

> XRANGE mystream - + COUNT 2
1) 1) 1519073278252-0
   2) 1) "foo"
      2) "value_1"
2) 1) 1519073279157-0
   2) 1) "foo"
      2) "value_2"

다음 두 요소로 계속 반복하려면 수신된 마지막 ID(예: 1519073279157-0)를 선택하고 1을 추가해야 합니다. 시퀀스 번호.
결과 ID(이 경우 1519073279157-1)는 이제 다음 호출에 대한 새로운 범위 시작 인수로 사용될 수 있습니다. 엑스레인지:

> XRANGE mystream 1519073279157-1 + COUNT 2
1) 1) 1519073280281-0
   2) 1) "foo"
      2) "value_3"
2) 1) 1519073281432-0
   2) 1) "foo"
      2) "value_4"

등등. 복잡하기 때문에 엑스레인지 검색하는 경우 O(log(N))이고 M개의 요소를 반환하는 경우 O(M)이므로 각 반복 단계가 빠릅니다. 따라서, 엑스레인지 스트림을 효율적으로 반복할 수 있습니다.

팀 XREVRANGE 동등하다 엑스레인지, 그러나 요소를 역순으로 반환합니다.

> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
   2) 1) "foo"
      2) "value_10"

명령어를 참고해주세요 XREVRANGE 범위 인수 시작 및 중지를 역순으로 취합니다.

XREAD를 사용하여 새 항목 읽기

종종 스트림을 구독하고 새 메시지만 수신하는 작업이 발생합니다. 이 개념은 Redis Pub/Sub 또는 Redis List 차단과 유사해 보일 수 있지만 Redis Stream을 사용하는 방법에는 근본적인 차이점이 있습니다.

1. 각각의 새 메시지는 기본적으로 모든 구독자에게 전달됩니다. 이 동작은 새 메시지를 한 명의 구독자만 읽는 차단 Redis 목록과 다릅니다.
2. Redis Pub/Sub에서는 모든 메시지가 잊혀지고 지속되지 않지만 Stream에서는 모든 메시지가 무기한 보관됩니다(클라이언트가 명시적으로 삭제를 유발하지 않는 한).
3. Redis 스트림을 사용하면 하나의 스트림 내에서 메시지에 대한 액세스를 차별화할 수 있습니다. 특정 구독자는 자신의 개인 메시지 기록만 볼 수 있습니다.

다음 명령을 사용하여 스레드를 구독하고 새 메시지를 받을 수 있습니다. XREAD. 생각보다 조금 더 복잡해요 엑스레인지이므로 먼저 더 간단한 예부터 시작하겠습니다.

> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1519073278252-0
         2) 1) "foo"
            2) "value_1"
      2) 1) 1519073279157-0
         2) 1) "foo"
            2) "value_2"

위의 예는 비차단 형식을 보여줍니다. XREAD. COUNT 옵션은 선택 사항입니다. 실제로 필요한 유일한 명령 옵션은 해당 최대 식별자와 함께 스트림 목록을 지정하는 STREAMS 옵션입니다. 우리는 "STREAMS mystream 0"이라고 썼습니다. "0-0"보다 큰 식별자를 가진 mystream 스트림의 모든 레코드를 수신하려고 합니다. 예제에서 볼 수 있듯이 동시에 여러 스레드를 구독할 수 있으므로 명령은 스레드 이름을 반환합니다. 예를 들어 "STREAMS mystream otherstream 0 0"이라고 쓸 수 있습니다. STREAMS 옵션 다음에는 먼저 필요한 모든 스트림의 이름을 제공한 다음 식별자 목록만 제공해야 합니다.

이 간단한 형식에서 명령은 다음과 비교하여 특별한 작업을 수행하지 않습니다. 엑스레인지. 그런데 흥미로운 점은 우리가 쉽게 방향을 바꿀 수 있다는 것입니다. XREAD BLOCK 인수를 지정하여 차단 명령에:

> XREAD BLOCK 0 STREAMS mystream $

위의 예에서는 시간 제한이 0밀리초인 새 BLOCK 옵션이 지정되었습니다(이는 무기한 대기를 의미함). 또한 mystream 스트림에 대한 일반적인 식별자를 전달하는 대신 특수 식별자 $가 전달되었습니다. 이 특수 식별자는 다음을 의미합니다. XREAD mystream의 최대 식별자를 식별자로 사용해야 합니다. 따라서 우리는 듣기 시작한 순간부터 새로운 메시지만 받게 됩니다. 어떤 면에서는 이는 Unix "tail -f" 명령과 유사합니다.

BLOCK 옵션을 사용할 때 반드시 특수 식별자 $를 사용할 필요는 없습니다. 스트림에 존재하는 모든 식별자를 사용할 수 있습니다. 팀이 차단 없이 즉시 요청을 처리할 수 있으면 그렇게 하고, 그렇지 않으면 차단됩니다.

블로킹 XREAD 한 번에 여러 스레드를 들을 수도 있습니다. 이름만 지정하면 됩니다. 이 경우 명령은 데이터를 수신한 첫 번째 스트림의 레코드를 반환합니다. 특정 스레드에 대해 차단된 첫 번째 구독자가 먼저 데이터를 수신합니다.

소비자 그룹

특정 작업에서는 하나의 스레드 내의 메시지에 대한 구독자 액세스를 제한하려고 합니다. 이것이 유용할 수 있는 예는 스레드에서 다양한 메시지를 수신하여 메시지 처리를 확장할 수 있는 작업자가 있는 메시지 대기열입니다.

세 명의 구독자 C1, C2, C3과 메시지 1, 2, 3, 4, 5, 6, 7을 포함하는 스레드가 있다고 가정하면 메시지는 아래 다이어그램과 같이 제공됩니다.

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1

이 효과를 달성하기 위해 Redis Stream은 소비자 그룹이라는 개념을 사용합니다. 이 개념은 스트림에서 데이터를 수신하지만 실제로는 그룹 내의 여러 구독자가 서비스를 제공하여 특정 보장을 제공하는 의사 구독자와 유사합니다.

1. 각 메시지는 그룹 내의 다른 구독자에게 전달됩니다.
2. 그룹 내에서 가입자는 대소문자를 구분하는 문자열인 이름으로 식별됩니다. 가입자가 일시적으로 그룹에서 탈퇴한 경우 고유한 이름을 사용하여 그룹에 복원할 수 있습니다.
3. 모든 소비자 그룹은 "읽지 않은 첫 번째 메시지" 개념을 따릅니다. 구독자가 새 메시지를 요청하면 이전에 그룹 내의 구독자에게 전달된 적이 없는 메시지만 수신할 수 있습니다.
4. 구독자가 메시지를 성공적으로 처리했음을 명시적으로 확인하는 명령이 있습니다. 이 명령이 호출될 때까지 요청된 메시지는 "보류 중" 상태로 유지됩니다.
5. 소비자 그룹 내에서 각 가입자는 자신에게 전달되었지만 아직 처리되지 않은("보류 중" 상태) 메시지 기록을 요청할 수 있습니다.

어떤 의미에서 그룹의 상태는 다음과 같이 표현될 수 있습니다.

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" with pending messages  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" with pending messages 
|       ... (and so forth)                             
+----------------------------------------+

이제 소비자 그룹의 주요 명령, 즉 다음과 같은 사항에 대해 알아볼 차례입니다.

• X그룹 그룹을 생성, 삭제 및 관리하는 데 사용됩니다.
• XREAD그룹 그룹을 통해 스트림을 읽는 데 사용됩니다.
• XACK - 이 명령을 사용하면 구독자는 메시지가 성공적으로 처리된 것으로 표시할 수 있습니다.

소비자그룹 창설

mystream이 이미 존재한다고 가정해 보겠습니다. 그러면 그룹 생성 명령은 다음과 같습니다.

> XGROUP CREATE mystream mygroup $
OK

그룹을 생성할 때 그룹이 메시지를 수신할 식별자를 전달해야 합니다. 모든 새 메시지를 수신하려면 위의 예와 같이 특수 식별자 $를 사용할 수 있습니다. 특수 식별자 대신 0을 지정하면 스레드의 모든 메시지를 그룹에서 사용할 수 있습니다.

이제 그룹이 생성되었으므로 다음 명령을 사용하여 즉시 메시지 읽기를 시작할 수 있습니다. XREAD그룹. 이 명령은 다음과 매우 유사합니다. XREAD 선택적 BLOCK 옵션을 지원합니다. 그러나 항상 두 개의 인수, 즉 그룹 이름과 구독자 이름을 사용하여 지정해야 하는 필수 GROUP 옵션이 있습니다. COUNT 옵션도 지원됩니다.

스레드를 읽기 전에 여기에 몇 가지 메시지를 넣어 보겠습니다.

> XADD mystream * message apple
1526569495631-0
> XADD mystream * message orange
1526569498055-0
> XADD mystream * message strawberry
1526569506935-0
> XADD mystream * message apricot
1526569535168-0
> XADD mystream * message banana
1526569544280-0

이제 그룹을 통해 이 스트림을 읽어 보겠습니다.

> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

위 명령은 다음과 같이 축어적으로 읽습니다.

“나, mygroup의 회원인 구독자 Alice는 이전에 누구에게도 전달된 적이 없는 mystream의 메시지 하나를 읽고 싶습니다.”

구독자는 그룹에서 작업을 수행할 때마다 그룹 내에서 자신을 고유하게 식별하는 이름을 제공해야 합니다. 위 명령에는 특수 식별자 ">"라는 매우 중요한 세부 사항이 하나 더 있습니다. 이 특수 식별자는 메시지를 필터링하여 이전에 전달된 적이 없는 메시지만 남깁니다.

또한 특별한 경우에는 0이나 다른 유효한 식별자와 같은 실제 식별자를 지정할 수 있습니다. 이 경우 명령은 XREAD그룹 지정된 구독자(Alice)에게 전달되었지만 아직 명령을 사용하여 확인되지 않은 "보류 중" 상태의 메시지 기록을 반환합니다. XACK.

옵션 없이 ID 0을 즉시 지정하여 이 동작을 테스트할 수 있습니다. COUNT. 우리는 단순히 단일 보류 메시지, 즉 사과 메시지를 보게 될 것입니다:

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"

그러나 메시지가 성공적으로 처리되었음을 확인하면 더 이상 표시되지 않습니다.

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)

이제 Bob이 읽을 차례입니다.

> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569498055-0
         2) 1) "message"
            2) "orange"
      2) 1) 1526569506935-0
         2) 1) "message"
            2) "strawberry"

mygroup의 회원인 Bob은 두 개 이하의 메시지를 요청했습니다. 이 명령은 특수 식별자 ">"로 인해 배달되지 않은 메시지만 보고합니다. 보시다시피 "apple"이라는 메시지는 이미 Alice에게 전달되었기 때문에 표시되지 않습니다. 따라서 Bob은 "orange"와 "strawberry"를 받습니다.

이렇게 하면 Alice, Bob 및 그룹의 다른 모든 구독자가 동일한 스트림에서 다른 메시지를 읽을 수 있습니다. 또한 처리되지 않은 메시지 기록을 읽거나 메시지를 처리된 것으로 표시할 수도 있습니다.

명심해야 할 몇 가지 사항이 있습니다.

• 구독자가 메시지를 명령으로 간주하는 즉시 XREAD그룹, 이 메시지는 "보류 중" 상태가 되고 해당 특정 구독자에게 할당됩니다. 다른 그룹 가입자는 이 메시지를 읽을 수 없습니다.
• 구독자는 처음 언급될 때 자동으로 생성되므로 명시적으로 생성할 필요가 없습니다.
• 과 XREAD그룹 동시에 여러 다른 스레드에서 메시지를 읽을 수 있지만 이 작업을 수행하려면 먼저 다음을 사용하여 각 스레드에 대해 동일한 이름을 가진 그룹을 만들어야 합니다. X그룹

장애 후 복구

구독자는 오류를 복구하고 "보류 중" 상태의 메시지 목록을 다시 읽을 수 있습니다. 그러나 현실 세계에서는 구독자가 궁극적으로 실패할 수 있습니다. 구독자가 오류를 복구할 수 없는 경우 구독자의 정체된 메시지는 어떻게 되나요?
Consumer Group은 메시지 소유자를 변경해야 하는 경우에 사용되는 기능을 제공합니다.

가장 먼저 해야 할 일은 명령을 호출하는 것입니다. 확장 중, 그룹에서 '보류 중' 상태의 모든 메시지를 표시합니다. 가장 간단한 형태로 명령은 스레드 이름과 그룹 이름이라는 두 개의 인수만 사용하여 호출됩니다.

> XPENDING mystream mygroup
1) (integer) 2
2) 1526569498055-0
3) 1526569506935-0
4) 1) 1) "Bob"
      2) "2"

팀은 전체 그룹과 각 구독자에 대해 처리되지 않은 메시지 수를 표시했습니다. Alice가 요청한 유일한 메시지가 확인되었기 때문에 두 개의 미해결 메시지가 있는 Bob만 있습니다. XACK.

더 많은 인수를 사용하여 더 많은 정보를 요청할 수 있습니다.

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} - 식별자 범위(“-” 및 “+” 사용 가능)
{count} — 배송 시도 횟수
{소비자 이름} - 그룹 이름

> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-0
   2) "Bob"
   3) (integer) 74170458
   4) (integer) 1
2) 1) 1526569506935-0
   2) "Bob"
   3) (integer) 74170458
   4) (integer) 1

이제 각 메시지에 대한 세부 정보(ID, 구독자 이름, 밀리초 단위의 유휴 시간, 마지막으로 배달 시도 횟수)가 있습니다. Bob이 보낸 두 개의 메시지가 있으며 해당 메시지는 74170458밀리초, 즉 약 20시간 동안 유휴 상태였습니다.

단순히 다음을 사용하여 메시지 내용이 무엇인지 확인하는 것을 막을 수 있는 사람은 아무도 없습니다. 엑스레인지.

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

인수에서 동일한 식별자를 두 번 반복하면 됩니다. 이제 몇 가지 아이디어를 얻었으므로 Alice는 20시간의 가동 중지 시간이 지나면 Bob이 회복하지 못할 것이라고 판단하고 이제 Bob을 위해 해당 메시지를 쿼리하고 처리를 재개해야 할 때라고 판단할 수 있습니다. 이를 위해 우리는 명령을 사용합니다 X클레임:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}

이 명령을 사용하면 소유자를 {consumer}로 변경하여 아직 처리되지 않은 "외부" 메시지를 수신할 수 있습니다. 그러나 최소 유휴 시간({min-idle-time})도 제공할 수 있습니다. 이렇게 하면 두 클라이언트가 동일한 메시지의 소유자를 동시에 변경하려고 하는 상황을 방지하는 데 도움이 됩니다.

Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0

첫 번째 고객은 가동 중지 시간을 재설정하고 배송 카운터를 늘립니다. 따라서 두 번째 클라이언트는 이를 요청할 수 없습니다.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

이제 메시지를 처리하고 확인할 수 있는 Alice가 메시지를 성공적으로 요청했습니다.

위의 예에서 성공적인 요청은 메시지 자체의 내용을 반환하는 것을 볼 수 있습니다. 그러나 반드시 그럴 필요는 없습니다. JUSTID 옵션은 메시지 ID만 반환하는 데 사용할 수 있습니다. 이는 메시지의 세부사항에 관심이 없고 시스템 성능을 향상시키려는 경우에 유용합니다.

배송 카운터

출력에 표시되는 카운터 확장 중 각 메시지의 전달 횟수입니다. 이러한 카운터는 두 가지 방법으로 증가합니다. 메시지가 다음을 통해 성공적으로 요청된 경우 X클레임 또는 통화가 사용될 때 XREAD그룹.

일부 메시지가 여러 번 전달되는 것은 정상입니다. 가장 중요한 것은 모든 메시지가 결국 처리된다는 것입니다. 메시지 자체가 손상되어 메시지를 처리할 때 문제가 발생하거나, 메시지 처리로 인해 핸들러 코드에 오류가 발생하는 경우도 있습니다. 이 경우 누구도 이 메시지를 처리하지 못할 수도 있습니다. 배송 시도 카운터가 있으므로 이 카운터를 사용하여 이러한 상황을 감지할 수 있습니다. 따라서 배달 횟수가 지정한 최대 수에 도달하면 해당 메시지를 다른 스레드에 넣고 시스템 관리자에게 알림을 보내는 것이 더 현명할 것입니다.

스레드 상태

팀 신포 스레드 및 해당 그룹에 대한 다양한 정보를 요청하는 데 사용됩니다. 예를 들어 기본 명령은 다음과 같습니다.

> XINFO STREAM mystream
 1) length
 2) (integer) 13
 3) radix-tree-keys
 4) (integer) 1
 5) radix-tree-nodes
 6) (integer) 2
 7) groups
 8) (integer) 2
 9) first-entry
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) last-entry
12) 1) 1526569544280-0
    2) 1) "message"
       2) "banana"

위 명령은 지정된 스트림에 대한 일반 정보를 표시합니다. 이제 약간 더 복잡한 예를 들어보겠습니다.

> XINFO GROUPS mystream
1) 1) name
   2) "mygroup"
   3) consumers
   4) (integer) 2
   5) pending
   6) (integer) 2
2) 1) name
   2) "some-other-group"
   3) consumers
   4) (integer) 1
   5) pending
   6) (integer) 0

위의 명령은 지정된 스레드의 모든 그룹에 대한 일반 정보를 표시합니다.

> XINFO CONSUMERS mystream mygroup
1) 1) name
   2) "Alice"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 9104628
2) 1) name
   2) "Bob"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 83841983

위 명령은 지정된 스트림 및 그룹의 모든 구독자에 대한 정보를 표시합니다.
명령 구문을 잊어버린 경우 명령 자체에 도움을 요청하세요.

> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Subcommands are:
2) CONSUMERS {key} {groupname}  -- Show consumer groups of group {groupname}.
3) GROUPS {key}                 -- Show the stream consumer groups.
4) STREAM {key}                 -- Show information about the stream.
5) HELP                         -- Print this help.

스트림 크기 제한

많은 애플리케이션은 데이터를 스트림으로 영원히 수집하는 것을 원하지 않습니다. 스레드당 허용되는 최대 메시지 수를 지정하는 것이 유용한 경우가 많습니다. 다른 경우에는 지정된 스레드 크기에 도달하면 스레드의 모든 메시지를 다른 영구 저장소로 이동하는 것이 유용합니다. 명령에서 MAXLEN 매개변수를 사용하여 스트림 크기를 제한할 수 있습니다. XADD:

> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

MAXLEN을 사용하면 오래된 레코드가 지정된 길이에 도달하면 자동으로 삭제되므로 스트림의 크기가 일정합니다. 그러나 이 경우 정리는 Redis 메모리에서 가장 효율적인 방식으로 발생하지 않습니다. 다음과 같이 상황을 개선할 수 있습니다.

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

위 예의 ~ 인수는 스트림 길이를 특정 값으로 제한할 필요가 없음을 의미합니다. 이 예에서는 1000보다 크거나 같은 숫자일 수 있습니다(예: 1000, 1010 또는 1030). 우리는 스트림이 최소 1000개의 레코드를 저장하도록 명시적으로 지정했습니다. 이는 Redis 내에서 메모리 관리를 훨씬 더 효율적으로 만듭니다.

팀도 따로 있고 엑스트림, 동일한 작업을 수행합니다.

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10


영구 저장 및 복제

Redis 스트림은 슬레이브 노드에 비동기식으로 복제되며 AOF(모든 데이터의 스냅샷) 및 RDB(모든 쓰기 작업 로그)와 같은 파일에 저장됩니다. 소비자 그룹 상태 복제도 지원됩니다. 따라서 메시지가 마스터 노드에서 "보류 중" 상태인 경우 슬레이브 노드에서 이 메시지는 동일한 상태를 갖습니다.

스트림에서 개별 요소 제거

메시지를 삭제하는 특별한 명령이 있습니다 XDEL. 이 명령은 스레드 이름과 삭제할 메시지 ID를 가져옵니다.

> XRANGE mystream - + COUNT 2
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"
> XDEL mystream 1526654999635-0
(integer) 1
> XRANGE mystream - + COUNT 2
1) 1) 1526655000369-0
   2) 1) "value"
      2) "3"

이 명령을 사용할 때 실제 메모리가 즉시 해제되지 않는다는 점을 고려해야 합니다.

길이가 XNUMX인 스트림

스트림과 다른 Redis 데이터 구조의 차이점은 다른 데이터 구조에 더 이상 요소가 없으면 부작용으로 데이터 구조 자체가 메모리에서 제거된다는 것입니다. 예를 들어 ZREM 호출이 마지막 요소를 제거하면 정렬된 세트가 완전히 제거됩니다. 대신 스레드는 내부에 요소가 없어도 메모리에 남아 있을 수 있습니다.

결론

Redis Stream은 메시지 브로커, 메시지 대기열, 통합 로깅 및 기록 유지 채팅 시스템을 만드는 데 이상적입니다.

내가 한번 말했듯이 니클라우스 워스, 프로그램은 알고리즘과 데이터 구조를 더한 것이며 Redis는 이미 두 가지를 모두 제공합니다.

출처 : habr.com