Redis transmisiones, consumidores y operaciones de bloqueo

Redis introdujo las transmisiones que imitan la estructura de datos de registro con la versión 5.0. Stream es una estructura de datos solo de appendición con un conjunto de operaciones más rico que en un archivo de registro. Es uno de los tipos de datos más complejos en Redis, ya que implementa operaciones de bloqueo adicionales que permiten a los clientes esperar los nuevos datos de flujo. Esto es algo similar al comportamiento de las listas de redis pub/sub o de bloqueo, pero hay diferencias fundamentales cuando se trata de cómo los consumidores consumen los datos de la secuencia de redis.

Como se muestra en la ilustración anterior, se pueden ver varias ventajas sobre Redis Pub/Sub y Listas de bloqueo. Cada nuevo elemento de datos se entrega a cada consumidor. A diferencia de las listas que eliminan el elemento de la lista siempre que se les llame a BLPOP o BRPOP, los elementos de la transmisión permanecen como está en la transmisión. El comando Xread funciona como un candidato de bloqueo y no bloqueo en las transmisiones de Redis.

El comando xread

El comando xread puede obtener las entradas de múltiples transmisiones simultáneamente, mientras que las entradas devueltas tienen una identificación más grande que la última ID recibida para un consumidor determinado. Puede operar tanto de bloqueo como de bloqueo. En la naturaleza sin bloqueo, el comando se comporta muy similar al comando xrange pero con algunas características adicionales enumeradas a continuación:

• Puede obtener las entradas a partir de la entrada más reciente que tiene la mejor identificación que cualquier otro elemento en la transmisión.
• Puede leer de múltiples transmisiones al mismo tiempo.

Este comando tiene una complejidad de tiempo lineal cuando el número de N de los elementos se almacena en la transmisión. Por lo tanto, con un recuento de retorno fijo, la complejidad del tiempo es constante.

El comando xread sigue la siguiente sintaxis:

Sintaxis:

Xread [Count Number_of_returned_elements] [BLOCK BLOCKING_TIME_IN_MILLISECONDS] CLAVE CLAVE [CLAVE ...] ID [ID ...]

CONTAR : El número de elementos que se devolverán por el comando. Limita las filas devueltas a un número especificado.

BLOQUEAR : El tiempo máximo para esperar a que aparezca un nuevo elemento en la transmisión.

Las dos opciones anteriores son opcionales para el comando.

Transmisiones: La clave de la transmisión. Esta es una opción obligatoria y debe ser la última opción en el comando, ya que acepta la longitud variable de claves e ID de entrada.

: La identificación de la entrada de la corriente.

Se pueden especificar varias claves ya que el comando le permite leer desde más de una transmisión. Al mismo tiempo, se pueden proporcionar múltiples ID.

Este comando devuelve una respuesta de la matriz. Cada elemento de matriz consta de dos elementos como se muestra en el siguiente formato:

Ejemplo 1: Inspeccione los datos meteorológicos para dos ubicaciones con Xread sin bloqueo

Supongamos que tenemos dos transmisiones que contienen los datos meteorológicos para LA y NYC. En nuestro sitio de publicación de datos meteorológicos, necesitamos consumir de transmisiones y obtener los últimos datos meteorológicos para estas dos ubicaciones. El comando xread es el candidato ideal para usar en este escenario con su variante que no es de bloqueo.

Es hora de crear dos transmisiones nombradas Clima: NYC y Clima: LA y llenar un par de entradas con algunos pares de valor de campo como se muestra en lo siguiente:

clima xadd: NYC * Viento 45 Humedad 78 Temp 12
xadd clima: la * viento 12 humedad 45 temperatura 22

Ambas transmisiones Clima: NYC y Clima: LC se crean con éxito y las ID de entrada devueltas son 1658114094434-0 y 1658114110474-0, respectivamente.

Usemos el comando xread para leer desde ambas transmisiones al mismo tiempo de manera que no es de bloqueo.

