Redis Stream — um novo tipo abstrato de dado introduzido no Redis com o lançamento da versão 5.0
Conceitualmente, o Redis Stream é uma lista na qual você pode adicionar registros. Cada registro possui um identificador único. Por padrão, o identificador é gerado automaticamente e inclui um timestamp. Assim, você pode consultar intervalos de registros por tempo ou receber novos dados à medida que eles chegam ao stream, assim como o comando Unix “tail -f” lê um arquivo de log e aguarda novos dados. Note que o stream pode ser escutado por vários clientes simultaneamente, assim como muitos processos “tail -f” podem ler um arquivo ao mesmo tempo, sem entrar em conflito.

Para entender todas as vantagens do novo tipo de dado, vamos relembrar rapidamente as estruturas do Redis que já existem e que em parte replicam a funcionalidade do Redis Stream.

Redis PUB/SUB

O Redis Pub/Sub é um sistema de mensagens simples, já integrado ao seu armazenamento chave-valor. No entanto, essa simplicidade tem um custo:

Se o editor falhar por algum motivo, ele perde todos os seus assinantes.
O editor precisa saber o endereço exato de todos os seus assinantes.
O editor pode sobrecarregar o trabalho de seus assinantes, se os dados forem publicados mais rapidamente do que podem ser processados.
A mensagem é removida do buffer do editor imediatamente após a publicação, independentemente de quantos assinantes a receberam e quão rapidamente conseguiram processar essa mensagem.
Todos os assinantes receberão a mensagem ao mesmo tempo. Os assinantes devem de alguma forma coordenar entre si a ordem de processamento da mesma mensagem.
Não há um mecanismo embutido de confirmação de recebimento bem-sucedido da mensagem pelo assinante. Se o assinante recebeu a mensagem e falhou durante o processamento, o editor não terá conhecimento disso.

Redis List

O Redis List é uma estrutura de dados que suporta comandos de leitura com bloqueio. Você pode adicionar e ler mensagens do início ou do final da lista. Com base nessa estrutura, é possível criar uma boa pilha ou fila para seu sistema distribuído e, na maioria dos casos, isso será suficiente. As principais diferenças em relação ao Redis Pub/Sub:

A mensagem é entregue a um único cliente. O primeiro cliente bloqueado na leitura receberá os dados primeiro.
Clint deve iniciar a operação de leitura de cada mensagem. List nada sabe sobre os clientes.
As mensagens são armazenadas até que alguém as leia ou as exclua explicitamente. Se você configurou o servidor Redis para gravar dados no disco, a confiabilidade do sistema aumenta consideravelmente.

Introdução ao Stream

Adicionando uma entrada ao stream

Time XADD adiciona uma nova entrada ao stream. Uma entrada não é apenas uma string, mas consiste em um ou mais pares chave-valor. Assim, cada entrada já é estruturada e lembra a estrutura de um arquivo CSV.

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

No exemplo acima, estamos adicionando ao stream chamado (chave) 'mystream' dois campos: 'sensor-id' e 'temperature' com os valores '1234' e '19.8', respectivamente. Como segundo argumento, o comando aceita um identificador que será atribuído à entrada - este identificador identifica de forma única cada entrada no stream. No entanto, neste caso, passamos *, porque queremos que o Redis gere um novo identificador para nós. Cada novo identificador será incremental. Portanto, cada nova entrada terá um identificador maior em relação às entradas anteriores.

Formato do identificador

O identificador da entrada retornado pelo comando XADD, consiste em duas partes:

{millisecondsTime}-{sequenceNumber}
millisecondsTime — O tempo Unix em milissegundos (tempo servidor Redis). No entanto, se o tempo atual for igual ou menor que o tempo da entrada anterior, será utilizada a marca temporal da entrada anterior. Portanto, se o tempo do servidor voltar ao passado, o novo identificador ainda manterá sua propriedade de incremento.

