Shulou(Shulou.com)06/02 Report--

This article mainly introduces how to use the Stream type in Redis. It is very detailed and has a certain reference value. Friends who are interested must finish it!

I. background

Recently, after looking at the knowledge of redis, I found that a new data type, Stream, has emerged in redis5, which is somewhat similar to the design of kafka and can be used as a simple message queue.

II. The characteristics of Stream types in redis

It is persistent and can guarantee that the data is not lost.

Support message multicasting and packet consumption.

Support the ordering of messages.

III. The structure of Stream

Explanation:

Consumer groups: Consumer Group, even if created with the XGROUP CREATE command, there can be multiple consumers in a consumer group that compete with each other.

The same message can only be obtained by a consumer in this consumer group.

Multiple consumers are independent of each other and do not interfere with each other.

Consumer: Consumer consumption message.

Last_delivered_id: this id ensures that a message can only be obtained by one consumer in the same consumer group. Every time a consumer in the consumer group reads the message, the value of the last_delivered_id moves back one bit to ensure that the consumer does not read the duplicate message.

Pending_ids: the id list of messages read by the consumer is recorded, but these messages may not have been processed, so if you think a message is being processed, you need to call the ack command. This ensures that a message will be executed once.

Message content: is the format of a key-value pair.

ID of messages in Stream: by default, ID uses *, and redis can automatically generate one in the format of timestamp-sequence number, or you can specify it yourself. Generally, you can use the default generated id, and the generated id number is larger than the previous generated one.

4. Stream command 1. XADD adds message to the end of Stream. Command format: xadd key [NOMKSTREAM] [MAXLEN | MINID [= | ~] threshold [LIMIT count]] * | ID field value [field value.]

2. Examples

The xadd command returns the id of the data, xx-yy (xx refers to the number of milliseconds, and yy refers to the number of messages in this millisecond)

1. Add a piece of data to the stream

127.0.0.1 username 6379 > xadd stream-key * username zhangsan # add a data in which username is zhangsan to the stream stream-key * indicates that the automatic generation of id "1635999858912-0" # returns ID127.0.0.1:6379 > keys * 1) "stream-key" # you can see that stream automatically creates 127.0.0.16379 >

2. Add data to the stream without automatically creating the stream

127.0.0.1 nomkstream 6379 > xadd not-exists-stream nomkstream * username lisi # failed to join (nil) 127.0.0.1 not-exists-stream 6379 > keys * (empty array) 127.0.0.1 not-exists-stream 6379 > because the nomkstream parameter was specified and did not exist before

3. Specify the value of ID manually

127.0.0.1 username lisi 6379 > xadd stream-key 1-1 username lisi # where the value of id is the 1-1 passed by yourself, instead of using * automatically generate "1-1" # the value returned is 127.0.0.1 id 6379 >

4. Set a fixed size Stream1 and specify the exact size of the Stream.

Specifying the size of the specified Stream consumes a little more performance than blurring the size of the specified Stream.

2. Blur the size of the specified Stream

127.0.0.1xadd stream-key maxlen 6379 > xadd stream-key maxlen ~ 1 * first first "1636001034141-0" 127.0.0.1xadd stream-key maxlen ~ 1 * second second "1636001044506-0" 127.0.0.1xadd stream-key maxlen ~ 1 * third third "1636001057846-0" 127.0.16379 > xinfo stream stream-key 1) "length" 2) (integer) 3 3) "radix-tree-keys" 4) (integer) 1 5) " Radix-tree-nodes "6) (integer) 27)" last-generated-id "8)" 1636001057846-0 "9)" groups "10) (integer) 011)" first-entry "12) 1)" 1636001034141-0 "2)" first "2)" first "13)" last-entry "14) 1)" 1636001057846-0 "2)" third "2)" third "127.0.0.1 third 6379 >

Blurring the size of the specified stream, you can see that the specified size is 1, which is actually up to 3.

2. XRANGE views messages in Stream 1. Command format xrange key start end [COUNT count]