Xread Streams Weather: NYC Weather: LA 0 0

Como se esperaba, la salida contiene las entradas de ambas transmisiones con la secuencia de ID a partir de 0. Es aceptable especificar los ID incompletos como se ilustra anteriormente donde ambas ID son 0, que es la marca de tiempo de milisegundos sin la parte del número de secuencia. Por lo tanto, el comando anterior se puede escribir como en lo siguiente:

Xread Streams Weather: NYC Weather: LA 0-0 0-0

Agreguemos un par de entradas a ambas transmisiones ahora.

Xadd Weather: NYC * Viento 10 Humedad 60 Temp 10
xadd clima: la * viento 18 humedad 80 temperatura 5

Dado que ya tenemos las últimas ID de entrada para ambas transmisiones de los comandos anteriores, volvamos a llamar al comando xread para obtener todas las entradas con identificaciones más grandes que las que ya consultamos.

Xread Streams Weather: NYC Weather: LA 1658114094434-0 1658114110474-0

Como puede ver, las ID especificadas son de la consulta anterior. Ahora, la llamada de comando devuelve todas las entradas que tienen ID de mayores que las especificadas.

Como puede ver, las entradas recién agregadas se devuelven del comando anterior. A continuación, lo que puede hacer es tomar las ID de entrada devueltas desde el comando anterior y llamar al xread con esas IDS hasta que la matriz devuelta esté vacía.

Ejemplo 2: Obtenga las últimas promociones de pizza con el bloqueo xread

Hay otra variante del comando xread que se puede usar para esperar hasta que los editores publiquen un nuevo datos en la transmisión sin terminar inmediatamente como una llamada sin bloqueo. Supongamos que un escenario en el que los chicos de las pizza quieran impulsar las notificaciones a todos los clientes con respecto a las últimas promociones disponibles. Es posible que no haya promociones en ciertos días. Por lo tanto, los clientes deben esperar hasta que las nuevas promociones estén disponibles. Se puede lograr con el comando xread con la opción de bloque en su lugar.

Supongamos que Pizza Company está publicando los detalles de la promoción a una transmisión llamada Pizzapromos: diariamente. Por lo tanto, podemos usar el comando xread para esperar hasta que se agregue un nuevo elemento promocional a la transmisión.

Xread Block 50000 Streams Pizzapromosnew: Diario $

En este caso, especificamos la ID de entrada como $ que se interpreta como la ID de entrada superior. Por lo tanto, el comando consultará solo las nuevas entradas agregadas a la transmisión y no las entradas históricas.

Dado que no hemos agregado nuevas entradas a la transmisión, se desarrollará después de 50000 milisegundos con un nulo Regrese como se muestra en lo siguiente:

Ahora, agregemos una entrada a la transmisión utilizando el XADD mientras otro consumidor está esperando los datos con el comando xread como se muestra en lo siguiente:

Como se esperaba, el consumidor consume la entrada adicional. De la siguiente llamada, debemos asegurarnos de que pasamos la identificación que se devuelve de este comando y no del $. Si no, perderemos las entradas agregadas entre.

Si varios clientes esperan la misma transmisión, los datos recién agregados se empujan a todos inmediatamente. El comando xread es un comando muy útil y recomendado para usar para bloquear las aplicaciones de la naturaleza.

Conclusión

Para resumir, el comando xread es uno de los comandos ampliamente utilizados que operan en las transmisiones Redis. Puede operar tanto en formas de bloqueo como de bloqueo. Como se discutió, la variante sin bloqueo es muy similar al comando xrange con un par de diferencias. Además, este comando se puede usar con la opción de bloque para esperar hasta que los editores publiquen un nuevo datos en la transmisión. En general, el comando xread se especializa en el consumo de datos de múltiples transmisiones simultáneamente. Es una característica útil que las aplicaciones modernas están buscando.