sequenceNumber é usado para entradas criadas no mesmo milissegundo. sequenceNumber será incrementado em 1 em relação à entrada anterior. Uma vez que sequenceNumber tem um tamanho de 64 bits, na prática, você não deve encontrar limites na quantidade de entradas que podem ser geradas em um milissegundo.

À primeira vista, o formato desses identificadores pode parecer estranho. Um leitor cético pode questionar por que o tempo faz parte do identificador. A razão é que os streams do Redis suportam consultas de intervalo por identificadores. Como o identificador está relacionado ao momento de criação do registro, isso permite solicitar intervalos de tempo. Vamos considerar um exemplo específico quando começarmos a estudar o comando. XRANGE.

Se, por algum motivo, o usuário precisar especificar seu próprio identificador, que, por exemplo, esteja relacionado a algum sistema externo, podemos passá-lo para o comando XADD em vez do caractere * como mostrado abaixo:

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

Observe que, nesse caso, você deve cuidar do incremento do identificador. No nosso exemplo, o identificador mínimo é "0-1", portanto, o comando não aceitará outro identificador que seja igual ou menor que "0-1".

> XADD somestream 0-1 foo bar
(error) ERR O ID especificado em XADD é igual ou menor que o item de topo do stream de destino.

Número de registros no stream

Pode-se obter o número de registros no stream utilizando apenas o comando XLEN. Para o nosso exemplo, este comando retornará o seguinte valor:

> XLEN somestream
(integer) 2

Consultas de intervalo — XRANGE e XREVRANGE

Para consultar dados em um intervalo, precisamos especificar dois identificadores — o início e o fim do intervalo. O intervalo retornado incluirá todos os elementos, incluindo as bordas. Existem também dois identificadores especiais "-" e "+", que significam, respectivamente, o menor (primeiro registro) e o maior (último registro) identificador no stream. O exemplo abaixo exibirá todos os registros do stream.

> 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"

Cada registro retornado representa um array de dois elementos: o identificador e uma lista de pares chave-valor. Já mencionamos que os identificadores de registro estão relacionados ao tempo. Portanto, podemos solicitar um intervalo de um período específico. No entanto, podemos indicar na solicitação não um identificador completo, mas apenas o tempo Unix, omitindo a parte relacionada ao sequenceNumberA parte omitida do identificador será automaticamente igual a zero no início do intervalo e ao valor máximo possível no final do intervalo. Veja abaixo um exemplo de como solicitar um intervalo de dois milissegundos.

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

Temos apenas um registro neste intervalo; no entanto, em conjuntos de dados reais, o resultado retornado pode ser imenso. Por esse motivo, XRANGE o suporte à opção COUNT é fornecido. Especificando a quantidade, podemos simplesmente obter os primeiros N registros. Se precisarmos obter os próximos N registros (paginação), podemos usar o último identificador obtido, incrementá-lo sequenceNumber em uma unidade e solicitar novamente. Vamos ilustrar isso no próximo exemplo. Começamos a adicionar 10 elementos usando XADD (supondo que o stream mystream já tenha sido preenchido com 10 elementos). Para começar a iteração, recebendo 2 elementos por comando, começamos com o intervalo completo, mas com COUNT igual a 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"

Para continuar a iteração com os próximos dois elementos, precisamos pegar o último identificador obtido, que é 1519073279157-0, e adicionar 1 ao sequenceNumber.
identificador resultante, neste caso 1519073279157-1, que agora pode ser usado como um novo argumento de início de intervalo para a próxima chamada. XRANGE:

> 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"

E assim por diante. Como a complexidade XRANGE é O(log (N)) para a busca e O(M) para retornar M elementos, cada passo da iteração é rápido. Assim, com XRANGE é possível iterar streams de maneira eficiente.

Time XREVRANGE é equivalente XRANGE, mas retorna os elementos em ordem inversa:

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

Observe que o comando XREVRANGE recebe os argumentos de intervalo start e stop em ordem inversa.

A leitura de novos registros usando XREAD

