Redis Streams 介绍（一）-京东云开发者社区

Stream 是 Redis 5.0 引入的一种新数据类型，它以一种抽象的方式构建日志结构数据。由于在内存中存储，它实现了强大的操作来克服日志文件的局限性。本质上，Redis Streams是新增的仅追加的数据结构，与其他数据结构相比，它附加的 API 更复杂，功能更强大。

Stream可以用于消息队列，与其它的mq服务相比，它可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置；当消息被消费后，它不会移除消息，而是通过游标下移来控制访问下一条。stream中定义的一个个key相当于mq中的队列。

对于Redis Streams可以应用于缓存来记录一些具有时序特征的数据；也可以作为轻量级的消息队列应用于项目中。

下面一起来看下Redis Streams的相关API。 
### 1 新增条目
XADD 命令将一个新条目追加到指定的流中。如果流不存在会自动创建。
XADD key ID field value [field value ...]
```bash
 >XADD stream1 *  name zs sex n
1625988825375-0
```
上述命令将一个条目 name:zs, sex:n 添加到key为stream1的流中，返回值是该条条目的ID。命令中的*表示希望服务器为我们生成一个新的 ID。
#### 1.1 条目ID
流中每个新的 ID 都会单调递增，条目的ID由两部分组成：
```bash
<millisecondsTime>-<sequenceNumber>

```

其中毫秒时间部分实际上是本地 Redis 节点中生成流 ID 的本地时间，但是如果当前毫秒时间恰好小于前一个入口时间，则使用前一个入口时间，因此如果发生时钟向后跳，单调递增的 ID 属性仍然成立。序列号用于在同一毫秒内创建的条目。由于序列号是 64 位宽，因此实际上在同一毫秒内可以生成的条目数量几乎没有限制。

Redis流支持按 ID 进行范围查询，条目ID包含时间信息天然支持了按时间范围查询。

当然，我们也可以自定义使用ID，这种的新增需要我们显示指定ID。如下：

```bash
> XADD stream2 0-1 name sf sex n
0-1
```

**请注意，在这种情况下，最小 ID 为 0-1，并且该命令不会接受等于或小于前一个 ID 的 ID：**

```bash
> XADD stream2 0-1 name ssaf sex e
ERR The ID specified in XADD is equal or smaller than the target stream top item
```
#### 1.2 限制个数
许多应用程序不想永远将数据收集到流中。有时在流中最多包含给定数量的项目很有用，其他时候一旦达到给定大小，可以将数据从 Redis 移动到不在内存中但适合存储的存储空间中。

XADD命令的**MAXLEN**选项支持对流中的条目数进行限制。

```bash
> XADD stream2 MAXLEN 2 0-2 name sf sex n
0-2
```
使用MAXLEN当达到指定的长度时，旧条目会自动逐出，以便流保持恒定大小。不过使用此参数是不划算的，它对于元素的修改比较耗时，可以使用以下特殊形式的命令：

```bash
XADD stream2 MAXLEN ~ 2 0-2 name sf sex n
```

~ 表示我们并不真的需要这恰好2个项目，它可以是3个或5个，只要确保最少2个项目。使用此参数，仅当我们可以删除整个节点时才执行修改。

还有XTRIM命令，它执行的操作与上面的MAXLEN选项非常相似，只是它可以自己运行：
```bash
> XTRIM stream2 MAXLEN 2
0
```
或者
```bash
> XTRIM stream2 MAXLEN ~ 2
0
```

另一种限制策略是MINID，它驱逐 ID 低于指定值的条目。

```bash
> XTRIM stream2 MINID 0-2
1
```
### 2 查询个数
XLEN 命令 查询流中的条目个数。
XLEN key
```bash
> XLEN stream1
1
```

### 3 查询流中数据
#### 3.1 按范围查询 XRANGE 和 XREVRANGE
要按范围查询流，我们只需要指定两个 ID，start和end。返回的范围将包括以开始或结束作为 ID 的元素，因此范围是包含在内的。两个特殊 ID-和+分别表示可能的最小和最大 ID。

