剖析nsq消息队列(二) 去中心化源码解析

时间 2019-11-06

标签剖析 nsq 消息队列中心源码解析繁體版

原文原文链接

在上一篇帖子剖析nsq消息队列(一) 简介及去中心化实现原理中，我介绍了nsq的两种使用方式，一种是直接链接，还有一种是经过nslookup来实现去中心化的方式使用，并大概说了一下实现原理，没有什么难理解的东西，这篇帖子我把nsq实现去中心化的源码和其中的业物逻辑展现给你们看一下。html

nsqd和nsqlookupd的通讯实现

上一篇中在启动nsqd时我用了如下命令，我指定了一个参数 --lookupd-tcp-addresssql

./nsqd -tcp-address ":8000"  -http-address ":8001" --lookupd-tcp-address=127.0.0.1:8200 --lookupd-tcp-address=127.0.0.1:7200 -data-path=./a

--lookupd-tcp-address 用于指定nsqlookupd的tcp监听地址。json

nsqd 和 nsqlookupd的通讯交流简单来讲就是下图这样
api

nsqd启动后链接nsqlookupd,链接成功后，要发送一个魔法标识nsq.MagicV1，这个标识有啥魔法么，固然不是，他只是用于标明，客户端和服务端双方使用的信息通讯版本，不能的版本有不一样的处理方式，为了后期作新的消息处理版本方便吧。
nsqlookupd 的代码块app

func (p *tcpServer) Handle(clientConn net.Conn) {   
    // ...
    buf := make([]byte, 4)
    _, err := io.ReadFull(clientConn, buf)
    // ...
    protocolMagic := string(buf)
    // ...
    var prot protocol.Protocol
    switch protocolMagic {
    case "  V1":
        prot = &LookupProtocolV1{ctx: p.ctx}
    default:
        // ...
        return
    }
    err = prot.IOLoop(clientConn)
    //...
}

这个时候的nsqd已经和nsqlookupd创建好了链接，可是这时，仅仅说明他俩链接成功。
nsqlookupd也并无把这个链接加到可用的nsqd列表里。
创建链接完成后，nsqd会发送IDENTIFY命令，这个命令里包含了nsq的基本信息
nsqd的代码tcp

ci := make(map[string]interface{})
        ci["version"] = version.Binary
        ci["tcp_port"] = n.RealTCPAddr().Port
        ci["http_port"] = n.RealHTTPAddr().Port
        ci["hostname"] = hostname
        ci["broadcast_address"] = n.getOpts().BroadcastAddress

        cmd, err := nsq.Identify(ci)
        if err != nil {
            lp.Close()
            return
        }
        resp, err := lp.Command(cmd)

包含了nsqd 提供的tcp和http端口，主机名，版本等等，发送给nsqlookupd,nsqlookupd收到IDENTIFY命令后，解析信息而后加到nsqd的可用列表里
nsqlookupd 的代码块oop

func (p *LookupProtocolV1) IDENTIFY(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
    var err error
    if client.peerInfo != nil {
        return nil, protocol.NewFatalClientErr(err, "E_INVALID", "cannot IDENTIFY again")
    }
    var bodyLen int32
    err = binary.Read(reader, binary.BigEndian, &bodyLen)
    // ...
    body := make([]byte, bodyLen)
    _, err = io.ReadFull(reader, body)
    // ...  
    peerInfo := PeerInfo{id: client.RemoteAddr().String()}
    err = json.Unmarshal(body, &peerInfo)
    // ...
    client.peerInfo = &peerInfo
    // 把nsqd的链接加入到可用列表里    
    if p.ctx.nsqlookupd.DB.AddProducer(Registration{"client", "", ""}, &Producer{peerInfo: client.peerInfo}) {
        p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) REGISTER category:%s key:%s subkey:%s", client, "client", "", "")
    }
    // ...
    return response, nil
}

而后每过15秒，会发送一个PING心跳命令给nsqlookupd，这样保持存活状态，nsqlookupd每次收到发过来的PING命令后，也会记下这个nsqd的最后更新时间，这样作为一个筛选条件，若是长时间没有更新，就认为这个节点有问题，不会把这个节点的信息加入到可用列表。
到此为止，一个nsqd就把本身的信息注册到nsqlookupd的可用列表了，咱们能够启动多个nsqd和多个nsqlookupd，为nsqd
指定多个nsqlookupd，就如同我上一篇帖子写的那样this

--lookupd-tcp-address=127.0.0.1:8200 --lookupd-tcp-address=127.0.0.1:7200

nsqd和全部的nsqlookupd创建链接，注册服务信息，并保持心跳，保证可用列表的更新.url

nsqlookupd 挂掉的处理方式

上面咱们说了nsqd若是出现问题，nsqlookupd的nsqd可用列表里就会处理掉这个链接信息。如nsqlookupd挂了怎么办呢

目前的处理方式是这样的，
不管是心跳，仍是其余命令，nsqd会给全部的nsqlookup发送信息，当nsqd发现nsqlookupd出现问题时，在每次发送命令时，会不断的进行从新链接:3d