frequentemente levanta a questão de assinar um stream e receber apenas novas mensagens. Este conceito pode parecer semelhante ao Redis Pub/Sub ou a uma lista bloqueadora do Redis, mas há diferenças fundamentais na utilização do Redis Stream:

Cada nova mensagem é entregue, por padrão, a todos os assinantes. Esse comportamento difere da lista Redis bloqueante, onde uma nova mensagem só será lida por um único assinante.
Enquanto no Redis Pub/Sub todas as mensagens são esquecidas e nunca são armazenadas, no Stream todas as mensagens são armazenadas por tempo indeterminado (a menos que o cliente chame explicitamente a exclusão).
O Redis Stream permite restringir o acesso às mensagens dentro de um único fluxo. Um assinante específico pode ver apenas seu próprio histórico pessoal de mensagens.

Você pode se inscrever no fluxo e receber novas mensagens usando o comando XREAD. Isso é um pouco mais complicado do que XRANGE, então vamos começar com exemplos mais simples primeiro.

> 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"

No exemplo acima, está indicada uma forma não bloqueante. XREADObserve que a opção COUNT não é obrigatória. Na verdade, a única opção obrigatória do comando é a opção STREAMS, que define a lista de fluxos juntamente com o identificador máximo correspondente. Escrevemos "STREAMS mystream 0" — queremos receber todos os registros do fluxo mystream com um identificador maior que "0-0". Como pode ser visto no exemplo, o comando retorna o nome do fluxo, porque podemos nos inscrever em vários fluxos ao mesmo tempo. Poderíamos escrever, por exemplo, "STREAMS mystream otherstream 0 0". Observe que, após a opção STREAMS, precisamos primeiro fornecer os nomes de todos os fluxos necessários e somente então a lista de identificadores.

Nesta forma simples, o comando não faz nada de especial em comparação com XRANGE. No entanto, o interessante é que podemos facilmente transformar XREAD em um comando bloqueante, especificando o argumento BLOCK:

> XREAD BLOCK 0 STREAMS mystream $
No exemplo acima, foi indicada uma nova opção BLOCK com um tempo de espera de 0 milissegundos (o que significa espera infinita). Além disso, em vez de passar um identificador comum para o fluxo mystream, foi passado um identificador especial $. Esse identificador especial significa que XREAD deve usar como identificador o identificador máximo no fluxo mystream. Assim, receberemos apenas novas mensagens a partir do momento em que começamos a ouvir. De certa forma, isso é semelhante ao comando Unix "tail -f".

Observe que ao usar a opção BLOCK, não precisamos necessariamente usar um identificador especial $. Podemos utilizar qualquer identificador existente no fluxo. Se a equipe puder atender nossa solicitação imediatamente, sem bloqueios, ela o fará; caso contrário, ela será bloqueada.

Bloqueante XREAD também pode escutar vários fluxos ao mesmo tempo, bastando indicar seus nomes. Nesse caso, a equipe retornará a gravação do primeiro fluxo em que os dados chegarem. O primeiro assinante bloqueado para esse fluxo receberá os dados primeiro.

Grupos de Consumidores

Em certas tarefas, queremos delimitar o acesso dos assinantes às mensagens dentro de um único fluxo. Um exemplo em que isso pode ser útil é uma fila de mensagens com trabalhadores, que receberão diferentes mensagens do fluxo, permitindo escalar o processamento das mensagens.

Se imaginarmos que temos três assinantes C1, C2, C3 e um fluxo que contém as mensagens 1, 2, 3, 4, 5, 6, 7, o atendimento das mensagens ocorrerá como no diagrama abaixo:

1 -> C1 2 -> C2 3 -> C3 4 -> C1 5 -> C2 6 -> C3 7 -> C1
Para obter esse efeito, o Redis Stream utiliza um conceito chamado Grupo de Consumidores. Este conceito é semelhante a um pseudo-assinante que recebe dados do fluxo, mas é na verdade atendido por vários assinantes dentro do grupo, fornecendo certas garantias:

Cada mensagem é entregue a diferentes assinantes dentro do grupo.
Dentro do grupo, os assinantes são identificados por um nome, que é uma string que diferencia maiúsculas de minúsculas. Se algum assinante sair temporariamente do grupo, ele pode se reconectar ao grupo usando seu próprio nome único.
Cada Grupo de Consumidores segue o conceito de 'primeira mensagem não lida'. Quando um assinante solicita novas mensagens, ele pode receber apenas aquelas que nunca foram entregues a nenhum assinante dentro do grupo.
Existe um comando de confirmação explícita para o processamento bem-sucedido da mensagem pelo assinante. Enquanto esse comando não for chamado, a mensagem solicitada permanecerá no status 'pendente'.
Dentro do Grupo de Consumidores, cada assinante pode solicitar o histórico das mensagens que foram entregues a ele, mas que ainda não foram processadas (no status 'pendente').

De certa forma, o estado do grupo pode ser representado assim:

+----------------------------------------+
| consumer_group_name: mygroup          
| consumer_group_stream: somekey        
| last_delivered_id: 1292309234234-92    
|                                                           
| consumers:                                          
|    "consumer-1" com mensagens pendentes  
|       1292309234234-4                          
|       1292309234232-8                          
|    "consumer-42" com mensagens pendentes 
|       ... (e assim por diante)                             
+----------------------------------------+

Agora é hora de se familiarizar com os principais comandos para o Consumer Group, a saber:

XGROUP usado para criar, destruir e gerenciar grupos
XREADGROUP usado para ler o fluxo através do grupo
XACK — este comando permite que o assinante marque a mensagem como processada com sucesso

Criando o Consumer Group

Suponha que o fluxo mystream já exista. Então, o comando para criar o grupo seria:

> XGROUP CREATE mystream mygroup $ OK
Ao criar o grupo, devemos passar o identificador a partir do qual o grupo começará a receber mensagens. Se quisermos simplesmente receber todas as novas mensagens, podemos usar o identificador especial $ (como no nosso exemplo acima). Se em vez do identificador especial colocarmos 0, todas as mensagens do fluxo estarão disponíveis para o grupo.

Agora que o grupo foi criado, podemos imediatamente começar a ler mensagens usando o comando XREADGROUP. Este comando é muito semelhante ao XREAD e suporta a opção opcional BLOCK. No entanto, há uma opção obrigatória GROUP, que deve ser sempre especificada com dois argumentos: o nome do grupo e o nome do assinante. A opção COUNT também é suportada.

Antes de ler o fluxo, vamos colocar algumas mensagens lá:

> 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

Agora vamos tentar ler este fluxo através do grupo:

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

O comando acima diz literalmente o seguinte:

«Eu, Alice, assinante, membro do grupo mygroup, quero ler do fluxo mystream uma mensagem que nunca foi entregue a ninguém antes».

Toda vez que um assinante realiza uma operação com um grupo, ele deve indicar seu nome, identificando-se de forma única dentro do grupo. Na equipe acima, há outro detalhe muito importante — um identificador especial ‘>’. Este identificador especial filtra as mensagens, deixando apenas aquelas que ainda não foram entregues.

Além disso, em casos especiais, você pode especificar um identificador real, como 0 ou qualquer outro identificador válido. Nesse caso, o comando XREADGROUP retornará o histórico de mensagens com status ‘pending’, que foram entregues ao assinante especificado (Alice), mas ainda não foram confirmadas com o comando XACK.

Podemos verificar esse comportamento, especificando imediatamente o identificador 0, sem a opção CONTAGEM. Apenas veremos a única mensagem pendente, ou seja, a mensagem com a maçã:

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

No entanto, se confirmarmos a mensagem como processada com sucesso, ela não será mais exibida:

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (lista ou conjunto vazio)

Agora é a vez de Bob ler algo:

> 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"

Bob, membro do grupo mygroup, pediu não mais do que duas mensagens. O comando informa apenas sobre as mensagens não entregues devido ao identificador especial ‘>’. Como você pode ver, a mensagem ‘apple’ não aparece, pois já foi entregue a Alice, portanto Bob recebe ‘orange’ e ‘strawberry’.

