Sudis

# Stream命令

# XACK

# 语法

XACK key group id [id ...]

# 描述

XACK 命令用于从流（Stream）消费者组的待处理条目列表（PEL）中，移除一条或多条消息

# 参数

  • key

Stream 键

  • group

消费者组名称

  • id

需要确认（ACK）的一条或多条消息 ID

# 示例

> XACK mystream mygroup 1526569495631-0
(integer) 1

# XADD

# 语法

XADD key [NOMKSTREAM] [KEEPREF | DELREF | ACKED]
  [IDMPAUTO producer-id | IDMP producer-id idempotent-id]
  [<MAXLEN | MINID> [= | ~] threshold [LIMIT count]] <* | id>
  field value [field value ...]

# 描述

将指定的流条目追加到指定键的流中。如果该键不存在，XADD 会在执行命令时顺带创建一个新键，并写入这条流数据。如果你不想让它自动建键，可以加上 NOMKSTREAM 选项来关闭这个行为

# 参数

  • key

Stream 键

  • id

Stream条目的 ID。使用 * 可自动生成唯一 ID

  • field value

构成流条目（stream entry）的一个或多个字段-值对。您必须至少提供一个字段-值对

# 示例

> XADD mystream * name Sara surname OConnor
4378417975-0"
> XADD mystream * field1 value1 field2 value2 field3 value3
4378417976-0"
> XLEN mystream
eger) 2
> XRANGE mystream - +
1) 1) "1774378417975-0"
   2) 1) "name"
      2) "Sara"
      3) "surname"
      4) "OConnor"
2) 1) "1774378417976-0"
   2) 1) "field1"
      2) "value1"
      3) "field2"
      4) "value2"
      5) "field3"
      6) "value3"

> XADD mystream IDMP producer1 msg1 * field value
"1774378417976-0"
> XADD mystream IDMP producer1 msg1 * field different_value
"1774378417976-0"
> XADD mystream IDMPAUTO producer2 * field value
"1774378417977-0"
> XADD mystream IDMPAUTO producer2 * field value
"1774378417977-0"
> XCFGSET mystream IDMP-DURATION 300 IDMP-MAXSIZE 1000
"OK"

# XAUTOCLAIM

# 语法

XAUTOCLAIM key group consumer min-idle-time start [COUNT count]
  [JUSTID]

# 描述

该命令会将符合指定条件的、处于待处理状态的流条目转移所有权。从概念上讲，XAUTOCLAIM 相当于先调用 XPENDING 再调用 XCLAIM，但它提供了类似 SCAN 的语义，让处理消息投递失败的情况变得更加简单直接

# 参数

  • key

Stream 键

  • group

消费者组名称

  • consumer

将拥有已认领消息的消费者的名称

  • min-idle-time

仅认领空闲时间至少达到此时长（单位：毫秒）的消息。

  • start

开始扫描的起始消息 ID。传入 0 表示从 PEL 的开头开始

# 示例

> XAUTOCLAIM mystream mygroup Alice 3600000 0-0 COUNT 25
1) "0-0"
2) 1) 1) "1609338752495-0"
      2) 1) "field"
         2) "value"
3) (empty array)

# XCLAIM

# 语法

XCLAIM key group consumer min-idle-time id [id ...] [IDLE ms]
  [TIME unix-time-milliseconds] [RETRYCOUNT count] [FORCE] [JUSTID]
  [LASTID lastid]

# 描述

将存储在 key 的哈希中 field 字段的整数值增加 increment。如果 key 不存在，则创建一个保存哈希的新键。如果 field 不存在，则在执行操作前将其设置为 0。支持的整数范围限于 64 位有符号整数。

# 参数

  • key

保存stream的键名。

  • group

消费者组名称

  • consumer

将拥有已认领消息的消费者的名称

  • id

一个或多个待认领（/待确认/待处理）的消息 ID

# 示例

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"

# XDEL

# 语法

XDEL key id [id ...]

# 描述

从stream中移除指定的条目，并返回被删除的条目数量。如果指定的某些 ID 在流中并不存在，那么返回的删除数量可能会少于传入命令的 ID 数量

# 参数

  • key

Stream键的名称。

  • id

一个或多个待认领（/待确认/待处理）的消息 ID

# 示例

> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(integer) 1
127.0.0.1:6379> XRANGE mystream - +
1) 1) 1538561698944-0
   2) 1) "a"
      2) "1"
2) 1) 1538561701744-0
   2) 1) "c"
      2) "3"

# XGROUP CREATE

# 语法

XGROUP CREATE key group <id | $> [MKSTREAM]
  [ENTRIESREAD entries-read]

# 描述

为 key 对应的流创建一个新的消费者组，使用 groupname 作为唯一标识