2. Prepare data 127.0.0.1xadd stream-key 6379 > multiOK127.0.0.1:6379 (TX) > xadd stream-key * username zhangsanQUEUED127.0.0.1:6379 (TX) > xadd stream-key * username lisiQUEUED127.0.0.1:6379 (TX) > exec1) "1636003481706-0" 2) "1636003481706-1" 127.0.0.1xadd stream-key 6379 > xadd stream-key * username wangwu "1636003499055-0" 127.0.1Ze6379 >

Use the transaction operation of redis to obtain multiple pieces of data generated in the same millisecond with the same timestamp and different sequence numbers

3. Give an example

1. Get all the data (use of-and +)

127.0.0.1 xrange stream-key 6379 > lisi-+ 1) 1) "1636003481706-0" 2) 1) "username" 2) "zhangsan" 2) 1) "1636003481706-1" 2) "username" 2) "lisi" 3) 1) "1636003499055-0" 2) "username" 2) "wangwu" 127.0.0.1

-: the value that represents the minimum id

+: the value that represents the maximum id

2. Get the data within the specified id range and close the interval.

127.0.0.1 xrange stream-key 1636003481706-1 1636003499055-01) 1) "1636003481706-1" 2) "username" 2) "lisi" 2) 1) "1636003499055-0" 2) 1) "username" 2) "wangwu" 127.0.0.1

3. Get the data within the specified id range and open the interval

127.0.0.1 xrange stream-key 6379 > lisi (1636003481706-0 (1636003499055-01) 1) "1636003481706-1" 2) "username" 2) "lisi" 127.0.0.1

(: indicates an open interval

4. Get all the data after a certain millisecond

127.0.0.1 xrange stream-key 1636003481706 + 1) 1) "1636003481706-0" 2) 1) "username" 2) "zhangsan" 2) 1) "1636003481706-1" 2) "username" 2) "lisi" 3) 1) "1636003499055-0" 2) "username" 2) "wangwu" 127.0.0.1 zhangsan 6379 >

Just write in milliseconds without writing the following serial number.

5. Obtain single piece of data

127.0.0.1 xrange stream-key 1636003499055-01636003499055-01) 1) "1636003499055-0" 2) "username" 2) "wangwu" 127.0.0.16379 >

Start and end are written with the same value to obtain one-on-one data.

6. Obtain data with a fixed number of entries

127.0.0.1 username 6379 > xrange stream-key-+ count 11) 1) "1636003481706-0" 2) 1) "username" 2) "zhangsan" 127.0.0.1

Use count to restrict

3. XREVRANGE checks the message XREVRANGE key end start [COUNT count] in Stream in reverse.

The way of use is similar to that of XRANGE.

4. XDEL delete message 1, command format xdel key ID [ID...] 2, prepare data 127.0.0.1ID 6379 > xadd stream-key * username zhangsan "1636004176924-0" 127.0.0.1username zhangsan 6379 > xadd stream-key * username lisi "1636004183638-0" 127.0.0.1username zhangsan 6379 > xadd stream-key * username wangwu "1636004189211-0" 127.0.1ID 6379 > 3, for example

Requirements: add 3 messages to the Stream, then delete the second message

127.0.0.1 xdel stream-key 6379 > 1636004183638-0 (integer) 1 # returns the number of deleted records 127.0.0.1xrang stream-key-+ 127.0.0.1 xrang stream-key > xrange stream-key-+ 1) 1) "1636004176924-0" 2) "username" 2) "zhangsan" 2) 1) "1636004189211-0" 2) 1) "username" 2) "wangwu" 127.0.0.1lane 6379 >

Note:

It is important to note that when we delete a message from Stream, the message is not actually deleted, but marked for deletion, at which time the message still occupies content space. Memory space will be reclaimed only if all messages in all Stream are marked to be deleted. But this Stream will not be deleted.

5. XLEN to view the length of elements in Stream 1. Command format xlen key2, for example

View the length of the element in the Stream

127.0.1 xadd stream-key > xadd stream-key * username zhangsan "1636004690578-0" 127.0.0.1 > xlen stream-key (integer) 1127.0.1 > xlen not-exists-stream-key (integer) 0127.0.0.1 >

Note:

Returns 0 if the key behind xlen does not exist, otherwise the number of elements is returned.

