传送门：运用Go从零完成一个Redis（三）：完成RESP协议解析器

代码地址：JaricY/miniRedis (github.com)

此代码参考了大佬的godis。godis是一个非常值得学习的项目！！！

大佬的博客 ：Finley (cnblogs.com)

大佬的源码 ：github.com/hdt3213/god…

---

Redis的pubsub是指发布/订阅形式（Publish/Subscribe），是Redis供给的一种音讯传递形式。在该形式中，音讯的发送者（发布者）将音讯发布到一个特定的频道（Channel），而订阅者则能够订阅一个或多个频道，以接纳相应的音讯。

具体地说，发布者能够运用Redis的PUBLISH指令向指定的频道发布音讯，而订阅者能够运用SUBSCRIBE指令订阅一个或多个频道，并运用UNSUBSCRIBE指令撤销订阅。当有音讯被发布到一个现已被订阅的频道时，订阅者将会接纳到该音讯，并能够对其进行相应的处理。

Redis的pubsub形式能够用于完成实时音讯传递、即时通讯、音讯推送等功能，是Redis非常重要和常用的功能之一。

PubSub解说

pubsub包括订阅频道（channel）和订阅形式（pattern）,它们的首要区别在于：

1. 频道是一个简单的字符串，表明一个具体的音讯频道，而形式是一个包括通配符的字符串，表明多个音讯频道的调集。例如，频道能够是”news”或”chat”，而形式能够是”news. “或”chat. “，表明匹配以”news.”或”chat.”最初的一切频道。
2. 订阅频道时，只能接纳到该频道发布的音讯，而订阅形式时，能够接纳到一切与该形式匹配的频道发布的音讯。
3. 订阅频道时，运用SUBSCRIBE指令，而订阅形式时，运用PSUBSCRIBE指令。
4. 撤销订阅频道时，运用UNSUBSCRIBE指令，而撤销订阅形式时，运用PUNSUBSCRIBE指令。

需要留意的是，频道和形式是不行嵌套的，也就是说，不能在一个频道或形式中包括另一个频道或形式。同时，形式匹配是基于订阅时的形式字符串来匹配的，所以假如在订阅后新增了一个与形式匹配的频道，订阅者将无法接纳到该频道发布的音讯。

在miniRedis中暂时只完成订阅频道功能

运用PubSub

1. 创立订阅者衔接：运用Redis客户端，衔接到Redis服务器，并订阅指定的频道或形式.

例如订阅channel1 和 channel 2

SUBSCRIBE channel1 channel2

2. 创立发布者衔接：相同运用Redis客户端，衔接到Redis服务器，并向指定的频道或形式发布音讯，会回来消费音讯的订阅者数量。

例如发布”Hello miniRedis”到channel1中

PUBLISH channel1 "Hello, world!"

示例1：发布者发布音讯，但没有订阅者消费，回来0

示例2：发布者发布音讯，一个订阅者消费音讯。

订阅者初始状况：

发布者发布音讯，回来1

查看订阅者消费到的音讯内容

示例3：发布者发布音讯，多个订阅者消费音讯。

发布者发布音讯，回来2

查看订阅者1消费到的音讯内容

查看订阅者2消费到的音讯内容

示例4：多个发布者发布音讯，订阅者订阅多个频道。

订阅者初始状况：

发布者1发布音讯

发布者2发布音讯

查看订阅者消费到的音讯内容

PubSub的代码完成

定义订阅者和频道的键值对

pubsub/hub.go

// Hub stores all subscribe relations
type Hub struct {
   // 存储频道与订阅者的联系，键是频道，值是订阅者
   subs dict.Dict
   // 用于多协程同步读写频道内容
   subsLocker *lock.Locks
}
// MakeHub creates new hub
func MakeHub() *Hub {
   return &Hub{
      subs:       dict.MakeConcurrent(4),
      subsLocker: lock.Make(16),
   }
}

发布者发布音讯

