发布与订阅 P52

Redis 实现了发布与订阅（publish/subscribe）模式，又称 pub/sub 模式（与设计模式中的观察者模式类似）。订阅者负责订阅频道，发送者负责向频道发送二进制字符串消息。每当有消息被发送至给定频道时，频道的所有订阅者都会接收到消息。

发布与订阅命令 P52

命令	格式	描述
SUBSCRIBE	SUBSCRIBE channel [channel ...]	订阅一个或多个频道
UNSUBSCRIBE	UNSUBSCRIBE [channel [channel ...]]	退订一个或多个频道；没有指定频道，则退订全部频道
PUBLISH	PUBLISH channel message	给指定频道发送消息，返回接收到消息的订阅者数量
PSUBSCRIBE	PSUBSCRIBE pattern [pattern ...]	订阅一个或多个模式，与模式匹配的频道均会订阅
PUNSUBSCRIBE	PUNSUBSCRIBE [pattern [pattern ...]]	退订一个或多个模式；没有指定模式，则退订全部模式

相关演示代码如下：

// 执行发布订阅相关操作（注意：pubSubConn 中的 Conn 对象不能是 conn 对象，即必须建立两个不同的连接）

func executePubSubOperation(pubSubConn redis.PubSubConn, conn redis.Conn) {

	// 监听频道消息并输出

	go func() {

		for ; ; {

			switch result := pubSubConn.Receive().(type) {

			case redis.Message:

				// byte 转 string

				resultMap := map[string]string  {

					"Channel": result.Channel,

					"Pattern": result.Pattern,

					"Data": string(result.Data),

				}

				handleResult(resultMap, nil)

			case redis.Subscription:

				handleResult(result, nil)

			}

		}

	}()

	// 订阅两个频道（由于 Subscribe 内没有执行 Receive，所以只有 error，没有错误时就输出 nil）

	// 订阅者收到相应的消息订阅信息，分别输出 -> {subscribe channel_1 1} 和 {subscribe channel_2 2}

	handleResult(nil, pubSubConn.Subscribe("channel_1", "channel_2"))

	// 订阅两个模式，分别以 _1 和 g_2 为结尾的频道 （由于 PSubscribe 内没有执行 Receive，所以只有 error，没有错误时就输出 nil）

	// 订阅者收到相应的消息订阅信息，分别输出 -> {psubscribe *_1 3} 和 {psubscribe *g_2 4}

	handleResult(nil, pubSubConn.PSubscribe("*_1", "*g_2"))

	time.Sleep(time.Second)

	// 发布消息到频道 channel_1，输出 -> 2，两个订阅者接收到消息

	// 订阅者分别输出 -> map[Channel:channel_1 Data:channel1 Pattern:] 和 map[Channel:channel_1 Data:channel1 Pattern:*_1]

	handleResult(conn.Do("PUBLISH", "channel_1", "channel1"))

	// 发布消息到频道 channel_2，输出 -> 1，一个订阅者接收到消息

	// 订阅者输出 -> map[Channel:channel_2 Data:channel1 Pattern:]

	handleResult(conn.Do("PUBLISH", "channel_2", "channel1"))

	// 退订两个频道（由于 Subscribe 内没有执行 Receive，所以只有 error，没有错误时就输出 nil）

	// 订阅者收到相应的消息退订信息，分别输出 -> {unsubscribe channel_1 3} 和 {unsubscribe channel_2 2}

	handleResult(nil, pubSubConn.Unsubscribe("channel_1", "channel_2"))

	// 退订两个频道（由于 Subscribe 内没有执行 Receive，所以只有 error，没有错误时就输出 nil）

	// 订阅者收到相应的消息退订信息，分别输出 -> {punsubscribe *_1 1} 和 {punsubscribe *g_2 0}

	handleResult(nil, pubSubConn.PUnsubscribe("*_1", "*g_2"))

	time.Sleep(time.Second)

}

风险 P54

