Redis 实战 —— 05. Redis 其余命令简介

发布与订阅 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