Godis_发布订阅

什么是发布订阅

发布/订阅（Publish/subscribe 或 pub/sub）是一种消息范式，消息的发送者（发布者）不是计划发送其消息给特定的接收者（订阅者）。 而是发布的消息分为不同的类别，而不需要知道什么样的订阅者订阅。 订阅者对一个或多个类别表达兴趣，于是只接收感兴趣的消息，而不需要知道什么样的发布者发布的消息。

这个与设计模式中的观察者模式是一个意思。

有一些是可以在程序里面实现发布订阅，而业界就针对这个模式开发出了中间件，也就是消息队列。现在成熟的消息队列就有 Kafak、RabbitMQ、ActiveMQ、RocketMQ 等。

Redis 发布订阅的使用

Redis 客户端可以订阅任意数量的频道。

下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：

当有新消息通过 PUBLISH 命令发送给频道 channel1 时， 这个消息就会被发送给订阅它的三个客户端：

Redis 发布订阅模式的命令使用：

1）发布消息

客户端：

1
2
3
4
5
6
7
8
9
// 使用redis-cli工具连接到redis
redis-cli
127.0.0.1:6379>
// 订阅`redis` 主题的消息
127.0.0.1:6379> SUBSCRIBE redis
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redis"
3) (integer) 1

服务端：

1
2
3
4
5
// 使用redis-cli工具连接到redis
redis-cli
127.0.0.1:6379>
127.0.0.1:6379> PUBLISH redis test
(integer) 1

服务端向 redis 主题发布了一条 test 消息，显示发送成功的客户端数量，这里只有一个客户端订阅了 redis 这个主题，所以在发送成功后就显示数量 1。

发布了之后再看看客户端的情况

1
2
3
4
5
6
7
8
9
127.0.0.1:6379> SUBSCRIBE redis
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redis"
3) (integer) 1
// 下面是收到的消息
5) "message"
6) "redis"
7) "test"

从上面消息可以看出，除了订阅主题消息成功的提示外，还收到了一条新消息，从头到尾分别是：

  • 第一行：消息类型为消息，这里是 message
  • 第二行：消息主题名，这里是 redis
  • 实际的消息：这里是 test

2）取消订阅

取消订阅也很简单，直接使用 UNSUBSCRIBE 命令即可

语法：

1
UNSUBSCRIBE [channel [channel ...]]

支持取消多个通道

3）支持模式匹配

给客户端订阅指定模式的通道，以 * 号字作为匹配符。

支持的 glob 样式模式：

  • h?llo 订阅了 hello、hallo 和 hxllo

  • h*llo订阅 hllo 和 heeeello

  • h〔ae〕llo订阅 hello 和 hallo，但不订阅 hillo

语法：

1
PSUBSCRIBE pattern [pattern ...]

Godis 的实现

先在代码中找到这几个命令的实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
database/database.go 79 line

// Exec executes command
// parameter `cmdLine` contains command and its arguments, for example: "set key value"
func (mdb *MultiDB) Exec(c redis.Connection, cmdLine [][]byte) (result redis.Reply) {
	//.....
	//....

   // special commands which cannot execute within transaction
   if cmdName == "subscribe" {
      if len(cmdLine) < 2 {
         return protocol.MakeArgNumErrReply("subscribe")
      }
      return pubsub.Subscribe(mdb.hub, c, cmdLine[1:])
   } else if cmdName == "publish" {
      return pubsub.Publish(mdb.hub, cmdLine[1:])
   } else if cmdName == "unsubscribe" {
      return pubsub.UnSubscribe(mdb.hub, c, cmdLine[1:])
   }
   //省略其他代码
   // .....
}

可以看到 Godis 支持了 subscribe、publish 和 unsubscribe 命令，根据模式匹配主题这个并没有实现。然后对应命令都会跳转到 pubsub 包的对应函数。

subscribe

如果传入的命令是 subscribe 的话，则会调用 pubsub 包的 Subscribe 函数，该函数要求传入一个 Hub 类型、redis.Connection 连接类型、args 参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
// pubsub/pubsub.go 68 line

// Subscribe puts the given connection into the given channel
func Subscribe(hub *Hub, c redis.Connection, args [][]byte) redis.Reply {
   // 声明存放主题的切片
   channels := make([]string, len(args))
   // 将主题放到channels切片里面
   for i, b := range args {
      channels[i] = string(b)
   }

   // 将这些主题所对应的分片上锁
   hub.subsLocker.Locks(channels...)
   defer hub.subsLocker.UnLocks(channels...)

   for _, channel := range channels {
	  // 对每一个主题调用subscribe0函数
      if subscribe0(hub, channel, c) {
	     // 成功了就将消息格式化成RESP协议返回给调用者
         _ = c.Write(makeMsg(_subscribe, channel, int64(c.SubsCount())))
      }
   }
   return &protocol.NoReply{}
}


