How to apply the stream of Redis special data type

This article mainly introduces how to apply the stream of Redis special data type. In daily operation, I believe that many people have doubts about how to apply the stream of Redis special data type. The editor consulted various materials and sorted out a simple The easy-to-use method of operation, I hope it will help you to answer your doubts about how to apply the stream of special data types in Redis! Next, please follow the editor to learn together!

Redis Stream is a newly added data type in Redis version 5.0. Redis is a data type specially designed for message queues.

Before Redis 5.0 Stream came out, the implementation methods of message queues had their own defects, for example:

• The publish-subscribe model cannot be persisted Unable to reliably save messages, and the defect that historical messages cannot be read for offline reconnected clients;

• The way List implements message queues cannot be consumed repeatedly, and one message is consumed will be deleted, and the producer needs to implement a globally unique ID by itself.

Based on the above problems, Redis 5.0 launched the Stream type, which is also the most important function of this version. It is used to perfectly implement message queues. It supports message persistence, support Automatically generate a globally unique ID, support the ack confirmation message mode, support the consumer group mode, etc., making the message queue more stable and reliable.

Common commands

Stream message queue operation command:

• XADD: Insert messages to ensure order , can automatically generate a globally unique ID

• XDEL: delete the message according to the message ID;

• DEL: delete the entire Stream;

# XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|id field value [field value ...]
127.0.0.1:6379> XADD s1 * name sid10t
"1665047636078-0"
127.0.0.1:6379> XADD s1 * name sidiot
"1665047646214-0"
# XDEL key id [id ...]
127.0.0.1:6379> XDEL s1 1665047646214-0
(integer) 1
# DEL key [key ...]
127.0.0.1:6379> DEL s1
(integer) 1

• XLEN : query message length;

• XREAD: Used to read messages, you can read data by ID;

• XRANGE: Read interval messages;

• XTRIM : the number of trimming queue messages;

# XLEN key
127.0.0.1:6379> XLEN s1
(integer) 2
# XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
127.0.0.1:6379> XREAD streams s1 0-0
1) 1) "s1"
   2) 1) 1) "1665047636078-0"
         2) 1) "name"
            2) "sid10t"
      2) 1) "1665047646214-0"
         2) 1) "name"
            2) "sidiot"
127.0.0.1:6379> XREAD count 1 streams s1 0-0
1) 1) "s1"
   2) 1) 1) "1665047636078-0"
         2) 1) "name"
            2) "sid10t"
    # XADD 
    127.0.0.1:6379> XREAD streams s1 1665047636078-0
    1) 1) "s1"
       2) 1) 1) "1665047646214-0"
             2) 1) "name"
                2) "sidiot"
          2) 1) "1665053702766-0"
             2) 1) "age"
                2) "18"
# XRANGE key start end [COUNT count]
127.0.0.1:6379> XRANGE s1 - +
1) 1) "1665047636078-0"
   2) 1) "name"
      2) "sid10t"
2) 1) "1665047646214-0"
   2) 1) "name"
      2) "sidiot"
3) 1) "1665053702766-0"
   2) 1) "age"
      2) "18"
127.0.0.1:6379> XRANGE s1 1665047636078-0 1665047646214-0
1) 1) "1665047636078-0"
   2) 1) "name"
      2) "sid10t"
2) 1) "1665047646214-0"
   2) 1) "name"
      2) "sidiot"
# XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]
127.0.0.1:6379> XLEN s1
(integer) 3
127.0.0.1:6379> XTRIM s1 maxlen 2
(integer) 1
127.0.0.1:6379> XLEN s1
(integer) 2

• XGROUP CREATE : Create a consumer group;

• XREADGROUP : Press Read messages in consumer group form;

• XPENDING and XACK:

XPENDING command can be used to query each consumer group All consumers in the "read, but not yet acknowledged" message;

XACK command is used to confirm to the message queue that the message processing has been completed;