# 参数

  • key

    存储stream的键名。

  • group

消费者组名称

  • id | $

该组在读取时，从这条已投递消息的 ID 之后开始。如果想从流中当前的最后一条消息开始，可以使用 $

# XGROUP CREATECONSUMER

# 语法

XGROUP CREATECONSUMER key group consumer

# 描述

在存储于指定键的流中，于消费者组 groupname 下创建一个名为 consumername 的消费者。

# 参数

  • key

Stream键的名称。该流必须已经存在

  • group

消费者组名称

  • consumer

将拥有已认领消息的消费者的名称

# XGROUP DELCONSUMER

# 语法

XGROUP DELCONSUMER key group consumer

# 描述

XGROUP DELCONSUMER 命令用于从消费者组中删除一个消费者

# 参数

  • key

Stream键的名称。该流必须已经存在

  • group

消费者组名称

  • consumer

将拥有已认领消息的消费者的名称

# XGROUP DESTROY

# 语法

XGROUP DESTROY key group

# 描述

XGROUP DESTROY 命令会彻底销毁一个消费者组。

# 参数

  • key

Stream键的名称。该流必须已经存在

  • group

消费者组名称

# XGROUP SETID

# 语法

XGROUP SETID key group <id | $> [ENTRIESREAD entries-read]

# 描述

为消费组设置最后已投递的 ID

# 参数

  • key

Stream键的名称。该流必须已经存在

  • group

消费者组名称

  • id | $

用于设置为该群组最后已投递的 ID。使用 $ 代表当前stream中的最后一条消息，或使用 0 重置到起始位置

# XINFO CONSUMERS

# 语法

XINFO CONSUMERS key group

# 描述

该命令会返回指定 key 下，属于 groupname 消费者组的所有消费者列表

# 参数

  • key

Stream键的名称。该流必须已经存在

  • group

消费者组名称

# 示例

> XINFO CONSUMERS mystream mygroup
1) 1) name
   2) "Alice"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 9104628
   7) inactive
   8) (integer) 18104698
2) 1) name
   2) "Bob"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 83841983
   7) inactive
   8) (integer) 993841998

# XINFO GROUP

# 语法

XINFO GROUPS key

# 描述

该命令会返回存储在指定 key 中的stream的所有消费者组列表

# 参数

  • key

Stream键的名称

# 示例

> XINFO GROUPS mystream
1)  1) "name"
    2) "mygroup"
    3) "consumers"
    4) (integer) 2
    5) "pending"
    6) (integer) 2
    7) "last-delivered-id"
    8) "1638126030001-0"
    9) "entries-read"
   10) (integer) 2
   11) "lag"
   12) (integer) 0
2)  1) "name"
    2) "some-other-group"
    3) "consumers"
    4) (integer) 1
    5) "pending"
    6) (integer) 0
    7) "last-delivered-id"
    8) "1638126028070-0"
    9) "entries-read"
   10) (integer) 1
   11) "lag"
   12) (integer) 1

# XINFO STREAM

# 语法

XINFO STREAM key [FULL [COUNT count]]

# 描述

该命令会返回存储在指定 key中stream的相关信息

# 参数

  • key

    stream键的名称

  • FULL

开始扫描的起始消息 ID。传入 0 表示从 PEL 的开头开始

  • COUNT

指定要返回的条目数量。设为 0 会返回所有条目；默认值为 10

# 示例

> XADD mystream * message apple
"1638125133432-0"
> XADD mystream * message banana
"1638125141232-0"
> XCFGSET mystream DURATION 100 MAXSIZE 100
OK
> XADD mystream IDMP producer1 msg1 * field value
"1638125150000-0"

> XINFO STREAM mystream
 1) "length"
 2) (integer) 2
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "last-generated-id"
 8) "1638125141232-0"
 9) "max-deleted-entry-id"
10) "0-0"
11) "entries-added"
12) (integer) 2
13) "recorded-first-entry-id"
14) "1719505260513-0"
15) "idmp-duration"
16) (integer) 100
17) "idmp-maxsize"
18) (integer) 100
19) "pids-tracked"
20) (integer) 1
21) "iids-tracked"
22) (integer) 1
23) "iids-added"
24) (integer) 1
25) "iids-duplicates"
26) (integer) 0
27) "groups"
28) (integer) 1
29) "first-entry"
30) 1) "1638125133432-0"
    2) 1) "message"
       2) "apple"
31) "last-entry"
32) 1) "1638125141232-0"
    2) 1) "message"
       2) "banana"

> XADD mystream * foo bar
"1638125133432-0"
> XADD mystream * foo bar2
"1638125141232-0"
> XGROUP CREATE mystream mygroup 0-0
OK
> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) "1638125133432-0"
         2) 1) "foo"
            2) "bar"