• 稳定性：旧版 Redis 的客户端读取消息不够快时，不断积压的消息就会使 Redis 的缓冲区越来越大，可能导致 Redis 的速度变慢，甚至直接崩溃，也有使 Redis 可能被操作系统强制杀死。新版 Redis 会自动断开不符合 client-output-buffer-limit pubsub 配置选项要求的客户端。
• 可靠性：任何网络系统在执行操作时都有可能会遇上断线情况，而断线产生的连接错误通常会使得网络连接两端中的其中一端进行重新连接。如果客户端在执行订阅操作的过程中断线，那么客户端将丢失在断线期间发送的所有消息。

排序 P54

SORT 命令可以对列表、集合和有序集合进行排序 ，可以将 SORT 命令看作使 SQL 中的 order by 子句。 P55

命令	格式	描述
SORT	SORT key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]] [ASC|DESC] [ALPHA] [STORE destination]	根据给定的选项，返回或保存给定列表、集合、有序集合 key 中经过排序的元素

可实现功能： P55

• 根据升序或降序排序元素（使用 [ASC|DESC]，默认为升序）
• 将元素看作数字或者字符串进行排序（使用 [ALPHA] 可以当作字符串排序，默认为数字）
• 使用被排序元素之外的其他值作为权重来排序，甚至还可以从输入的列表、集合、有序集合以外的其他地方进行取值（使用 [BY pattern] 可以根据指定值排序；可以使用不存在的键作为参数选项跳过排序没直接返回结果）
• 使用被排序元素之外的其他值作为返回结果（使用 [GET pattern [GET pattern ...]] 可以根据排序结果返回相应的值）
• 保存排序结果（使用 [STORE destination] 可以指定将结果保存到指定 key，此时返回保存的元素的数量）
• 限制返回结果（使用 [LIMIT offset count] 可以指定要跳过的元素数量和返回的元素数量）

相关演示代码如下：

// 执行 SORT 命令

func executeSortOperation(conn redis.Conn) {

	// 删除原有值

	handleResult(redis.Int(conn.Do("DEL", "id", "age", "name", "destination")))

	// 初始化

	handleResult(redis.Int(conn.Do("RPUSH", "id", 1, 4, 3, 2, 5)))

	handleResult(redis.String(conn.Do("SET", "age_1", 15)))

	handleResult(redis.String(conn.Do("SET", "age_2", 14)))

	handleResult(redis.String(conn.Do("SET", "age_3", 11)))

	handleResult(redis.String(conn.Do("SET", "age_4", 12)))

	handleResult(redis.String(conn.Do("SET", "age_5", 10)))

	handleResult(redis.String(conn.Do("SET", "name_1", "tom")))

	handleResult(redis.String(conn.Do("SET", "name_2", "jerry")))

	handleResult(redis.String(conn.Do("SET", "name_3", "bob")))

	handleResult(redis.String(conn.Do("SET", "name_4", "mary")))

	handleResult(redis.String(conn.Do("SET", "name_5", "jack")))

	// 根据 id 降序排序，跳过第一个元素，获取接下来的两个元素，输出 -> [4 3]

	handleResult(redis.Ints(conn.Do("SORT", "id", "LIMIT", "1", "2", "DESC")))

	// 根据 age_{id} 升序排序，按照 id age_{id} name_{id} 顺序返回结果，输出 -> [5 10 jack 3 11 bob 4 12 mary 2 14 jerry 1 15 tom]

	handleResult(redis.Strings(conn.Do("SORT", "id", "BY", "age_*", "GET", "#", "GET", "age_*", "GET", "name_*", "ALPHA")))

	// 根据 name_{id} 字典序降序排序，按照 id age_{id} name_{id} 顺序返回结果，存储到 destination 中

	// 输出 -> 15

	handleResult(redis.Int(conn.Do("SORT", "id", "BY", "name_*", "GET", "#", "GET", "age_*", "GET", "name_*", "ALPHA", "DESC", "STORE", "destination")))

	// 输出 列表 结果，输出 -> [1 15 tom 4 12 mary 2 14 jerry 5 10 jack 3 11 bob]

	handleResult(redis.Strings(conn.Do("LRANGE", "destination", 0, -1)))

}

基本的 Redis 事务

