主题名是大小写敏感的,必须是非空字符串,不能包含空格,使用“.”符号来分层,mqtt中使用“/”分层。
星号“*”匹配一层,跟mqtt中的“+”一样。
大于号“>”匹配后面所有层,跟mqtt中的“#”一样。
代码都在sublist.go
。
先列出数据结构:
// A Sublist stores and efficiently retrieves subscriptions.
type Sublist struct {
sync.RWMutex
genid uint64
matches uint64
cacheHits uint64
inserts uint64
removes uint64
cache map[string]*SublistResult
root *level
count uint32
}
// A level represents a group of nodes and special pointers to
// wildcard nodes.
type level struct {
nodes map[string]*node
pwc, fwc *node //pwc代表'*'节点,fwc代表'>'节点
}
// A node contains subscriptions and a pointer to the next level.
type node struct {
next *level
psubs []*subscription //普通订阅者列表
qsubs [][]*subscription //queue订阅者列表
}
// New will create a default sublist
func NewSublist() *Sublist {
return &Sublist{root: newLevel(), cache: make(map[string]*SublistResult)}
}
// Create a new default level. We use FNV1A as the hash
// algorithm for the tokens, which should be short.
//FNV1A?历史遗留注释吧,这儿分明直接用了golang自带的map哈希。
func newLevel() *level {
return &level{nodes: make(map[string]*node)}
}
一开始使用NewSublist创建一个Sublist。Sublist保存了所有subject。
Sublist初始化时创建了root节点。
来看看怎么插入一个subject:
// Insert adds a subscription into the sublist
func (s *Sublist) Insert(sub *subscription) error {
// copy the subject since we hold this and this might be part of a large byte slice.
subject := string(sub.subject)
tsa := [32]string{}
tokens := tsa[:0]
start := 0
for i := 0; i < len(subject); i++ {
if subject[i] == btsep {
tokens = append(tokens, subject[start:i])
start = i + 1
}
}
tokens = append(tokens, subject[start:])
s.Lock()
sfwc := false
l := s.root
var n *node
for _, t := range tokens {
lt := len(t)
if lt == 0 || sfwc { //如果此层长度为0或者上一层已经是'>'了,表示Subject是非法的
s.Unlock()
return ErrInvalidSubject
}
if lt > 1 { //不是*和>,直接map定位
n = l.nodes[t]
} else {
switch t[0] {
case pwc:
n = l.pwc
case fwc:
n = l.fwc
sfwc = true //表示此层只能是最后一层
default: //不是*和>,直接map定位
n = l.nodes[t]
}
}
if n == nil { //node节点还没有则创建
n = newNode()
if lt > 1 {
l.nodes[t] = n
} else {
switch t[0] {
case pwc:
l.pwc = n
case fwc:
l.fwc = n
default:
l.nodes[t] = n
}
}
}
if n.next == nil {
n.next = newLevel()
}
l = n.next //下一层
}
//上面循环结束后此时n是最后一层的node节点
if sub.queue == nil { //不是queue,把sub加到psubs中。psubs切片存储了所有订阅此subject的client
n.psubs = append(n.psubs, sub)
} else {
// This is a queue subscription
if i := findQSliceForSub(sub, n.qsubs); i >= 0 {
n.qsubs[i] = append(n.qsubs[i], sub)
} else {
n.qsubs = append(n.qsubs, []*subscription{sub})
}
}
s.count++
s.inserts++
s.addToCache(subject, sub)
atomic.AddUint64(&s.genid, 1)
s.Unlock()
return nil
}
从Insert方法中可以理出整个数据结构:
Sublist第一个层节点是root,root是个level结构,level代表一层。level包含了一个nodes map,nodes存储了此层的所有node,pwc和fwc分别代表了*和> node 。
node的next指向了下一层level,node的psubs存储了普通subject订阅者client,qsubs存储的是queue类别的subject订阅者client。
整个list是个树结构,只不过每层的node节点使用map哈希存储。
先不管Cache干什么用的,先来看看查找匹配:
// matchLevel is used to recursively descend into the trie.
func matchLevel(l *level, toks []string, results *SublistResult) {
var pwc, n *node
for i, t := range toks {
if l == nil {
return
}
if l.fwc != nil { //全匹配,把下面的所有订阅者都加入到results中
addNodeToResults(l.fwc, results)
}
if pwc = l.pwc; pwc != nil { //层匹配,递归子层
matchLevel(pwc.next, toks[i+1:], results)
}
n = l.nodes[t] //查找节点
if n != nil { //找到继续下一层
l = n.next
} else {
l = nil
}
}
if n != nil { //找到节点,把订阅者加入到results中
addNodeToResults(n, results)
}
if pwc != nil { //最后一层*通配符的订阅者加入到results中
addNodeToResults(pwc, results)
}
}
从指定level比如root开始遍历匹配下面的每一层nodes,如果匹配则把订阅者加入到result中,注意通配符的处理。
每一层使用map快速定位node,使用切片存储此层所有订阅者。
因为使用了map,查询定位比mosquitto的遍历链表树快的多。何况qnatsd还做了Cache。
下面再来看看Cache:
因为查找一个subject的所有订阅者比较费时间,所以使用cache缓存一部分subject订阅者信息,每次查找先去cache中查找,找不到再去sublist中查找,如果找到就加入到cache中,新增一个subject时也要加入到cache中。
cache也有数量限制,当超过一定数量时删除最早的部分cache,防止cache过多。