nsq 之 nsqlookupd 源码解析

nsqlookupd是nsq管理集群拓扑信息以及用于注册和发现nsqd服务；所以，也可以把nsqlookupd理解为注册发现服务；当nsq集群中有多个nsqlookupd服务时，因为每个nsqd都会向所有的nsqlookupd上报本地信息，因此nsqlookupd具有最终一致性。

重点数据结构

• NSQLookupd

  NSQLookupd 是nsqlookupd的核心数据结构，它包含了nsqlookupd 启动和服务的核心配置和组件。

  type NSQLookupd struct {
     sync.RWMutex
     // 配置项
     opts         *Options
     // TCP listener
     tcpListener  net.Listener
     // HTTP listener
     httpListener net.Listener
     // TCP handler
     tcpServer    *tcpServer
     waitGroup    util.WaitGroupWrapper
    // 存储topic、channel对应的生产者信息，可以看作是一个注册中心
     DB           *RegistrationDB
  }

  源代码

• Registration

  Registration 相关可以看作类似注册中心，负责注册记录各个nsqd实例上的topic、channel信息。

  // RegistrationDB
  type RegistrationDB struct {
     sync.RWMutex
     // registrationMap 记录了每个topic/channel 对应的一组生产者
     registrationMap map[Registration]ProducerMap
  }
  // RegistrationDB.AddRegistration 添加一个Registration
  // RegistrationDB.AddProducer 为一个Registration 添加一个producer
  // RegistrationDB.RemoveProducer 为一个Registration 移除一个producer
  // RegistrationDB.RemoveRegistration 移除一个Registration以及它的所有producer
  // RegistrationDB.FindRegistrations 根据 Category、key、subKey找到一组对应的Registration
  // RegistrationDB.FindProducers 根据 Category、key、subKey找到一组对应的producer
  // RegistrationDB.LookupRegistrations 根据producer peerInfo.id 找到其对应的Registrations
  
  // Registration 唯一标识一个消费单位（topic级别或者channel级别）
  type Registration struct {
      // category 表明该Registration是topic级别还是channel级别
      Category string
      // topic name
      Key      string
      // 如果Category 是channel则subKey 是channel name， 如果Category是topic则SubKey为空
      SubKey   string
  }
  
  type ProducerMap map[string]*Producer
  
  type Producer struct {
      peerInfo     *PeerInfo
  }
  
  // 记录prodcer的信息
  type PeerInfo struct {
      lastUpdate       int64
      id               string
      RemoteAddress    string `json:"remote_address"`
      Hostname         string `json:"hostname"`
      BroadcastAddress string `json:"broadcast_address"`
      TCPPort          int    `json:"tcp_port"`
      HTTPPort         int    `json:"http_port"`
      Version          string `json:"version"`
  }
  
  type Registrations []Registration

  源代码

源码分析流程

• NSQLookupd.Main

  nsqLookupd启动主流程入口

  func (l *NSQLookupd) Main() error {
     ctx := &Context{l}
  
     exitCh := make(chan error)
     var once sync.Once
     exitFunc := func(err error) {
        once.Do(func() {
           if err != nil {
              l.logf(LOG_FATAL, "%s", err)
           }
           exitCh <- err
        })
     }
  
     l.tcpServer = &tcpServer{ctx: ctx}
     // 启动TCPServer 监听响应nsqd register/unregister/ping 请求
     l.waitGroup.Wrap(func() {
        exitFunc(protocol.TCPServer(l.tcpListener, l.tcpServer, l.logf))
     })
     // 启动HTTPServer 监听响应HTTP 请求（主要是topic/channel 的管理以及一些供debug 的信息）
     httpServer := newHTTPServer(ctx)
     l.waitGroup.Wrap(func() {
        exitFunc(http_api.Serve(l.httpListener, httpServer, "HTTP", l.logf))
     })
  
     err := <-exitCh
     return err
  }

  源代码

• nsqlookupd TCPServer

  TCPServer 负责监听和处理TCP请求。

  func TCPServer(listener net.Listener, handler TCPHandler, logf lg.AppLogFunc) error {
     logf(lg.INFO, "TCP: listening on %s", listener.Addr())
  
     var wg sync.WaitGroup
  
     for {
        // accept TCP conn
        clientConn, err := listener.Accept()
        if err != nil {
           if nerr, ok := err.(net.Error); ok && nerr.Temporary() {
              logf(lg.WARN, "temporary Accept() failure - %s", err)
              runtime.Gosched()
              continue
           }
           // theres no direct way to detect this error because it is not exposed
           if !strings.Contains(err.Error(), "use of closed network connection") {
              return fmt.Errorf("listener.Accept() error - %s", err)
           }
           break
        }
  
        wg.Add(1)
        go func() {
           // go TCP handler.Handle
           handler.Handle(clientConn)
           wg.Done()
        }()
     }
  
     // wait to return until all handler goroutines complete
     wg.Wait()
  
     logf(lg.INFO, "TCP: closing %s", listener.Addr())
  
     return nil
  }

  源代码

  这个TCPServer 函数和nsqd 的是共用的，但handler 不是同一个handler，nsqlookupd TCP Handle 方法如下：

  func (p *tcpServer) Handle(clientConn net.Conn) {
     p.ctx.nsqlookupd.logf(LOG_INFO, "TCP: new client(%s)", clientConn.RemoteAddr())
  
     /**
     读取并解析 protocolMagic
      */
     buf := make([]byte, 4)
     _, err := io.ReadFull(clientConn, buf)
     if err != nil {
        clientConn.Close()
        return
     }
     protocolMagic := string(buf)
  
     var prot protocol.Protocol
     switch protocolMagic {
     case "  V1":
        prot = &LookupProtocolV1{ctx: p.ctx}
     default: // 返回协议错误
        protocol.SendResponse(clientConn, []byte("E_BAD_PROTOCOL"))
        clientConn.Close()
        return
     }
  
     p.conns.Store(clientConn.RemoteAddr(), clientConn)
  
     // LookupProtocolV1.IOLoop 处理请求
     err = prot.IOLoop(clientConn)
     if err != nil {
        p.ctx.nsqlookupd.logf(LOG_ERROR, "client(%s) - %s", clien