Redis 有 5 个命令可以让用户在不被打断的情况下对多个键执行操作，它们分别是： WATCH 、 MULTI 、 EXEC 、UNWATCH 和 DISCART 。基本的 Redis 事务只用 MULTI 和 EXEC 即可，使用多个命令的事务将在以后进行介绍。 P56

Redis 的基本事务可以让一个客户端在不被其他客户端打断的情况下执行多个命令。当一个事务执行完毕之后， Redis 才会处理其他客户端的命令。 P56

假如某个(或某些) key 正处于 WATCH 命令的监视之下，且事务块中有和这个(或这些) key 相关的命令，那么 EXEC 命令只在这个(或这些) key 没有被其他命令所改动的情况下执行并生效，否则该事务被打断(abort)。

命令	格式	描述
MULTI	MULTI	标记一个事务块的开始，总是返回 OK
EXEC	EXEC	执行所有事务块内的命令，按顺序返回命令的执行结果。当操作被打断时，返回 nil

相关演示代码如下：

// 执行事务命令

func executeTransactionOperation(conn redis.Conn) {

	// 删除原有值

	handleResult(redis.Int(conn.Do("DEL", "counter")))

	// 开启事务（采用流水线方式，降低通信开销）

	handleResult(nil, conn.Send("MULTI"))

	// 事务中执行自增操作（采用流水线方式，降低通信开销）

	handleResult(nil, conn.Send("INCR", "counter"))

	handleResult(nil, conn.Send("INCR", "counter"))

	handleResult(nil, conn.Send("INCR", "counter"))

	// 执行命令，依次执行自增操作，分别返回操作结果，输出 -> [1 2 3]

	handleResult(redis.Ints(conn.Do("EXEC")))

}

练习题：移除竞争条件 P58

简单实践 - 文章投票 中 VoteArticle 函数内曾说明没有事务控制，会存在并发问题。该函数包含一个竞争条件以及一个因为竞争条件而出现的 bug 。函数的竞争条件可能会造成内存泄漏，而函数的 bug 则可能会导致不正确的投票结果出现。你能想办法修复它们吗？

提示：如果你觉得很难理解竞争条件为什么会导致内存泄漏，那么可以在分析 简单实践 - 文章投票 中的 PostArticle 的函数的同时，阅读一下 6.2.5 节。

• 感觉还是无法理解为什么会有这种情况，强行猜测以下可能性（虽然都不是竞争条件造成的）：

  • PostArticle 函数中，在将作者加入到投票用户集合中后，给其设定过期时间。如果设定过期时间之前由于某些原有异常导致没有进行相关操作，那么这个集合将一直在内存中，不会过期，从而造成内存泄漏。
  • VoteArticle 函数中，如果将投票用户添加到投票用户集合中后，还没来得及给文章的相关信息进行设置，那么这个用户以后不能再投票，并且文章的投票信息不对。

• 不是太明白究竟在竞争什么，只能针对以上问题处理。用事务只能再添加一个集合在事务中标记事务是否执行成功，处理流程大致如下：

  1. 先将用户与文章作为值加入到这个集合
  2. 再将用户加入到投票集合中
  3. 然后开启事务，依次发送更新信息的命令和删除那个集合中的相关信息，并执行
  4. 最后有一个 worker 扫描这个集合，将其中值拿出来解析出用户和文章，再查改用户是否已在集合中，如果在集合中，则重新执行 步骤3，最后删除该值

练习题：提高性能 P58

简单实践 - 文章投票 中 ListArticles 函数在获取整个页面的文章时，需要在 Redis 与客户端之间最多会进行 26 次通信往返，这种做法十分低效，你能否想个办法将 ListArticles 函数的往返次数降低为 2 次呢？

提示：使用流水线

• 获取文章列表时，先获取相应的 id 列表（最多 25 个），再循环获取每个 id 对应的文章，所以最多会进行 26 次通信往返
• 由于必须先获取 id 列表，再获取每个 id 对应的文章，所以只能将这两块分开，所以最低至少有 2 次通信往返。大致流程如下：
  1. 先获取 id 列表
  2. 使用流水线，依次将获取每个 id 的文章的命令发送至缓冲区，最后与服务端通信并执行命令（Go 中可以使用上述事务演示代码的方式进行操作 ）
  3. 最后按照顺序解析结果