> XINFO STREAM mystream FULL
 1) "length"
 2) (integer) 2
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "last-generated-id"
 8) "1638125141232-0"
 9) "max-deleted-entry-id"
10) "0-0"
11) "entries-added"
12) (integer) 2
13) "recorded-first-entry-id"
14) "1719505260513-0"
15) "entries"
16) 1) 1) "1638125133432-0"
       2) 1) "foo"
          2) "bar"
    2) 1) "1638125141232-0"
       2) 1) "foo"
          2) "bar2"
17) "groups"
18) 1)  1) "name"
        2) "mygroup"
        3) "last-delivered-id"
        4) "1638125133432-0"
        5) "entries-read"
        6) (integer) 1
        7) "lag"
        8) (integer) 1
        9) "pel-count"
       10) (integer) 1
       11) "pending"
       12) 1) 1) "1638125133432-0"
              2) "Alice"
              3) (integer) 1638125153423
              4) (integer) 1
       13) "consumers"
       14) 1) 1) "name"
              2) "Alice"
              3) "seen-time"
              4) (integer) 1638125133422
              5) "active-time"
              6) (integer) 1638125133432
              7) "pel-count"
              8) (integer) 1
              9) "pending"
              10) 1) 1) "1638125133432-0"
                     2) (integer) 1638125133432
                     3) (integer) 1

# XINFO

# 语法

XINFO

# 描述

这是一个用于流内省命令的容器命令

# XLEN

# 语法

XLEN key

# 描述

返回流（Stream）内部的条目数量。如果指定的key不存在，命令会返回零，因此，如果想确认一个键到底存不存在，应该使用 TYPE 或 EXISTS 命令来进行检查。

# 参数

  • key

Stream键的名称。

# 示例

> XADD mystream * item 1
"1782978888016-0"
> XADD mystream * item 2
"1782978888017-0"
> XADD mystream * item 3
"1782978888018-0"
> XLEN mystream
(integer) 3

# XPENDING

# 语法

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

# 描述

XPENDING 命令用于查看待处理消息的列表。因此，它是观察和了解Stream消费者组运行状态的一个非常重要的命令，比如：哪些客户端处于活跃状态、哪些消息正在等待被消费

# 参数

  • key

Stream键的名称

  • group

消费者组名称

  • consumer

将拥有已认领消息的消费者的名称

# 示例

> XGROUP CREATE mystream group55 0-0
OK

> XREADGROUP GROUP group55 consumer-123 COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526984818136-0
         2) 1) "duration"
            2) "1532"
            3) "event-id"
            4) "5"
            5) "user-id"
            6) "7782813"

> XPENDING mystream group55
1) (integer) 1
2) 1526984818136-0
3) 1526984818136-0
4) 1) 1) "consumer-123"
      2) "1"

# XRANGE

# 语法

XRANGE key start end [COUNT count]

# 描述

该命令会返回匹配指定 ID 范围的流（stream）条目。范围由一个最小 ID 和一个最大 ID 来确定。所有 ID 介于这两个值之间，或者正好等于其中某个指定 ID 的条目都会被返回

# 参数

  • key

Stream键的名称

  • start

开始扫描的起始消息 ID。传入 0 表示从 PEL 的开头开始

  • COUNT

要返回的最大条目数

# 示例

> XADD writers * name Virginia surname Woolf
"1782980069463-0"
> XADD writers * name Jane surname Austen
"1782980069464-0"
> XADD writers * name Toni surname Morrison
"1782980069464-1"
> XADD writers * name Agatha surname Christie
"1782980069464-2"
> XADD writers * name Ngozi surname Adichie
"1782980069465-0"
> XLEN writers
(integer) 5
> XRANGE writers - + COUNT 2
1) 1) "1782980069463-0"
   2) 1) "name"
      2) "Virginia"
      3) "surname"
      4) "Woolf"
2) 1) "1782980069464-0"
   2) 1) "name"
      2) "Jane"
      3) "surname"
      4) "Austen"

# XREAD

# 语法

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id
  [id ...]```

#### 描述

从单个或多个stream中读取数据，且只返回那些 ID 大于调用者上次接收到的 ID 的条目。如果当前没有可用的数据，该命令还提供了一个阻塞选项，其工作方式与 BRPOP、BZPOPMIN 等命令类似

#### 参数

- **COUNT**

指定要返回的条目数量。设为 0 会返回所有条目；默认值为 10

- **STREAMS key**

需要读取的键，后面紧跟每个键对应的 ID。对于每个stream，将返回 ID 大于给定 ID 的条目。如果传入 $，则只会读取在调用该命令之后新增的条目


#### 示例

```bash
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"
   2) 1) 1) 1526984818136-0
         2) 1) "duration"
            2) "1532"
            3) "event-id"
            4) "5"
            5) "user-id"
            6) "7782813"
      2) 1) 1526999352406-0
         2) 1) "duration"
            2) "812"
            3) "event-id"
            4) "9"
            5) "user-id"
            6) "388234"