Assim, Alice, Bob e qualquer outro assinante do grupo podem ler mensagens diferentes do mesmo fluxo. Eles também podem ler seu histórico de mensagens não processadas ou marcar mensagens como processadas.

Há algumas coisas a serem lembradas:

Assim que o assinante lê a mensagem com o comando XREADGROUP, essa mensagem se torna ‘pending’ e é atribuída a esse assinante específico. Outros assinantes do grupo não poderão ler essa mensagem.
Os assinantes são criados automaticamente na primeira menção, não há necessidade de criá-los explicitamente.
Com XREADGROUP Você pode ler mensagens de vários fluxos diferentes ao mesmo tempo, no entanto, para que isso funcione, você precisa criar previamente grupos com o mesmo nome para cada fluxo usando XGROUP

Recuperação após falha

O assinante pode se recuperar de uma falha e reler sua lista de mensagens com status "pending". No entanto, no mundo real, os assinantes podem falhar permanentemente. O que acontece com as mensagens pendentes do assinante se ele não conseguir se recuperar da falha?
O Consumer Group oferece uma função que é utilizada exatamente para esses casos — quando é necessário mudar a propriedade das mensagens.

Primeiro, é necessário chamar o comando XPENDING, que exibe todas as mensagens do grupo com status "pending". Em sua forma mais simples, o comando é chamado apenas com dois argumentos: o nome do fluxo e o nome do grupo:

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

O comando mostrou o número de mensagens não processadas para todo o grupo e para cada assinante. Temos apenas Bob com duas mensagens não processadas, pois a única mensagem solicitada por Alice foi confirmada com XACK.

Podemos solicitar informações adicionais, usando mais argumentos:

XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} — intervalo de identificadores (é possível usar "-" e "+")
{count} — número de tentativas de entrega
{consumer-name} — nome do grupo

> 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

Agora temos detalhes para cada mensagem: identificador, nome do assinante, tempo de espera em milissegundos e, por fim, o número de tentativas de entrega. Temos duas mensagens de Bob, e elas estão pendentes por 74170458 milissegundos, cerca de 20 horas.

Note que ninguém nos impede de verificar qual foi o conteúdo da mensagem, apenas usando XRANGE.

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "mensagem"
      2) "laranja"

Precisamos simplesmente repetir o mesmo identificador duas vezes nos argumentos. Agora que temos uma ideia, Alice pode decidir que, após 20 horas pendente, Bob provavelmente não se recuperará e é hora de solicitar essas mensagens e retomar seu processamento em vez de Bob. Para isso, usamos o comando XCLAIM:

XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
Com este comando, podemos obter uma mensagem "estranha" que ainda não foi processada, mudando o proprietário para {consumer}. No entanto, também podemos fornecer um tempo mínimo de inatividade {min-idle-time}. Isso ajuda a evitar situações em que dois clientes tentam mudar o proprietário das mesmas mensagens ao mesmo tempo:

Cliente 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0 Cliente 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
O primeiro cliente redefinirá o tempo de inatividade e aumentará o contador de entregas. Assim, o segundo cliente não poderá solicitá-lo.

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "mensagem"
      2) "laranja"

A mensagem foi solicitada com sucesso por Alice, que agora pode processá-la e confirmá-la.

Do exemplo acima, podemos ver que a execução bem-sucedida da solicitação retorna o conteúdo da própria mensagem. No entanto, isso não é obrigatório. A opção JUSTID pode ser usada para retornar apenas os identificadores da mensagem. Isso é útil se você não estiver interessado nos detalhes da mensagem e quiser melhorar o desempenho do sistema.

Contador de entrega

O contador que você observa na saída XPENDING — é o número de entregas de cada mensagem. Esse contador aumenta de duas maneiras: quando a mensagem é solicitada com sucesso através de XCLAIM ou quando é feita uma chamada XREADGROUP.