1. 解析参数信息
2. 获取一切订阅了这个频道的订阅者
3. 给订阅者写入回来内容
4. 回来消费到音讯的订阅者数量

pubsub/pubsub.go

// Publish 客户端发送信息到频道中
func Publish(hub *Hub, args [][]byte) redis.Reply {
   if len(args) != 2 {
      return &protocol.ArgNumErrReply{Cmd: "publish"}
   }
   channel := string(args[0])
   message := args[1]
   hub.subsLocker.Lock(channel)
   defer hub.subsLocker.UnLock(channel)
   // 获取订阅者
   raw, ok := hub.subs.Get(channel)
   if !ok {
      return protocol.MakeIntReply(0)
   }
   subscribers, _ := raw.(*list.LinkedList)
   subscribers.ForEach(func(i int, c interface{}) bool {
      client, _ := c.(redis.Connection)
      replyArgs := make([][]byte, 3)
      replyArgs[0] = messageBytes    // "message"
      replyArgs[1] = []byte(channel) // "ch1"
      replyArgs[2] = message         // "message1"
      _, _ = client.Write(protocol.MakeMultiBulkReply(replyArgs).ToBytes())
      return true
   })
   // 回来的是接纳到的订阅者的数量
   return protocol.MakeIntReply(int64(subscribers.Len()))
}

订阅者消费音讯

1. 获取订阅的频道
2. 判别当时是否是新订阅的channel，是则创立并将当时客户端加入到订阅列表中并回来true
3. 假如不是，则判别当时客户端是否在channel的订阅列表中，不在则回来false
4. 写入客户端的回来中
5. 假如是新的则回来subscribe,订阅的频道，数量（如上图所示）
6. 假如不是新的则不做处理

pubsub/pubsub.go Subscribe办法

// Subscribe 将客户端加入到指定的通道中
func Subscribe(hub *Hub, c redis.Connection, args [][]byte) redis.Reply {
   channels := make([]string, len(args))
   for i, b := range args {
      channels[i] = string(b)
   }
   // 加锁保证同步性
   hub.subsLocker.Locks(channels...)
   defer hub.subsLocker.UnLocks(channels...)
   for _, channel := range channels {
      if subscribe0(hub, channel, c) {
         _, _ = c.Write(makeMsg(_subscribe, channel, int64(c.SubsCount())))
      }
   }
   return &protocol.NoReply{}
}

pubsub/pubsub.go Subscribe0办法

// subscribe0 客户端订阅channel，假如是一个新的channel则回来true
func subscribe0(hub *Hub, channel string, client redis.Connection) bool {
   client.Subscribe(channel)
   // 从字典中获取这个channel对应的订阅者
   raw, ok := hub.subs.Get(channel)
   var subscribers *list.LinkedList
   if ok {
      // 假如存在该channel
      subscribers, _ = raw.(*list.LinkedList)
   } else {
      // 假如不存在该channel
      subscribers = list.Make()
      hub.subs.Put(channel, subscribers)
   }
   // 判别得到的订阅者列表是否包括当时订阅者（假如是新的列表必定不包括）
   if subscribers.Contains(func(a interface{}) bool {
      return a == client
   }) {
      return false
   }
   subscribers.Add(client)
   return true
}

订阅者撤销订阅频道

1. 从订阅列表中移除当时客户端
2. 假如订阅列表移除之后为空，则移除这个频道

pubsub/pubsub.go unsubscribe0办法

func unsubscribe0(hub *Hub, channel string, client redis.Connection) bool {
   client.UnSubscribe(channel)
   // remove from hub.subs
   raw, ok := hub.subs.Get(channel)
   if ok {
      subscribers, _ := raw.(*list.LinkedList)
      subscribers.RemoveAllByVal(func(a interface{}) bool {
         return utils.Equals(a, client)
      })
      if subscribers.Len() == 0 {
         // clean
         hub.subs.Remove(channel)
      }
      return true
   }
   return false
}