2) 1) "writers"
   2) 1) 1) 1526985676425-0
         2) 1) "name"
            2) "Virginia"
            3) "surname"
            4) "Woolf"
      2) 1) 1526985685298-0
         2) 1) "name"
            2) "Jane"
            3) "surname"
            4) "Austen"

STREAMS key_1 key_2 key_3 ... key_N ID_1 ID_2 ID_3 ... ID_N

> XREAD COUNT 2 STREAMS mystream writers 1526999352406-0 1526985685298-0
1) 1) "mystream"
   2) 1) 1) 1526999626221-0
         2) 1) "duration"
            2) "911"
            3) "event-id"
            4) "7"
            5) "user-id"
            6) "9488232"
2) 1) "writers"
   2) 1) 1) 1526985691746-0
         2) 1) "name"
            2) "Toni"
            3) "surname"
            4) "Morrison"
      2) 1) 1526985712947-0
         2) 1) "name"
            2) "Agatha"
            3) "surname"
            4) "Christie"

# XREADGROUP

# 语法

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds]
  [CLAIM min-idle-time] [NOACK] STREAMS key [key ...] id [id ...]

# 描述

XREADGROUP 命令是 XREAD 命令的一个特殊版本，增加了对消费者组的支持

# 参数

  • STREAMS key

需要读取的键，后面紧跟每个键对应的 ID。对于每个stream，将返回 ID 大于给定 ID 的条目。如果传入 $，则只会读取在调用该命令之后新增的条目

  • GROUP consumer

要读取的消费者组和消费者名称

# 示例

WHILE true
    entries = XREADGROUP GROUP $GroupName $ConsumerName BLOCK 2000 COUNT 10 STREAMS mystream >
    if entries == nil
        puts "Timeout... try again"
        CONTINUE
    end

    FOREACH entries AS stream_entries
        FOREACH stream_entries as message
            process_message(message.id,message.fields)

            # ACK the message as processed
            XACK mystream $GroupName message.id
        END
    END
END

# XREVRANGE

# 语法

XREVRANGE key end start [COUNT count]

# 描述

该命令的作用与 XRANGE 类似，但返回的条目是逆序的。使用时需要先指定结束 ID，再指定起始 ID。XREVRANGE 会返回这两个 ID 之间（包含这两个 ID）的条目，并且从范围的尾部开始返回

# 参数

  • key

Stream键的名称

  • start

开始扫描的起始消息 ID。传入 0 表示从 PEL 的开头开始

  • end

ID 范围的结束值（包含该 ID）。返回的条目是逆序的。使用 + 代表最大的 ID

  • COUNT

要返回的最大条目数

# 示例

> XADD writers * name Virginia surname Woolf
"1782981076345-0"
> XADD writers * name Jane surname Austen
"1782981076346-0"
> XADD writers * name Toni surname Morrison
"1782981076347-0"
> XADD writers * name Agatha surname Christie
"1782981076348-0"
> XADD writers * name Ngozi surname Adichie
"1782981076348-1"
> XLEN writers
(integer) 5
> XREVRANGE writers + - COUNT 1
1) 1) "1782981076348-1"
   2) 1) "name"
      2) "Ngozi"
      3) "surname"
      4) "Adichie"

# XSETID

# 语法

XSETID key last-id [ENTRIESADDED entries-added]
  [MAXDELETEDID max-deleted-id]

# 描述

XSETID 是一个内部命令，用户不应直接调用它

# 参数

  • key

Stream键的名称

  • end

ID 范围的结束值（包含该 ID）


---

## XTRIM

#### 语法

```bash
XTRIM key <MAXLEN | MINID> [= | ~] threshold [LIMIT count] [KEEPREF | DELREF | ACKED]

# 描述

如果需要，XTRIM 会通过淘汰较旧的条目（即 ID 较小的条目）来截断Stream

# 参数

  • key

Stream键的名称

# 示例

> XADD mystream * field1 A field2 B field3 C field4 D
"1782981471050-0"
> XTRIM mystream MAXLEN 2
(integer) 0
> XRANGE mystream - +
1) 1) "1782981471050-0"
   2) 1) "field1"
      2) "A"
      3) "field2"
      4) "B"
      5) "field3"
      6) "C"
      7) "field4"
      8) "D"

← JSON命令 →