nsqlookupd 源码分析（2）- topic通道注册，查询代码分析
上篇文章介绍了一下nsqlookupd 启动服务的简单流程，限于篇幅到这里来写一下nsqlookupd主要的工作：通道注册登记，以及查询服务。
nsqlookupd 有两个主要的功能：
- nsqd来注册所有topic、channel的分布信息，nsqlookupd记录哪些topic在哪些机器上；
- 给消费端来查询对应的topic所在的机器列表，以供订阅消费；
下面先来讲一下TCP协议端，nsqd是怎么告诉nsqlookupd,他有个新的topic创建的呢？ 这就得看 REGISTER操作了。
一、REGISTER 注册topic
1.nsqd端发送 REGISTER 命令 等级topic、channel
之前我们了解到，当nsqd创建一个topic的时候，NewTopic函数会调用nsqd.Notify(t) 来通知lookupd有新的topic产生了。 后者实际上通过channel通知lookupLoop协程进行处理的，处理代码：
case val := <-n.notifyChan: //通知chanel， 一般有topic，channel上的增删的时候会写入管道, 进而通知其他lookup去处理
var cmd *nsq.Command
var branch string
switch val.(type) {
case *Channel:
// notify all nsqlookupds that a new channel exists, or that it's removed
branch = "channel"
channel := val.(*Channel)
if channel.Exiting() == true {
cmd = nsq.UnRegister(channel.topicName, channel.name)
} else {//生成对应的指令
cmd = nsq.Register(channel.topicName, channel.name)
}
}
所以如果nsqd上新增了一个topic， 那么就会调用nsq.Register（） 生成一条REGISTER 命令： &Command{[]byte("REGISTER"), params, nil}，之后遍历所有的lookupPeers，给他们发送注册指令：
//循环在所有的lookup上执行这条指令
for _, lookupPeer := range lookupPeers {
n.logf(LOG_INFO, "LOOKUPD(%s): %s %s", lookupPeer, branch, cmd)
_, err := lookupPeer.Command(cmd)
if err != nil {
n.logf(LOG_ERROR, "LOOKUPD(%s): %s - %s", lookupPeer, cmd, err)
}
}
所以这里注册指令的来源比较清楚了。下面开始了解一下nsqlookupd是怎么处理的。
nsqlookupd 登记topic、channel对应的nsqd列表
nsqlookupd的TCP协议处理函数实际是LookupProtocolV1::IOLoop, 这函数比较简单，主要就是调用了Exec函数，后者很简单，一共4个命令，我们只关注 REGISTER， 其他都比较简单或者类似：
func (p *LookupProtocolV1) Exec(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
switch params[0] {
case "PING":
return p.PING(client, params)
case "IDENTIFY":
return p.IDENTIFY(client, reader, params[1:])
case "REGISTER":
return p.REGISTER(client, reader, params[1:])
case "UNREGISTER":
return p.UNREGISTER(client, reader, params[1:])
}
return nil, protocol.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("invalid command %s", params[0]))
}
下面直接看REGISTER 函数， 函数首先用getTopicChan 读取nsqd发送过来的topic和channel， 然后根据是否有channel来进行对应的处理：如果有channel，就需要记录(topic,channel)对，否则只需要记录topic就行了，这里数据结构是共用的.
对于topic,channel 到 nsqd的映射，之前的文章“nsqlookupd 源码分析（1）- 启动服务” 里面讲过 RegistrationDB的结构，这里参考来看会比较直接。
func (p *LookupProtocolV1) REGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
//nsqd发送注册登记指令REGISTER， 来通知lookup他新建了一个topic和channel
//nsqlookupd需要记录topic对应的nsqd列表，以备用来查询等
topic, channel, err := getTopicChan("REGISTER", params)
if channel != "" {//如果有channel， 需要单独记录一下channel，因为topic和channel可以1:n的
key := Registration{"channel", topic, channel}
//调用AddProducer 将映射关系放入map里面
p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) ；
}
key := Registration{"topic", topic, ""}
p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) ；
return []byte("OK"), nil
}
AddProducer 比较简单，就是对registrationMap 的查询，如果当前这个客户端以及在映射表里面就不需要处理，否则就加进去。
func (r *RegistrationDB) AddProducer(k Registration, p *Producer) bool {
//将一个nsqd加入到k参数指定的topic、channel映射表里面，先查是否已经在里面了，没有就加入
r.Lock()
defer r.Unlock()
producers := r.registrationMap[k]
found := false
for _, producer := range producers {
if producer.peerInfo.id == p.peerInfo.id {
found = true
break
}
}
if found == false {
r.registrationMap[k] = append(producers, p)
}
return !found
}
UNREGISTER 注销也是类似的，当nsqd退出，或者topic删除了的话，会调用UNREGISTER 命令，类似REGISTER处理就行了。
二、“/lookup” 查询topic所在机器列表以及channel列表
查询方法
当消费者需要查询某个topic在哪些nsqd上时，可以发送/lookup 的http命令查询，nsqlookupd的查询topic信息的接口是这样的：
curl 'http://127.0.0.1:4161/lookup?topic=testtopic'
返回如下：
可以看到，/lookup命令会返回该topic对应的channel列表，以及都有哪些nsqd上面有这个topic。下面来看看代码：
/lookup 方法原理
/lookup指令的处理函数是 doLookup， 这是在Main的 newHTTPServer调用上设置的：
func newHTTPServer(ctx *Context) *httpServer {
router.Handle("GET", "/lookup", http_api.Decorate(s.doLookup, log, http_api.V1))
router.Handle("GET", "/topics", http_api.Decorate(s.doTopics, log, http_api.V1))
router.Handle("GET", "/channels", http_api.Decorate(s.doChannels, log, http_api.V1))
router.Handle("GET", "/nodes", http_api.Decorate(s.doNodes, log, http_api.V1))
router.Handle("POST", "/topic/create", http_api.Decorate(s.doCreateTopic, log, http_api.V1))
router.Handle("POST", "/topic/delete", http_api.Decorate(s.doDeleteTopic, log, http_api.V1))
}
doLookup函数先获取topic参数，然后查找nsqlookupd.DB 对应的topic的信息，如果没有查到就返回404，否则返回对应的json信息：
func (s *httpServer) doLookup(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
//lookup命令会返回该topic对应的channel列表，以及都有哪些nsqd上面有这个topic
reqParams, err := http_api.NewReqParams(req)
topicName, err := reqParams.Get("topic")
//先查询map里面是否有对应topic的记录
registration := s.ctx.nsqlookupd.DB.FindRegistrations("topic", topicName, "")
//得到所有的channel列表
channels := s.ctx.nsqlookupd.DB.FindRegistrations("channel", topicName, "*").SubKeys()
//得到 有这个topic的nsqd列表，后面需要去掉不活跃的机器
producers := s.ctx.nsqlookupd.DB.FindProducers("topic", topicName, "")
producers = producers.FilterByActive(s.ctx.nsqlookupd.opts.InactiveProducerTimeout,
s.ctx.nsqlookupd.opts.TombstoneLifetime)
//返回json字符串
return map[string]interface{}{
"channels": channels,
"producers": producers.PeerInfo(),
}, nil
}
调用的FindRegistrations 和 FindProducers 函数实现都在/nsqlookupd/registration_db.go 里面，代码比较简单，基本上就是简单匹配，以及map操作，这里不多说了。
三、 创建一个topic
顺便讲一下nsqlookupd上的创建一个topic是怎么操作的，会发生什么。 同样是HTTP接口，调用方法为“/topic/create”， 实现函数是doCreateTopic ， 比较简单，其实就是在lookupd上增加了一条topic的记录，并且他的producers为空：
func (s *httpServer) doCreateTopic(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
//创建一个topic， 只是简单的记录了一下topic信息，并且加到DB里面，对应的nsqd数组为空
reqParams, err := http_api.NewReqParams(req)
topicName, err := reqParams.Get("topic")
key := Registration{"topic", topicName, ""}
s.ctx.nsqlookupd.DB.AddRegistration(key)
return nil, nil
}
func (r *RegistrationDB) AddRegistration(k Registration) {
r.Lock()
defer r.Unlock()
_, ok := r.registrationMap[k]
if !ok {
r.registrationMap[k] = Producers{}
}
}
到这里nsqlookupd 的主要操作：登记topic和查询topic的代码都介绍了一下，比较简单，并且没有持久化，他的持久化其实是借用的nsqd的持久化，因为所有nsqd启动的时候，都会立即链接nsqlookupd，然后注册对应的topic， 所以nsqlookupd 没必要进行持久化操作了。
另外在实际应用中，nsqlookupd可以多设置几个也是支持的，nsqd可以连接多个nsqlookupd服务，同时注册，灵活性比较高。
nsqlookupd 代码算是最简单的web服务的样子，简单的每个client一个go协程进行处理，内存map记录topic->nsqd 的映射关系，http提供查询，增删改操作。
近期评论