// 对每一个主题都进行操作的函数
func subscribe0(hub *Hub, channel string, client redis.Connection) bool {
   client.Subscribe(channel)

   // add into hub.subs
   raw, ok := hub.subs.Get(channel)
   var subscribers *list.LinkedList
   // 看这个主题是否有对应的list存在
   if ok {
	  // 存在则进行类型断言
      subscribers, _ = raw.(*list.LinkedList)
   } else {
	  // 不存在则创建一个新的list并存入得到hub的字典里
      subscribers = list.Make()
      hub.subs.Put(channel, subscribers)
   }
   // 如果该list已经存在了这个连接，返回false
   if subscribers.Contains(client) {
      return false
   }
   // list不存在这个连接，则把当前连接添加到list里面并返回true
   subscribers.Add(client)
   return true
}

在上面的流程可以看出，做了以下步骤：

  • 获取到对应的主题切片
  • 然后对这个主题切片进行上锁，保证并发安全
  • 然后对每一个主题调用 subscribe0 函数
    • 在 hub 对象根据主题名获取一个 list
    • 如果该 list 存在则代表该主题在之前有其他客户端订阅过，如果不存在则代表该主题是第一次被人订阅，然后创建一个新的 list 并存放到 hub 对象中
    • 判断该 list 是否包含了当前连接
      • 如果存在了直接返回 false
      • 否则把当前连接添加到该 list 中然后返回 true
  • 根据 subscribe0 函数的返回结果在返回 RESP 协议内容给调用者。

这一步其实也就是将连接信息存放到一个 list，该 list 又会作为 value 被存放到一个 map 中，这个 map 以主题为 key。这样在发布消息的时候就会以主题去 map 中查找，返回对应的 value，也就是一个 list。

然后遍历这个 list，把消息发给对应的连接即可。

unsubscribe

与 subscribe 流程大致相同，只不过把添加连接到 list 的步骤换成了删除

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
// pubsub/pubsub.go 100 line

// UnSubscribe removes the given connection from the given channel
func UnSubscribe(db *Hub, c redis.Connection, args [][]byte) redis.Reply {
   var channels []string
   if len(args) > 0 {
      channels = make([]string, len(args))
      for i, b := range args {
         channels[i] = string(b)
      }
   } else {
      channels = c.GetChannels()
   }

   db.subsLocker.Locks(channels...)
   defer db.subsLocker.UnLocks(channels...)

   if len(channels) == 0 {
      _ = c.Write(unSubscribeNothing)
      return &protocol.NoReply{}
   }

   for _, channel := range channels {
	  // 针对每一个主题调用unscribe0函数
	  // 如果成功则将构造返回消息写入到连接中
      if unsubscribe0(db, channel, c) {
         _ = c.Write(makeMsg(_unsubscribe, channel, int64(c.SubsCount())))
      }
   }
   return &protocol.NoReply{}
}


func unsubscribe0(hub *Hub, channel string, client redis.Connection) bool {
   client.UnSubscribe(channel)

   // remove from hub.subs
   // 获取主题对应的list
   raw, ok := hub.subs.Get(channel)
   if ok {
      subscribers, _ := raw.(*list.LinkedList)
      // 在list中删除这个连接信息
      subscribers.RemoveAllByVal(client)

	  // 如果删除完连接信息之后，list的长度==0的话，则在map中删除这个主题所对应的kev/val信息
      if subscribers.Len() == 0 {
         // clean
         hub.subs.Remove(channel)
      }
      return true
   }
   return false
}

publish

与 subscribe 流程大致相同

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

var messageBytes       = []byte("message")

// pubsub/pubsub.go 127 line

// Publish send msg to all subscribing client
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)

   // 获取list
   raw, ok := hub.subs.Get(channel)
   if !ok {
      return protocol.MakeIntReply(0)
   }
   subscribers, _ := raw.(*list.LinkedList)

   // 遍历这个list，然后将消息类型、主题名以及具体的消息构造并写入到连接当中
   subscribers.ForEach(func(i int, c interface{}) bool {
      client, _ := c.(redis.Connection)
      replyArgs := make([][]byte, 3)
      replyArgs[0] = messageBytes
      replyArgs[1] = []byte(channel)
      replyArgs[2] = message
      _ = client.Write(protocol.MakeMultiBulkReply(replyArgs).ToBytes())
      return true
   })
   return protocol.MakeIntReply(int64(subscribers.Len()))
}

总结

在阅读代码的时候就可以看出来，Godis 是用了一个专门的数据结构 Hub 去维护这些订阅关系，这个数据结构有 2 部分

  • map：key 是主题名称，val 是一个链表，该链表存放的元素类型是当前的 tcp 连接
  • lock：这个锁是用来所住主题名，保证并发安全

在订阅时，会根据主题名查找 map，如果没有找到则将初始化一个新的链表，并将当前连接信息存入到链表中。最后把链表作为 val 存放到 map 中

而在发布时，则根据主题名查找 map，获得主题对应的链表，如果链表为空则直接返回。否则遍历这个链表，把发布的消息写入到链表元素中。

源码阅读
#Go #第三方库
Godis_发布订阅
https://sunshinechen.cn/2023/06/05/Godis_发布订阅/
作者
sunshine
发布于
2023年6月5日
许可协议