6. XTRIM prunes the elements in Stream 1. Command format xtrim key MAXLEN | MINID [= | ~] threshold [LIMIT count] 2, Prepare data 127.0.0.1xadd stream-key 6379 > xadd stream-key * username zhangsan "1636009745401-0" 127.0.0.1multiOK127.0.0.1:6379 (TX) > xadd stream-key * username lisiQUEUED127.0.0.1:6379 (TX) > xadd stream-key * username wangwuQUEUED127.0.0.1:6379 (TX) > exec1) "1636009763955-0" 2) "1636009763955-1" 127.0.1multiOK127.0.0.1:6379 6379 > xadd stream-key * username zhaoliu "1636009769625-0" 127.0.1TX > 3. Give an example

1. Maxlen precise limit

127.0.0.1 xtrim stream-key maxlen 2 # keep the last two messages (integer) 2127.0.0.1 xrange stream-key-+ # you can see that the two messages previously added have been deleted 1) 1) "1636009763955-1" 2) "username" 2) "wangwu" 2) 1) "username" 2) "username" 2) "zhaoliu" 127.0.1

The above means to keep the last two messages in the stream-key Stream.

2. Minid fuzzy limit

Minid is to delete data smaller than this id, which is not tested during the local test.

7. XREAD independent consumption message

XREAD just reads the message and does not delete the message after reading it. Using XREAD to read messages is completely independent of consumer groups, and multiple clients can read messages at the same time.

1. Command format xread [COUNT count] [BLOCK milliseconds] STREAMS key [key.] ID [ID...]

2. Prepare data 127.0.0.1xadd stream-key 6379 > xadd stream-key * username zhangsan "1636011801365-0" 127.0.0.1xadd stream-key * username lisi "1636011806261-0" 127.0.0.16379 > xadd stream-key * username wangwu "1636011810905-0" 127.0.0.16379 > 3.

1. Get the data whose user name is wangwu

127.0.0.1stream-key 6379 > xread streams stream-key 1636011806261-0 # what is written here is the id of lisi, that is, the data read should be > 1636011806261-01) 1) "stream-key" 2) 1) 1) "1636011810905-0" 2) "username" 2) "wangwu"

2. Get 2 pieces of data

127.0.0.1 streams stream-key 0-01) 1) "stream-key" 2) 1) 1) "1636011801365-0" 2) 1) "username" 2) "zhangsan" 2) 1) "1636011806261-0" 2) 1) "username" 2) "lisi" 127.0.0.1 username 6379 >

Count restricts reading the last message at a time, because there may not be so many current reads.

3. Non-blocking reading of Stream tail data

That is, the next message at the end of the queue is always nil in non-blocking mode.

127.0.0.1 6379 > xread streams stream-key $(nil)

4. Block reading the data of Stream tail

Note:

$means to read the latest message in the queue, not the last message from Stream. After the xread block is executed, the xread block will not return until the message is added using xadd again.

Block 0 indicates permanent blocking, and the blocking is contacted only when the message arrives. Block 1000 indicates blocking 1000ms. If no message has arrived from 1000ms, nil is returned.

Xread for sequential consumption when using xread for sequential messages, you need to remember the returned message id, and the next time you call xread, you need to pass in the last returned message id.

Xread reads the message, completely ignoring the consumer group, and Stream can be understood as a normal list.

8. Consumer group related operations 1. Consumer group commands

2. Prepare data

1. The name of creating Stream is stream-key.

2. Create 2 messages, aa and bb

127.0.0.1xadd stream-key 6379 > xadd stream-key * aa aa "1636362619125-0" 127.0.0.1 bb bb "1636362623191-0" 3, create a consumer group

1. Create a consumer group that consumes from scratch

Xgroup create stream-key (Stream name) G1 (consumer group name) 0-0 (indicates consumption from scratch)

2. Create a consumer group that consumes from the latest message from Stream

Xgroup create stream-key G2 $

$means consuming from the last element, excluding the last element in Stream, that is, consuming the latest messages.

4. Create a consumer group xgroup create stream-key G3 1636362619125-0 # 1636362619125-0 that is consumed after a message. This is the value of the id of the aa message above.

1636362619125-0 the specific ID of a message. The messages in this G3 consumer group are all messages larger than the id.

3. Read messages from consumers