XRANGE key start end [COUNT count]
```bash
> XRANGE stream1 - +
1625988825375-0
name
zs
sex
n
```

返回的每个条目都是一个包含两项的数组：ID 和字段值对列表。

如果想要按时间范围查询，可以指定开始和结束ID的时间部分。例如我想查询一个三毫秒的时间段，可以使用：

```bash
> XRANGE stream1 1625988825375 1625988825377
1625988825375-0
name
zs
sex
n
```
当查询时间范围内数据条目很多时，我们可以使用Count选项,可以获得前N个条目。如果想要更多，可以获取返回的最后一个ID，将序列部分加一，然后再次查询。示例如下：

```bash
> XLEN stream1
6
> XRANGE stream1 - + COUNT 2
1625988825375-0
name
zs
sex
n
1625991055256-0
name
zs1
sex
n

> XRANGE stream1 1625991055256-1 + COUNT 2
1625991062418-0
name
zs2
sex
n
1625991066585-0
name
zs3
sex
n
```
命令XREVRANGE与XRANGE等效，但以相反的顺序返回元素，因此XREVRANGE的实际用途可以检查流中的最后一项是什么。

XREVRANGE key end start [COUNT count]

```bash
> XREVRANGE stream1 + - COUNT 1
1625991087101-0
name
zs5
sex
n
```
请注意，XREVRANGE命令以相反的顺序获取开始和停止参数。
#### 3.2 监听新消息
当我们想要的是侦听到达流中的新消息时，可以使用XREAD命令。
XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
```bash
> XREAD COUNT 2 STREAMS stream1 stream2 0 0
stream1
1625988825375-0
name
zs
sex
n
1625991055256-0
name
zs1
sex
n
stream2
0-1
name
q
sex
n
```
以上是XREAD的非阻塞形式。请注意，COUNT选项不是强制性的，实际上该命令的唯一强制性选项是STREAMS选项，它指定了一个键列表以及调用消费者已经为每个流看到的相应最大 ID，因此该命令将仅向客户端提供 ID 大于我们指定的 ID 的消息。示例中的0表示希望 返回消息的 ID 大于0-0。

**请注意在STREAMS选项之后我们先提供流的key集合，然后是 ID集合。**

可以使用BLOCK参数轻松地将XREAD转换为阻塞命令。

```bash
> XREAD BLOCK 0 STREAMS mystream $
```

上述示例指定了超时为 0 毫秒的新BLOCK选项（这意味着永不超时）。当没有新消息时，它会阻塞。其中$是特殊的ID，表示当前流中的最大ID，即上述命令只用返回当前之后的最新消息，类似Unix 命令tail -f。也可以指定具体ID。

XREAD的阻塞形式也可以监听多个 Streams，只需指定多个键名即可。其中至少有一个流有大于指定ID的消息时会返回结果。
### 4 删除条目
流还有一个特殊的命令，用于从流的中间删除项目（仅通过 ID）。
XDEL key ID [ID ...]
```bash
> XRANGE stream2  - +
0-1
name
sf
sex
n
0-2
name
ssaf
sex
n
> XDEL stream2 0-2
1
> XRANGE stream2  - +
0-1
name
sf
sex
n
```
但是在当前的实现中，直到流节点完全清空时，内存才会真正回收，因此不要滥用此功能。

### 5 总结
本文主要介绍了Redis5新增数据结构的新增、查询、删除API操作，在查询部分介绍了它的监听模式，同一个客户端还可以监听多个流。

对于Redis Streams数据结构的应用，个人理解可以用来记录重要数据的日志，使用监听模式可以方便实时查看新增日志信息。对于流的新增条目会返回递增ID这点，可以用来作为一种生成唯一ID的方式。

此外，对于流消息的消费，Redis还定义了消费者组的概念来允许一组客户端合作消费同一消息流的不同部分。这部分后续再详细介绍。

------------
###### 自猿其说Tech-JDL京东物流技术发展部
###### 作者：网规技术部 管碧强