É normal que algumas mensagens sejam entregues várias vezes. O importante é que, no final, todas as mensagens sejam processadas. Às vezes, ao processar uma mensagem, ocorrem problemas devido a um problema na própria mensagem ou o processamento da mensagem gera um erro no código do manipulador. Nesse caso, pode ser que ninguém consiga processar essa mensagem. Como temos um contador de tentativas de entrega, podemos usar esse contador para detectar tais situações. Portanto, assim que o contador de entregas alcançar um número alto que você definiu, provavelmente será mais prudente colocar essa mensagem em um outro fluxo e notificar o administrador do sistema.

Estado dos fluxos

Time XINFO é usado para solicitar várias informações sobre o fluxo e seus grupos. Por exemplo, a forma básica do comando é a seguinte:

> 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) "mensagem"
       2) "banana"

O comando acima exibe as informações gerais sobre o stream especificado. Agora um exemplo um pouco mais complexo:

> 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

O comando acima exibe informações gerais sobre todos os grupos do stream especificado

> 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

O comando acima exibe informações sobre todos os assinantes do stream e grupo especificados.
Se você esquecer a sintaxe do comando, basta pedir ajuda ao próprio comando:

> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Os subcomandos são:
2) CONSUMERS {key} {groupname}  -- Mostra os grupos de consumidores do grupo {groupname}.
3) GROUPS {key}                 -- Mostra os grupos de consumidores do stream.
4) STREAM {key}                 -- Mostra informações sobre o stream.
5) HELP                         -- Imprime esta ajuda.

Limite de tamanho do stream

Muitos aplicativos não desejam coletar dados em um stream para sempre. Muitas vezes, é útil ter um número máximo de mensagens no stream. Em outros casos, é útil mover todas as mensagens do stream para outro armazenamento permanente ao atingir o tamanho definido. O tamanho do stream pode ser limitado usando o parâmetro MAXLEN no comando 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"

Ao usar MAXLEN, os registros antigos são automaticamente removidos ao se atingir o comprimento especificado, mantendo assim o stream com um tamanho constante. No entanto, a remoção, nesse caso, não ocorre da maneira mais eficiente na memória do Redis. A situação pode ser melhorada da seguinte forma:

XADD mystream MAXLEN ~ 1000 * ... campos de entrada aqui ...
O argumento ~ no exemplo acima significa que não precisamos limitar o comprimento do stream a um valor específico. No nosso exemplo, isso pode ser qualquer número maior ou igual a 1000 (por exemplo, 1000, 1010 ou 1030). Apenas deixamos claro que queremos que nosso stream mantenha pelo menos 1000 registros. Isso torna o trabalho com a memória muito mais eficiente dentro do Redis.

Também existe um comando separado XTRIM, que executa a mesma função:

> XTRIM mystream MAXLEN 10

> XTRIM mystream MAXLEN ~ 10

Armazenamento permanente e replicação

O Redis Stream é replicado de forma assíncrona para nós escravos e é armazenado em arquivos do tipo AOF (snapshot de todos os dados) e RDB (registro de todas as operações de gravação). A replicação do estado dos Grupos de Consumidores também é suportada. Portanto, se uma mensagem está em status 'pendente' no nó mestre, essa mensagem terá o mesmo status nos nós escravos.

Remoção de elementos específicos do stream

Para remover mensagens, existe um comando especial XDEL. O comando recebe o nome do stream, seguido pelos identificadores das mensagens que precisam ser removidas:

> 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"

Ao usar esse comando, deve-se considerar que a memória não será liberada imediatamente.

Streams de comprimento zero

A diferença entre streams e outras estruturas de dados do Redis é que, quando outras estruturas não possuem mais elementos, como efeito colateral, a própria estrutura de dados é removida da memória. Por exemplo, um conjunto ordenado será completamente removido quando o comando ZREM excluir o último elemento. Em contraste, streams podem permanecer na memória mesmo sem ter nenhum elemento dentro.