传送门：使用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
}

作者：山腰间飘逸的红雨
链接：https://juejin.cn/post/7225124223098339385