func (lp *lookupPeer) Command(cmd *nsq.Command) ([]byte, error) {
    initialState := lp.state
    if lp.state != stateConnected {
        err := lp.Connect()
        if err != nil {
            return nil, err
        }
        lp.state = stateConnected
        _, err = lp.Write(nsq.MagicV1)
        if err != nil {
            lp.Close()
            return nil, err
        }
        if initialState == stateDisconnected {
            lp.connectCallback(lp)
        }
        if lp.state != stateConnected {
            return nil, fmt.Errorf("lookupPeer connectCallback() failed")
        }
    }
    // ...
}

若是链接成功，会再次调用connectCallback方法，进行IDENTIFY命令的调用等。

客户端和nsqlookupd、nsqd的通讯实现

上一篇帖子里介绍了，客户端如何链接nsqlookupd来进行通讯

adds := []string{"127.0.0.1:7201", "127.0.0.1:8201"}
    config := nsq.NewConfig()
    config.MaxInFlight = 1000
    config.MaxBackoffDuration = 5 * time.Second
    config.DialTimeout = 10 * time.Second

    topicName := "testTopic1"
    c, _ := nsq.NewConsumer(topicName, "ch1", config)
    testHandler := &MyTestHandler{consumer: c}

    c.AddHandler(testHandler)
    if err := c.ConnectToNSQLookupds(adds); err != nil {
        panic(err)
    }

须要注意adds里地址的端口，是nsqlookupd的http端口
这里我还使用上一篇帖子中的图，给你们详细分析

调用方法c.ConnectToNSQLookupds(adds)，他的实现是访问nsqlookupd的http端口http://127.0.0.1:7201/lookup?topic=testTopic1获得提供consumer订阅的topic全部的producers节点信息， url返回的数据信息以下。

{
  "channels": [
    "nsq_to_file",
    "ch1"
  ],
  "producers": [
    {
      "remote_address": "127.0.0.1:58606",
      "hostname": "li-peng-mc-macbook.local",
      "broadcast_address": "li-peng-mc-macbook.local",
      "tcp_port": 8000,
      "http_port": 8001,
      "version": "1.1.1-alpha"
    },
    {
      "remote_address": "127.0.0.1:58627",
      "hostname": "li-peng-mc-macbook.local",
      "broadcast_address": "li-peng-mc-macbook.local",
      "tcp_port": 7000,
      "http_port": 7001,
      "version": "1.1.1-alpha"
    }
  ]
}

方法queryLookupd就是进行的上图的操做

获得提供订阅的topic 的 nsqd列表
进行链接

func (r *Consumer) queryLookupd() {
    retries := 0
retry:
    endpoint := r.nextLookupdEndpoint()

    // ...  
    err := apiRequestNegotiateV1("GET", endpoint, nil, &data)
    if err != nil {
        // ...
    }
    var nsqdAddrs []string
    for _, producer := range data.Producers {
        broadcastAddress := producer.BroadcastAddress
        port := producer.TCPPort
        joined := net.JoinHostPort(broadcastAddress, strconv.Itoa(port))
        nsqdAddrs = append(nsqdAddrs, joined)
    }
    // 进行链接
    for _, addr := range nsqdAddrs {
        err = r.ConnectToNSQD(addr)
        if err != nil && err != ErrAlreadyConnected {
            r.log(LogLevelError, "(%s) error connecting to nsqd - %s", addr, err)
            continue
        }
    }
}

如何刷新nsqd的可用列表

有新的nsqd加入，是如何处理的呢？
在调用ConnectToNSQLookupd时会启动一个协程go r.lookupdLoop() 调用方法lookupdLoop的定时循环访问 queryLookupd 更新 nsqd的可用列表

// poll all known lookup servers every LookupdPollInterval
func (r *Consumer) lookupdLoop() {
    // ...
    var ticker *time.Ticker
    select {
    case <-time.After(jitter):
    case <-r.exitChan:
        goto exit
    }
    // 设置Interval 来循环访问 queryLookupd
    ticker = time.NewTicker(r.config.LookupdPollInterval)
    for {
        select {
        case <-ticker.C:
            r.queryLookupd()
        case <-r.lookupdRecheckChan:
            r.queryLookupd()
        case <-r.exitChan:
            goto exit
        }
    }

exit:
    // ...
}

处理 nsqd 的单点故障

当有nsqd出现故障时怎么办？当前的处理方式是

nsqdlookupd会把这个故障节点从可用列表中去除，客户端从接口获得的可用列表永远都是可用的。
客户端会把这个故障节点从可用节点上移除，而后要去判断是否使用了nsqlookup进行了链接，若是是则case r.lookupdRecheckChan <- 1 去刷新可用列表queryLookupd,若是不是，而后启动一个协程去定时作重试链接，若是故障恢复，链接成功，会从新加入到可用列表.
客户端实现的代码

func (r *Consumer) onConnClose(c *Conn) {
    // ...
    // remove this connections RDY count from the consumer's total
    delete(r.connections, c.String())
    left := len(r.connections)
    // ...
    r.mtx.RLock()
    numLookupd := len(r.lookupdHTTPAddrs)
    reconnect := indexOf(c.String(), r.nsqdTCPAddrs) >= 0
    // 若是使用的是nslookup则去刷新可用列表
    if numLookupd > 0 {
        // trigger a poll of the lookupd
        select {
        case r.lookupdRecheckChan <- 1:
        default:
        }
    } else if reconnect {
        // ... 
        }(c.String())
    }
}