过期时间 P58

只有少数几个命令可以原子地为键设置过期时间，并且对于列表、集合、哈希表和有序集合这样的容器来说，键过期命令只能为整个键设置过期时间，而没办法为键里面的单个元素设置过期时间（可以使用存储时间戳的有序集合来实现针对单个元素的过期时间；也可以以前缀的形式将容器中的单个元素变为字符串）。 P58

用于处理过期时间的 Redis 命令 P59

命令	格式	描述
PERSIST	PERSIST key	移除键的过期时间
TTL	TTL key	查看键距离过期时间还有多少秒
EXPIRE	EXPIRE key seconds	让键在指定的秒数之后过期
EXPIREAT	EXPIREAT key timestamp	让键在指定的 UNIX 秒级时间戳过期
PTTL	PTTL key	查看键距离过期时间还有多少毫秒
PEXPIRE	PEXPIRE key milliseconds	让键在指定的毫秒数之后过期
PEXPIREAT	PEXPIREAT key milliseconds-timestamp	让键在指定的 UNIX 毫秒级时间戳过期

相关演示代码如下：

// 指定过期时间相关的命令

func executeExpirationOperation(conn redis.Conn) {

	// 删除原有值

	handleResult(redis.Int(conn.Do("DEL", "string")))

	// 设置字符串的值为 value，输出 -> OK，string 变为 -> value

	handleResult(redis.String(conn.Do("SET", "string", "value")))

	// 查看 string 的过期时间，输出 -> -1，表示不过期

	handleResult(redis.Int(conn.Do("TTL", "string")))

	// 设置 string 在 3 秒后过期，输出 -> 1

	handleResult(redis.Int(conn.Do("EXPIRE", "string", 3)))

	time.Sleep(time.Second)

	// 查看 string 的过期时间，输出 -> 2

	handleResult(redis.Int(conn.Do("TTL", "string")))

	// 移除 string 的过期时间，输出 -> 1

	handleResult(redis.Int(conn.Do("PERSIST", "string")))

	// 查看 string 的过期时间，输出 -> -1，表示不过期

	handleResult(redis.Int(conn.Do("TTL", "string")))

	// 设置 string 在当前时间 2500 毫秒后过期，输出 -> 1

	handleResult(redis.Int(conn.Do("PEXPIREAT", "string", time.Now().UnixNano() / 1e6 + 2500)))

	time.Sleep(time.Second)

	// 查看 string 的过期时间，输出 -> 1499，表示还有 1499 毫秒过期

	handleResult(redis.Int(conn.Do("PTTL", "string")))

	time.Sleep(2 * time.Second)

	// 查看 string 的过期时间，输出 -> -2，表示已过期

	handleResult(redis.Int(conn.Do("PTTL", "string")))

}

练习题：使用 EXPIRE 命令代替时间戳有序集合 P59

在简单实践 - Web应用中使用了一个根据时间戳排序、用于清除会话信息的有序集合，通过这个有序集合，程序可以在清理会话的时候，对用户浏览过的商品以及用户购物车里面的商品进行分析。但是，如果我们决定不对商品进行分析的话，那么就可以使用 Redis 提供的过期时间操作来自动清理过期的会话信息，而无须使用清理函数。那么，你能否修改简单实践 - Web应用中定义的 UpdateToken 函数和 UpdateCartItem 函数，让它们使用过期时间操作来删除会话信息，从而代替目前使用有序集合来记录并清除会话信息的做法呢？

• UpdateToken 函数：令牌于 userId 的对应关系不在存储于哈希表中，而是以前缀的形式将容器中的单个元素变为字符串（上面提到过），并设置过期时间，并移除最近操作时间有序集合，这样令牌到期后就会自动删除，不需要清理函数了。
• UpdateCartItem 函数：由于当时此处把 Redis 当作数据库使用，认为购物车不应该随登录态的失效而消失，所以购物车与 userId 挂钩，不存在上述问题。但是如果要让购物车也自动过期，就需要在 UpdateToken 函数内同时设置购物车的过期时间即可。

本文首发于公众号：满赋诸机（点击查看原文） 开源在 GitHub ：reading-notes/redis-in-action