#### 第18章：
#### Redis
Redis是一个key-value存储系统，常被用作缓存或者消息订阅通知系统。是一个开源的使用C语言编写、遵守BSD协议、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。Redis支持数据持久化，支持多种数据结构的存储，支持master-slave模式的数据备份。
##### Redis的优势
- 性能极高 – Redis能读的速度是110000次/s,写的速度是81000次/s 。
- 丰富的数据类型 – Redis支持二进制案例的 Strings, Lists, Hashes, Sets 及 Ordered Sets 数据类型操作。
- 原子 – Redis的所有操作都是原子性的，意思就是要么成功执行要么失败完全不执行。单个操作是原子性的。多个操作也支持事务，即原子性，通过MULTI和EXEC指令包起来，但是不支持回滚(据说是作者一开始就这么设计为了纯粹的性能)。
- 丰富的特性 – Redis还支持 publish/subscribe, 通知, key 过期等等特性。
#### 18.1 安装Redis
##### Redis 的安装
Linux环境，下载地址：<http://redis.io/download>，下载最新稳定版本并安装：
```
# wget http://download.redis.io/releases/redis-6.0.8.tar.gz
# tar xzf redis-6.0.8.tar.gz
# cd redis-6.0.8
# make
```
执行完 make 命令后，redis-6.0.8 的 src 目录下会出现编译后的 redis 服务程序 redis-server，还有用于测试的客户端程序 redis-cli：
下面启动 redis 服务：
```
# cd src
# ./redis-server
```
注意这种方式启动 redis 使用的是默认配置。也可以通过启动参数告诉 redis 使用指定配置文件使用下面命令启动。
```
# cd src
# ./redis-server ../redis.conf
```
**redis.conf** 是一个默认的配置文件。我们可以根据需要使用自己的配置文件。
启动 redis 服务进程后，就可以使用测试客户端程序 redis-cli 和 redis 服务交互了。 比如：
```
# cd src
# ./redis-cli
redis> set foo bar
OK
redis> get foo
"bar"
```
#### 18.2 Redis 的配置文件
redis.conf 配置项说明如下：
| 配置项 | 说明 |
| ------------------------------------------------------------ | ------------------------------------------------------------ |
| `daemonize no` | Redis 默认不是以守护进程的方式运行，可以通过该配置项修改，使用 yes 启用守护进程（Windows 不支持守护线程的配置为 no ） |
| `pidfile /var/run/redis.pid` | 当 Redis 以守护进程方式运行时，Redis 默认会把 pid 写入 /var/run/redis.pid 文件，可以通过 pidfile 指定 |
| `port 6379` | 指定 Redis 监听端口，默认端口为 6379，作者在自己的一篇博文中解释了为什么选用 6379 作为默认端口，因为 6379 在手机按键上 MERZ 对应的号码，而 MERZ 取自意大利歌女 Alessia Merz 的名字 |
| `bind 127.0.0.1` | 绑定的主机地址 |
| `timeout 300` | 当客户端闲置多长秒后关闭连接，如果指定为 0 ，表示关闭该功能 |
| `loglevel notice` | 指定日志记录级别，Redis 总共支持四个级别：debug、verbose、notice、warning，默认为 notice |
| `logfile stdout` | 日志记录方式，默认为标准输出，如果配置 Redis 为守护进程方式运行，而这里又配置为日志记录方式为标准输出，则日志将会发送给 /dev/null |
| `databases 16` | 设置数据库的数量，默认数据库为0，可以使用SELECT 命令在连接上指定数据库id |
| `save <seconds> <changes>`Redis 默认配置文件中提供了三个条件：save 900 1 ，save 300 10，save 60 10000分别表示 900 秒（15 分钟）内有 1 个更改，300 秒（5 分钟）内有 10 个更改以及 60 秒内有 10000 个更改。 | 用于Redis的AOF持久化，指定在多长时间内，有多少次更新操作，就将数据同步到数据文件，可以多个条件配合 |
| `rdbcompression yes` | 指定存储至本地数据库时是否压缩数据，默认为 yes，Redis 采用 LZF 压缩，如果为了节省 CPU 时间，可以关闭该选项，但会导致数据库文件变的巨大 |
| `dbfilename dump.rdb` | 指定本地数据库文件名，默认值为 dump.rdb |
| `dir ./` | 指定本地数据库存放目录 |
| `slaveof <masterip> <masterport>` | 设置当本机为 slave 服务时，设置 master 服务的 IP 地址及端口，在 Redis 启动时，它会自动从 master 进行数据同步 |
| `masterauth <master-password>` | 当 master 服务设置了密码保护时，slav 服务连接 master 的密码 |
| `requirepass foobared` | 设置 Redis 连接密码，如果配置了连接密码，客户端在连接 Redis 时需要通过 AUTH <password> 命令提供密码，默认关闭 |
| ` maxclients 128` | 设置同一时间最大客户端连接数，默认无限制，Redis 可以同时打开的客户端连接数为 Redis 进程可以打开的最大文件描述符数，如果设置 maxclients 0，表示不作限制。当客户端连接数到达限制时，Redis 会关闭新的连接并向客户端返回 max number of clients reached 错误信息 |
| `maxmemory <bytes>` | 指定 Redis 最大内存限制，Redis 在启动时会把数据加载到内存中，达到最大内存后，Redis 会先尝试清除已到期或即将到期的 Key，当此方法处理 后，仍然到达最大内存设置，将无法再进行写入操作，但仍然可以进行读取操作。Redis 新的 vm 机制，会把 Key 存放内存，Value 会存放在 swap 区 |
| `appendonly no` | 指定是否在每次更新操作后进行日志记录，Redis 在默认情况下是异步的把数据写入磁盘，如果不开启，可能会在断电时导致一段时间内的数据丢失。因为 redis 本身同步数据文件是按上面 save 条件来同步的，所以有的数据会在一段时间内只存在于内存中。默认为 no |
| `appendfilename appendonly.aof` | 指定更新日志文件名，默认为 appendonly.aof |
| `appendfsync everysec` | 指定更新日志条件，共有 3 个可选值：**no**：表示等操作系统进行数据缓存同步到磁盘（快）**always**：表示每次更新操作后手动调用 fsync() 将数据写到磁盘（慢，安全）**everysec**：表示每秒同步一次（折中，默认值） |
| `vm-enabled no` | 指定是否启用虚拟内存机制，默认值为 no，简单的介绍一下，VM 机制将数据分页存放，由 Redis 将访问量较少的页即冷数据 swap 到磁盘上，访问多的页面由磁盘自动换出到内存中 |
| `vm-swap-file /tmp/redis.swap` | 虚拟内存文件路径，默认值为 /tmp/redis.swap，不可多个 Redis 实例共享 |
| `vm-max-memory 0` | 将所有大于 vm-max-memory 的数据存入虚拟内存，无论 vm-max-memory 设置多小，所有索引数据都是内存存储的(Redis 的索引数据 就是 keys)，也就是说，当 vm-max-memory 设置为 0 的时候，其实是所有 value 都存在于磁盘。默认值为 0 |
| `vm-page-size 32` | Redis swap 文件分成了很多的 page，一个对象可以保存在多个 page 上面，但一个 page 上不能被多个对象共享，vm-page-size 是要根据存储的 数据大小来设定的，作者建议如果存储很多小对象，page 大小最好设置为 32 或者 64bytes；如果存储很大大对象，则可以使用更大的 page，如果不确定，就使用默认值 |
| `vm-pages 134217728` | 设置 swap 文件中的 page 数量，由于页表（一种表示页面空闲或使用的 bitmap）是在放在内存中的，，在磁盘上每 8 个 pages 将消耗 1byte 的内存。 |
| `vm-max-threads 4` | 设置访问swap文件的线程数,最好不要超过机器的核数,如果设置为0,那么所有对swap文件的操作都是串行的，可能会造成比较长时间的延迟。默认值为4 |
| `glueoutputbuf yes` | 设置在向客户端应答时，是否把较小的包合并为一个包发送，默认为开启 |
| `hash-max-zipmap-entries 64 hash-max-zipmap-value 512` | 指定在超过一定的数量或者最大的元素超过某一临界值时，采用一种特殊的哈希算法 |
| `activerehashing yes` | 指定是否激活重置哈希，默认为开启（后面在介绍 Redis 的哈希算法时具体介绍） |
| `include /path/to/local.conf` | 指定包含其它的配置文件，可以在同一主机上多个Redis实例之间使用同一份配置文件，而同时各个实例又拥有自己的特定配置文件 |
#### 18.3 Redis 数据类型
Redis有五种数据l类型：
- 字符串
- 哈希
- 列表
- 集合
- 有序集合
##### String(字符串)
String是Redis最基本的数据类型，和memcached一样的数据类型，一个key对应一个value。
string 类型是二进制安全的。意思是 redis 的 string 可以包含任何数据。比如jpg图片或者序列化的对象。
string 类型是 Redis 最基本的数据类型，string 类型的值最大能存储 512MB。
例子：
```
redis 127.0.0.1:6379> SET msg1 "学习Redis"
OK
redis 127.0.0.1:6379> GET msg1
"学习Redis"
```
将key设置为str1使用SET命令。
下表列出了常用的 redis 字符串命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| SET key value 设置指定 key 的值 |
| GET key 获取指定 key 的值。 |
| GETRANGE key start end 返回 key 中字符串值的子字符 |
| GETSET key value 将给定 key 的值设为 value ，并返回 key 的旧值(old value)。 |
| GETBIT key offset 对 key 所储存的字符串值，获取指定偏移量上的位(bit)。 |
| [MGET key1 key2..\] 获取所有(一个或多个)给定 key 的值。 |
| SETBIT key offset value 对 key 所储存的字符串值，设置或清除指定偏移量上的位(bit)。 |
| SETEX key seconds value 将值 value 关联到 key ，并将 key 的过期时间设为 seconds (以秒为单位)。 |
| SETNX key value 只有在 key 不存在时设置 key 的值。 |
| SETRANGE key offset value 用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始。 |
| STRLEN key 返回 key 所储存的字符串值的长度。 |
| [MSET key value key value ...\] 同时设置一个或多个 key-value 对。 |
| [MSETNX key value key value ...\] 同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在。 |
| PSETEX key milliseconds value 这个命令和 SETEX 命令相似，但它以毫秒为单位设置 key 的生存时间，而不是像 SETEX 命令那样，以秒为单位。 |
| INCR key 将 key 中储存的数字值增一。 |
| INCRBY key increment 将 key 所储存的值加上给定的增量值（increment） 。 |
| INCRBYFLOAT key increment 将 key 所储存的值加上给定的浮点增量值（increment） 。 |
| DECR key 将 key 中储存的数字值减一。 |
| DECRBY key decrement key 所储存的值减去给定的减量值（decrement） 。 |
| APPEND key value 如果 key 已经存在并且是一个字符串， APPEND 命令将指定的 value 追加到该 key 原来值（value）的末尾。 |
##### Hash（哈希）
Redis hash 是一个键值(key=>value)对集合。
Redis hash 是一个 string 类型的 field 和 value 的映射表，hash 特别适合用于存储对象。
例子：
```
redis 127.0.0.1:6379> HMSET ob k1 "100" k2 "200"
"OK"
redis 127.0.0.1:6379> HGET ob k1
"100"
redis 127.0.0.1:6379> HGET ob k2
"200"
```
这里设置了一个哈希对象ob。其中k1的值为100，k2的值为200。
我们通过HMSET设置了两个` field=>value`对, HGET 获取对应 field对应的 value
每个 hash 可以存储 2的32次方-1个 键值对（40多亿）。
下表列出了 redis hash 基本的相关命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| [HDEL key field1 field2\] 删除一个或多个哈希表字段 |
| HEXISTS key field 查看哈希表 key 中，指定的字段是否存在。 |
| HGET key field 获取存储在哈希表中指定字段的值。 |
| HGETALL key 获取在哈希表中指定 key 的所有字段和值 |
| HINCRBY key field increment 为哈希表 key 中的指定字段的整数值加上增量 increment 。 |
| HINCRBYFLOAT key field increment 为哈希表 key 中的指定字段的浮点数值加上增量 increment 。 |
| HKEYS key 获取所有哈希表中的字段 |
| HLEN key 获取哈希表中字段的数量 |
| [HMGET key field1 field2\] 获取所有给定字段的值 |
| [HMSET key field1 value1 field2 value2 \] 同时将多个 field-value (域-值)对设置到哈希表 key 中。 |
| HSET key field value 将哈希表 key 中的字段 field 的值设为 value 。 |
| HSETNX key field value 只有在字段 field 不存在时，设置哈希表字段的值。 |
| HVALS key 获取哈希表中所有值。 |
| HSCAN key cursor [MATCH pattern\] [COUNT count] 迭代哈希表中的键值对。 |
##### List（列表或称为队列）
Redis 列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部或者尾部。
例子：
```
redis 127.0.0.1:6379> lpush list1 1
(integer) 1
redis 127.0.0.1:6379> lpush list1 2
(integer) 2
redis 127.0.0.1:6379> lpush list1 3
(integer) 3
redis 127.0.0.1:6379> lrange list1 0 10
1) "3"
2) "2"
3) "1"
redis 127.0.0.1:6379>
```
上面我们使用lpush命令向list1队列左边插入数据，如果没有list1时会自动创建list1队列对象。lrange命令按照偏移获取队列数据。
下表列出了列表相关的基本命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| [BLPOP key1 key2 \] timeout 移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。 |
| [BRPOP key1 key2 \] timeout 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。 |
| BRPOPLPUSH source destination timeout 从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。 |
| LINDEX key index 通过索引获取列表中的元素 |
| LINSERT key BEFORE\|AFTER pivot value 在列表的元素前或者后插入元素 |
| LLEN key 获取列表长度 |
| LPOP key 移出并获取列表的第一个元素 |
| [LPUSH key value1 value2\] 将一个或多个值插入到列表头部 |
| LPUSHX key value 将一个值插入到已存在的列表头部 |
| LRANGE key start stop 获取列表指定范围内的元素 |
| LREM key count value 移除列表元素 |
| LSET key index value 通过索引设置列表元素的值 |
| LTRIM key start stop 对一个列表进行修剪(trim)，就是说，让列表只保留指定区间内的元素，不在指定区间之内的元素都将被删除。 |
| RPOP key 移除列表的最后一个元素，返回值为移除的元素。 |
| RPOPLPUSH source destination 移除列表的最后一个元素，并将该元素添加到另一个列表并返回 |
| [RPUSH key value1 value2\] 在列表中添加一个或多个值 |
| RPUSHX key value 为已存在的列表添加值 |
#### 集合Set
Redis 的 Set 是 string 类型的无序集合。
集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是 O(1)。哈希表(hashtable)是利用哈希函数将特定的键映射到特定的值得一种数据结构，它维护着键和值的一一对应的关系，并且可以根据键快速检索到值，查询效率为O(1)。Redis利用哈希函数和连地址法将装有集合数据的bucket块通过链表链接在一起并解决了哈希冲突。
sadd 命令
添加一个 string 元素到 key 对应的 set 集合中，成功返回 1，如果元素已经在集合中返回 0。
```
sadd key member
```
例子：
```
redis 127.0.0.1:6379> DEL runoob
redis 127.0.0.1:6379> sadd runoob redis
(integer) 1
redis 127.0.0.1:6379> sadd runoob mongodb
(integer) 1
redis 127.0.0.1:6379> sadd runoob rabbitmq
(integer) 1
redis 127.0.0.1:6379> sadd runoob rabbitmq
(integer) 0
redis 127.0.0.1:6379> smembers runoob
1) "redis"
2) "rabbitmq"
3) "mongodb"
```
rabbitmq 添加了两次，但根据集合内元素的唯一性，第二次插入的元素将被忽略。
集合中最大的成员数为 2的32次方-1个(4294967295, 每个集合可存储40多亿个成员)。
下表列出了 Redis 集合基本命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| [SADD key member1 member2\] 向集合添加一个或多个成员 |
| SCARD key 获取集合的成员数 |
| [SDIFF key1 key2\] 返回第一个集合与其他集合之间的差异。 |
| [SDIFFSTORE destination key1 key2\] 返回给定所有集合的差集并存储在 destination 中 |
| [SINTER key1 key2\] 返回给定所有集合的交集 |
| [SINTERSTORE destination key1 key2\] 返回给定所有集合的交集并存储在 destination 中 |
| SISMEMBER key member 判断 member 元素是否是集合 key 的成员 |
| SMEMBERS key 返回集合中的所有成员 |
| SMOVE source destination member 将 member 元素从 source 集合移动到 destination 集合 |
| SPOP key 移除并返回集合中的一个随机元素 |
| [SRANDMEMBER key count\] 返回集合中一个或多个随机数 |
| [SREM key member1 member2\] 移除集合中一个或多个成员 |
| [SUNION key1 key2\] 返回所有给定集合的并集 |
| [SUNIONSTORE destination key1 key2\] 所有给定集合的并集存储在 destination 集合中 |
| SSCAN key cursor [MATCH pattern\] [COUNT count] 迭代集合中的元素 |
#### zset(sorted set：有序集合)
Redis zset 和 set 一样也是string类型元素的集合,且不允许重复的成员。有序集合底层的数据结构是跳跃表。
不同的是每个元素都会关联一个double类型的分数。redis正是通过分数来为集合中的成员进行从小到大的排序。
zset的成员是唯一的,但分数(score)却可以重复。
### zadd 命令
添加元素到集合，元素在集合中存在则更新对应score
```
zadd key score member
```
例子：
```
redis 127.0.0.1:6379> DEL runoob
redis 127.0.0.1:6379> zadd runoob 0 redis
(integer) 1
redis 127.0.0.1:6379> zadd runoob 0 mongodb
(integer) 1
redis 127.0.0.1:6379> zadd runoob 0 rabbitmq
(integer) 1
redis 127.0.0.1:6379> zadd runoob 0 rabbitmq
(integer) 0
redis 127.0.0.1:6379> ZRANGEBYSCORE runoob 0 1000
1) "mongodb"
2) "rabbitmq"
3) "redis"
```
下表列出了 redis 有序集合的基本命令:
| 命令及描述 |
| ------------------------------------------------------------ |
| [ZADD key score1 member1 score2 member2\] 向有序集合添加一个或多个成员，或者更新已存在成员的分数 |
| ZCARD key 获取有序集合的成员数 |
| ZCOUNT key min max 计算在有序集合中指定区间分数的成员数 |
| ZINCRBY key increment member 有序集合中对指定成员的分数加上增量 increment |
| [ZINTERSTORE destination numkeys key key ...\] 计算给定的一个或多个有序集的交集并将结果集存储在新的有序集合 destination 中 |
| ZLEXCOUNT key min max 在有序集合中计算指定字典区间内成员数量 |
| [ZRANGE key start stop WITHSCORES\] 通过索引区间返回有序集合指定区间内的成员 |
| [ZRANGEBYLEX key min max LIMIT offset count\] 通过字典区间返回有序集合的成员 |
| ZRANGEBYSCORE key min max [WITHSCORES\] [LIMIT] 通过分数返回有序集合指定区间内的成员 |
| ZRANK key member 返回有序集合中指定成员的索引 |
| [ZREM key member member ...\] 移除有序集合中的一个或多个成员 |
| ZREMRANGEBYLEX key min max 移除有序集合中给定的字典区间的所有成员 |
| ZREMRANGEBYRANK key start stop 移除有序集合中给定的排名区间的所有成员 |
| ZREMRANGEBYSCORE key min max 移除有序集合中给定的分数区间的所有成员 |
| [ZREVRANGE key start stop WITHSCORES\] 返回有序集中指定区间内的成员，通过索引，分数从高到低 |
| [ZREVRANGEBYSCORE key max min WITHSCORES\] 返回有序集中指定分数区间内的成员，分数从高到低排序 |
| ZREVRANK key member 返回有序集合中指定成员的排名，有序集成员按分数值递减(从大到小)排序 |
| ZSCORE key member 返回有序集中，成员的分数值 |
| [ZUNIONSTORE destination numkeys key key ...\] 计算给定的一个或多个有序集的并集，并存储在新的 key 中 |
| ZSCAN key cursor [MATCH pattern\] [COUNT count] 迭代有序集合中的元素（包括元素成员和元素分值） |
#### 18.4 命令
Redis 命令用于在 redis 服务上执行操作。
要在 redis 服务上执行命令需要一个 redis 客户端，并启动客户端。
启动 redis 服务器，打开终端并输入命令`redis-cli`，该命令会连接本地的 redis 服务。
```
$ redis-cli
redis 127.0.0.1:6379>
redis 127.0.0.1:6379> PING
PONG
```
我们连接到本地的 redis 服务并执行 **PING** 命令，该命令用于检测 redis 服务是否启动。
如果需要在远程 redis 服务上执行命令，同样我们使用的也是redis-cli命令。
```
$redis-cli -h 127.0.0.1 -p 6379 -a "mypass"
redis 127.0.0.1:6379>
redis 127.0.0.1:6379> PING
PONG
```
##### Redis 键(key)
Redis 键命令用于管理 redis 的键。
Redis 键相关的基本命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| DEL key该命令用于在 key 存在时删除 key。 |
| DUMP key序列化给定 key ，并返回被序列化的值。 |
| EXISTS key检查给定 key 是否存在。 |
| EXPIRE keyseconds 为给定 key 设置过期时间，以秒计。 |
| EXPIREAT key timestamp EXPIREAT 的作用和 EXPIRE 类似，都用于为 key 设置过期时间。 不同在于 EXPIREAT 命令接受的时间参数是 UNIX 时间戳(unix timestamp)。 |
| PEXPIRE key milliseconds设置 key 的过期时间以毫秒计。 |
| PEXPIREAT key milliseconds-timestamp 设置 key 过期时间的时间戳(unix timestamp) 以毫秒计 |
| KEYS pattern查找所有符合给定模式( pattern)的 key 。 |
| MOVE key db 将当前数据库的 key 移动到给定的数据库 db 当中。 |
| PERSIST key 移除 key 的过期时间，key 将持久保持。 |
| PTTL key以毫秒为单位返回 key 的剩余的过期时间。 |
| TTL key以秒为单位，返回给定 key 的剩余生存时间(TTL, time to live)。 |
| RANDOMKEY从当前数据库中随机返回一个 key 。 |
| RENAME key newkey修改 key 的名称 |
| RENAMENX key newkey仅当 newkey 不存在时，将 key 改名为 newkey 。 |
| SCAN cursor [MATCH pattern\] [COUNT count] 迭代数据库中的数据库键。Redis的多种存储对象根据存储的量以及所占空间大小的不同可以手动激活或者自动激活其底层数据结构和数据类型的变化来提高性能和减少空间使用量。 |
| TYPE key 返回 key 所储存的值的类型。 |
#### 18.5 事务
Redis 事务可以一次执行多个命令， 并且带有以下三个重要的保证：
- 批量操作在发送 EXEC 命令前被放入队列缓存。
- 收到 EXEC 命令后进入事务执行，事务中任意命令执行失败，其余的命令依然被执行。
- 在事务执行过程，其他客户端提交的命令请求不会插入到事务执行命令序列中。
一个事务从开始到执行会经历以下三个阶段：
- 开始事务。
- 命令入队。
- 执行事务。
注意：
- Redis的事务不会回滚。设计者设计之初希望Redis是纯粹的性能强悍的NoSQL数据库。
- Redis的事务不支持原子性(要么全部执行成功，要么错一条全部不成功)，但是在将命令加入事务队列时，Redis会检测事务中每一条命令是否错误，如果发现语法错误、命令不存在或命令参数不对则会返回错误给客户端并修改客户端状态，当输入事务执行令EXEC时，Redis服务器会拒绝执行此事务。但是Redis不支持检测程序员自己造成的逻辑错误。Redis事务队列中命令语法正确的前提下，由程序员自己造成的逻辑错误，甚至数据与数据类型异常的情况中，会在异常的那条命令会输出错误后执行之后的命令。
下表列出了 redis 事务的相关命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| DISCARD 取消事务，放弃执行事务块内的所有命令。 |
| EXEC 执行所有事务块内的命令。 |
| MULTI 标记一个事务块的开始。 |
| UNWATCH 取消 WATCH 命令对所有 key 的监视。 |
| [WATCH key key ...\] 监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。 |
例子：
```
redis 127.0.0.1:6379> MULTI
OK
redis 127.0.0.1:6379> SET str1 "1"
QUEUED
redis 127.0.0.1:6379> GET str1
QUEUED
redis 127.0.0.1:6379> SADD tag "1 "2" "3"
QUEUED
redis 127.0.0.1:6379> SMEMBERS tag
QUEUED
redis 127.0.0.1:6379> EXEC
1) OK
2) "1"
3) (integer) 3
4) 1) "2"
2) "1"
3) "3"
```
#### 18.6 Redis 脚本
Redis支持使用Lua语言写的脚本。有兴趣的同学可以自行学习Lua语言。
例子：
```
redis 127.0.0.1:6379> EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 first second
1) "key1"
2) "key2"
3) "first"
4) "second"
```
#### 18.7 Redis 发布订阅
Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。
Redis 客户端可以订阅任意数量的频道。
频道 channel1与客户端之间的关系：
1. 首先客户端控制服务端创建频道channel1。
2. 一个或者多个客户端client保持连接并订阅了(subscribe)一个频道channel1。
:-: 
1. 当有客户端使用PUBLISH命令发送给频道channel1时， 这个消息会被发送给订阅了它的一个或多个客户端client。
:-: 
例子：
首先在第一个客户端，我们创建并订阅一个频道。
```
redis 127.0.0.1:6379> SUBSCRIBE test1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redisChat"
3) (integer) 1
```
然后我们打开另外一个Redis客户端向这个频道发送信息。
```
redis 127.0.0.1:6379> PUBLISH test1 "111"
(integer) 1
redis 127.0.0.1:6379> PUBLISH test1 "222"
(integer) 1
# 订阅者的客户端会显示如下消息
1) "message"
2) "test1"
3) "111"
1) "message"
2) "test1"
3) "222"
```
下表列出了 redis 发布订阅常用命令：
| 命令及描述 |
| ------------------------------------------------------------ |
| [PSUBSCRIBE pattern pattern ...\] 订阅一个或多个符合给定模式的频道。 |
| PUBSUB subcommand [argument [argument ...\]] 查看订阅与发布系统状态。 |
| PUBLISH channel message 将信息发送到指定的频道。 |
| PUNSUBSCRIBE [pattern [pattern ...\]] 退订所有给定模式的频道。 |
| [SUBSCRIBE channel channel ...\] 订阅给定的一个或多个频道的信息。 |
| UNSUBSCRIBE [channel [channel ...\]] 指退订给定的频道。 |
##### PHP利用发布订阅实现即时通信
这里使用了PHP的动态扩展：Redis扩展。
```
<?php
//发布
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$message='新年快乐';
$ret=$redis->publish('中央广播电台',$message);
?>
```
客户端1向频道"中央广播电台"发送信息"新年快乐"。
```
<?php
//订阅
ini_set('default_socket_timeout', -1); //不超时
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$result=$redis->subscribe(array('中央广播电台'), 'callback');
function callback($instance,$channelName,$message){
echo $message;
}
?>
```
另一个保持连接的客户端2由于订阅了频道"中央广播电台"而执行回调函数，echo输出上一个客户端发布的消息"新年快乐"。而且不论客户端1在任何时候在频道"中央广播电台"发布任何消息，客户端2都会实时接收到频道传来的信息并输出。
#### 18.8 Redis Stream
Redis Stream 是 Redis 5.0 版本新增加的数据结构。
Redis Stream 主要用于消息队列，Redis 本身是有一个 Redis 发布订阅 (pub/sub) 来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃。
简单来说发布订阅 (pub/sub) 可以分发消息，但无法记录历史消息。
而 Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。
Redis Stream 的结构如下所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容：
:-: 
每个 Stream 都有唯一的名称，它就是 Redis 的 key，在我们首次使用 xadd 指令追加消息时自动创建。
上图解析：
- **Consumer Group** ：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer)。
- **last_delivered_id** ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
- **pending_ids** ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）。
Redis Stream 区别于RabbitMQ等消息队列软件的几点：
- Redis Stream的每条消息只能由一个消费者读取，其他消费者不可读取。而RabbitMQ可以按照策略将一条消息发送给多个消费者。
- Redis Stream是主动读取。RabbitMQ是由软件按照策略、算法、机制自动分发。
**消息队列相关命令：**
- **XADD** - 添加消息到末尾
- **XTRIM** - 对流进行修剪，限制长度
- **XDEL** - 删除消息
- **XLEN** - 获取流包含的元素数量，即消息长度
- **XRANGE** - 获取消息列表，会自动过滤已经删除的消息
- **XREVRANGE** - 反向获取消息列表，ID 从大到小
- **XREAD** - 以阻塞或非阻塞方式获取消息列表
**消费者组相关命令：**
- **XGROUP CREATE** - 创建消费者组
- **XREADGROUP GROUP** - 读取消费者组中的消息
- **XACK** - 将消息标记为"已处理"
- **XGROUP SETID** - 为消费者组设置新的最后递送消息ID
- **XGROUP DELCONSUMER** - 删除消费者
- **XGROUP DESTROY** - 删除消费者组
- **XPENDING** - 显示待处理消息的相关信息
- **XCLAIM** - 转移消息的归属权
- **XINFO** - 查看流和消费者组的相关信息；
- **XINFO GROUPS** - 打印消费者组的信息；
- **XINFO STREAM** - 打印流信息
例子：
使用了Redis扩展。
1. 这里先创建一个链接,给一个stream里面添加数据。
```
$streamKey = 'test:stream:queue';
$redis = new \Redis();
$redis->connect('127.0.0.1');
for ($i = 0; $i < 100; $i++) {
/**
* 队列名
* *: 表示由Redis自己生成消息ID:规则为[毫秒时间戳+自增数]
* 存储的数据
*/
$xAddResult = $redis->xAdd($streamKey, '*', ['field-'.$i => 'value:'.$i*2]);
}
```
1. 删除队列中的某一条消息。
```
/**
* 删除消息
* 队列名
* 消息ID
*/
$xDelResult = $redis->xDel($streamKey, ['1609131229884-0']);
```
1. 查看队列中的消息。
```
/**
* 取出所有的消息
* 队列名
* 消息开始ID: - 不限制开始ID
* 消息结束ID: + 表示不限制
*/
$streamResult = $redis->xRange($streamKey, '-', '+');
```
1. 此时如果要消费队列中的消息,需要先创建一个group与队列关联起来,才可以消费队列中的消息。
```
/**
* 创建一个消费组
* 操作类型:['HELP', 'SETID', 'DELGROUP', 'CREATE', 'DELCONSUMER']
* 队列名
* 消费者 : 这个时候自己随便起名字就可以
* 消息ID : 0 表示从头开始 $ 表示不接收老的消息
*/
$xGroupResult = $redis->xGroup('CREATE', $streamKey, $streamKey.':group_1', 0);
```
1. 获取队列中的消息。
```
/**
* group
* 消费者
* [队列名 => '>' : 特殊>ID，这意味着使用者只想接收从未传递给任何其他使用者的消息。这只是意味着，给我新消息。
* 队列名 => '0' : 任何其他ID（即0或任何其他有效ID或不完整的ID（仅毫秒时间部分））将具有以下效果：返回正在等待用户发送的ID大于提供的ID的命令的条目。因此，基本上，如果ID不是>，那么该命令将只允许客户端访问其挂起的条目：传递给它的消息，但尚未确认。]
* 一次性取多少条消息
*/
$xReadGroupResult = $redis->xReadGroup($streamKey.':group_1', 'consumerA', [$streamKey => '>'], 1);
```
1. 从消费者组内读取消息并处理完成后,需确认该条消息已处理。
```
/**
* 确认消息已处理
* 队列名
* 消费者组
* 消息ID
*/
$xAckResult = $redis->xAck($streamKey, $streamKey.':group_1', ['1609131229885-0']);
```
1. 用来获消费组或消费内消费者的未处理完毕的消息。
```
/**
* 用来获消费组或消费内消费者的未处理完毕的消息。
* 队列名
* 消费者组
*/
$pendingResult = $redis->xPending($streamKey, $streamKey.':group_1');
```