Redis Streams, consumidores e operações de bloqueio

Redis introduziu os fluxos que imitam a estrutura de dados do log com a versão 5.0. Stream é uma estrutura de dados somente de anexo com um conjunto de operações mais rico do que em um arquivo de log. É um dos tipos de dados mais complexos do Redis, pois implementa operações adicionais de bloqueio que permitem aos clientes esperar os novos dados do fluxo. Isso é um pouco semelhante ao comportamento de listas de pub/sub -bloqueio Redis, mas as diferenças fundamentais existem quando se trata de como os consumidores consomem os dados do Redis Stream.

Conforme mostrado na ilustração anterior, várias vantagens sobre o pub/sub e o bloqueio Redis podem ser vistas. Cada novo item de dados é entregue a todos os consumidores. Ao contrário das listas que removem o item da lista sempre que chamado para BLPOP ou BRPOP, os itens do fluxo permanecem como está no fluxo. O comando xread opera como um candidato bloqueador e não bloqueador nos riachos Redis.

O comando xread

O comando xread pode buscar as entradas de vários fluxos simultaneamente enquanto as entradas retornadas têm um ID maior que o último ID recebido para um determinado consumidor. Pode operar da maneira bloqueadora e não bloqueadora. Na natureza não bloqueadora, o comando se comporta muito semelhante ao comando xrange, mas com alguns recursos adicionais listados no seguinte:

• Ele pode buscar as entradas a partir da entrada mais recente que tem a maior identificação do que qualquer outro item no fluxo.
• Pode ler em vários fluxos ao mesmo tempo.

Este comando tem uma complexidade do tempo linear quando o N número de elementos é armazenado no fluxo. Portanto, com uma contagem de retorno fixo, a complexidade do tempo é constante.

O comando XRead segue a seguinte sintaxe:

Sintaxe:

XRead [contagem number_of_Returned_elements] [block Blocking_time_in_milliseconds] Streams Key [key…] id [id…]

CONTAR : O número de elementos a serem devolvidos pelo comando. Limita as linhas retornadas a um número especificado.

BLOQUEAR : O tempo máximo para esperar um novo item aparecer no fluxo.

As duas opções anteriores são opcionais para o comando.

Fluxos: A chave do fluxo. Esta é uma opção obrigatória e deve ser a última opção no comando, pois aceita o comprimento variável das chaves e os IDs de entrada.

: O ID da entrada do fluxo.

Várias chaves podem ser especificadas, pois o comando permite que você leia mais de um fluxo. Ao mesmo tempo, vários IDs podem ser fornecidos.

Este comando retorna uma resposta de matriz. Cada item da matriz consiste em dois elementos, como mostrado no seguinte formato:

Exemplo 1: Inspecione os dados climáticos para dois locais com xread não bloqueador

Vamos supor que temos dois fluxos contendo os dados climáticos para LA e NYC. Em nosso site de publicação de dados meteorológicos, precisamos consumir de fluxos e buscar os dados climáticos mais recentes para esses dois locais. O comando xread é o candidato ideal para usar neste cenário com sua variante não bloqueadora.

É hora de criar dois fluxos nomeados Clima: Nova York e Clima: LA e preencha algumas entradas com alguns pares de valor de campo, como mostrado no seguinte:

Xadd Weather: NYC * vento 45 umidade 78 Temp 12
Xadd Weather: la * vento 12 umidade 45 Temp 22

Ambos os fluxos Clima: Nova York e Clima: LC são criados com sucesso e os IDs de entrada retornados são 1658114094434-0 e 1658114110474-0, respectivamente.

Vamos usar o comando xread para ler de ambos os fluxos ao mesmo tempo de maneira não bloqueada.

XRead Streams Clima: NYC Clima: la 0 0

Como esperado, a saída contém as entradas de ambos os fluxos com a sequência de identificação a partir de 0. É aceitável especificar os IDs incompletos, como ilustrado anteriormente, onde os dois IDs são 0, que é o registro de data e hora de milissegundos sem a parte da sequência do número. Portanto, o comando anterior pode ser escrito como no seguinte:

XRead Streams Weather: NYC Clima: LA 0-0 0-0

Vamos adicionar algumas entradas aos dois riachos agora.

Xadd Weather: NYC * vento 10 umidade 60 Temp 10
Xadd Weather: la * vento 18 umidade 80 Temp 5

Como já temos os IDs de entrada mais recentes para ambos os fluxos dos comandos anteriores, vamos chamar o comando xread novamente para buscar todas as entradas com IDs maiores do que as que já consultamos.

XRead Streams Weather: NYC Clima: LA 1658114094434-0 1658114110474-0

Como você pode ver, os IDs especificados são da consulta anterior. Agora, a chamada de comando retorna todas as entradas que têm IDs maiores do que as especificadas.

Como você pode ver, as entradas recém -adicionadas são devolvidas do comando anterior. Em seguida, o que você pode fazer é pegar os IDs de inscrição retornados do comando anterior e chamar o XRead com esses IDs até que a matriz retornada esteja vazia.

Exemplo 2: Obtenha as últimas promoções de pizza com o bloqueio do XRead

Há outra variante do comando xread que pode ser usado para esperar até que os editores publiquem um novo dado para o fluxo sem terminar imediatamente como uma chamada não bloqueadora. Vamos supor um cenário em que os caras da pizza querem empurrar as notificações a todos os clientes em relação às promoções mais recentes disponíveis. Pode não haver promoções em certos dias. Portanto, os clientes devem esperar até que as novas promoções estejam disponíveis. Pode ser alcançado com o comando xread com a opção de bloco no lugar.

Vamos supor que a empresa de pizza está publicando os detalhes da promoção para um fluxo chamado Pizzapromos: diariamente. Portanto, podemos usar o comando xread para esperar até que um novo item promocional seja adicionado ao fluxo.

XRead Block 50000 Streams Pizzapromosnew: diariamente $ $

Nesse caso, especificamos o ID de entrada como $, que é interpretado como o ID de entrada superior. Portanto, o comando consultará apenas as novas entradas adicionadas ao fluxo e não às entradas históricas.

Como não adicionamos novas entradas ao fluxo, ele terá tempo limite após 50000 milissegundos com um nada Retorne conforme mostrado no seguinte:

Agora, vamos adicionar uma entrada ao fluxo usando o XADD, enquanto outro consumidor está aguardando os dados com o comando xread, conforme mostrado no seguinte:

Como esperado, a entrada adicionada é consumida pelo consumidor imediatamente. A partir da próxima chamada, precisamos ter certeza de que passamos o ID que é devolvido deste comando e não o $. Caso contrário, sentiremos falta das entradas adicionadas entre.

Se vários clientes estiverem aguardando o mesmo fluxo, os dados recém -adicionados serão empurrados para todos eles imediatamente. O comando xread é um comando muito útil e recomendado para usar no bloqueio dos aplicativos da natureza.

Conclusão

Para resumir, o comando xread é um dos comandos amplamente utilizados que operam nos riachos Redis. Pode operar de maneiras bloqueador e não bloqueador. Conforme discutido, a variante não bloqueadora é muito semelhante ao comando xRange com algumas diferenças. Além disso, este comando pode ser usado com a opção de bloco para esperar até que os editores publiquem um novo dados no fluxo. No geral, o comando xread é especializado em consumir os dados de vários fluxos simultaneamente. É uma característica útil que as aplicações modernas estão procurando.