127.0.0.1 xreadgroup group 6379 > Consumer Group name C1 (Consumer name) Automatically create) count 3 (read 3) streams stream-key (Stream name) > (read from messages in this consumer group that have not been assigned to another consumer) 1) 1) "stream-key" 2) 1) 1) "aa" 2) "aa" 2) "aa" 2) 1) "bb" "2)" bb "127.0.0.1 count 6379 > xreadgroup group G2 C1 count 3 streams stream-key > (nil) # returns nil because the G2 consumer group is read from the latest message ($was used when creating the consumer group) You need to execute the `xadd` command in another window to read the message 127.0.0.1 count 6379 > xreadgroup group g3 C1 count 3 streams stream-key > # only one message is read because the id of the id,bb message specified by the aa message is greater than aa when creating the consumer group, so it is read out. 1) 1) "stream-key" 2) 1) "1636362623191-0" 2) 1) "bb" 2) "bb" 127.0.0.1 bb 6379 >

4. Read the consumer's pending message

127.0.0.1 xgroup create stream-key g40-0OK127.0.0.1:6379 > xinfo consumers stream-key g11) 1) "name" 2) "C1" 3) "pending" 4) (integer) 25) "idle" 6) (integer) 88792127.0.1pending 6379 > xinfo consumers stream-key g4 (empty array) 127.0.0.1) xreadgroup group g1 C1 count 1 streams stream-key 1636362619125-01) 1) "stream-key" 2) 1) 1) "1636362623191-0" 2) "bb" 2) "bb" 127.0.0.1 count 6379 > xreadgroup group g4 C1 count 1 block 0 streams stream-key 1636362619125-01) 1) "stream-key" 2) (empty array) 127.0.0.1 count 6379 >

5. Transfer the news of consumers

127.0.0.1 xpending stream-key g1-+ 10 c11) 1) "1636362619125-0" 2) "C1" 3) (integer) 2686183 4) (integer) 12) 1) "1636362623191-0" 2) "C1" 3) (integer) 102274) (integer) 7127.0.1 1R 6379 > xpending stream-key g 1-+ 10 c2 (empty array) 127.0.1R 6379 > xclaim stream-key g1 c2 102274 16362623191 -01) 1) "1636362623191-0" 1) "bb" 2) "bb" 127.0.0.1 xpending stream-key 6379 > xpending stream-key g1-+ 10 c21) 1) "1636362623191-0" 2) "c2" 3) (integer) 17664) (integer) 8127.0.0.1 >

It can also be achieved through xautoclaim.

6. Some monitoring commands

1. View the pending messages of consumers in the consumption group

127.0.0.1) xpending stream-key G1-+ 10 c21) 1) "1636362623191-0" 2) "c2" 3) (integer) 1247680 4) (integer) 8127.0.1

2. View the consumer information in the consumption group

127.0.0.1 xinfo consumers stream-key G11) 1) "name" 2) "C1" 3) "pending" 4) (integer) 1 5) "idle" 6) (integer) 14748642) 1) "name" 2) "c2" 3) "pending" 4) (integer) 1 5) "idle" 6) (integer) 1290069127.0.1

3. View consumption group information

127.0.1 consumers 6379 > xinfo groups stream-key1) 1) "name" 2) "G1" 3) "consumers" 4) (integer) 2 5) "pending" 6) (integer) 27) "last-delivered-id" 8) "1636362623191-0" 2) 1) "name" 2) "G2" 3) "consumers".

4. View Stream information

127.0.0.1 xinfo stream stream-key 6379 > first-entry 1) "length" 2) (integer) 23) "radix-tree-keys" 4) (integer) 1 5) "radix-tree-nodes" 6) (integer) 27) "last-generated-id" 8) "1636362623191-0" 9) "groups" 10) (integer) 411) "first-entry" 12) 1) "1636362619125-0" 2) 1) "aa" 2) "aa" 13) "last-entry" 14) 1) "1636362623191-0" 2) "bb" 2) "bb" 127.0.1 bb 6379 > above are all the contents of the article "how to use Stream types in Redis" Thank you for reading! Hope to share the content to help you, more related knowledge, welcome to follow the industry information channel!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.