nsq部署_andlua辅助源码

nsq部署_andlua辅助源码NSQ源码-Nsq客户端

大家好,又见面了,我是你们的朋友全栈君。

看完lookupd和nsqd之后我们再来看下nsq client端的代码。 我是想把nsq系统完完整整的看一遍,从而对他形成一个更整体的
认识。对message queue来说他的client端就是生产者和消费者,生产者负责想nsq中投递消息,消费者负责从lookupd中获取到
指定nsqd之后,从nsqd中获取消息。

生产者

我们以nsq/apps/to_nsq/to_nsq.go为例,客户端这边的代码逻辑就简单很多,NewProducer实例化一个instance,publish消息
到nsqd。

/// nsq/apps/to_nsq/to_nsq.go
producer, err := nsq.NewProducer(addr, cfg)
err := producer.Publish(*topic, line)

下面来看下Publish里的具体逻辑。

// Publish synchronously publishes a message body to the specified topic, returning
// an error if publish failed
func (w *Producer) Publish(topic string, body []byte) error {
    // 生成具体的cmd
    return w.sendCommand(Publish(topic, body))
}
func (w *Producer) sendCommand(cmd *Command) error {
    doneChan := make(chan *ProducerTransaction)
    err := w.sendCommandAsync(cmd, doneChan, nil)
    if err != nil {
        close(doneChan)
        return err
    }
    t := <-doneChan
    return t.Error
}
func (w *Producer) sendCommandAsync(cmd *Command, doneChan chan *ProducerTransaction,
    args []interface{}) error {
    // keep track of how many outstanding producers we're dealing with
    // in order to later ensure that we clean them all up...
    atomic.AddInt32(&w.concurrentProducers, 1)
    defer atomic.AddInt32(&w.concurrentProducers, -1)

    if atomic.LoadInt32(&w.state) != StateConnected {
        // 这里是一个lazily connect
        err := w.connect()
        if err != nil {
            return err
        }
    }

    t := &ProducerTransaction{
        cmd:      cmd,
        doneChan: doneChan,
        Args:     args,
    }

    select {
    case w.transactionChan <- t:
    case <-w.exitChan:
        return ErrStopped
    }

    return nil
}

在connect函数里启动了一个go routine去处理transactionChan对应的东西

func (w *Producer) connect() error {
    w.closeChan = make(chan int)
    w.conn = NewConn(w.addr, &w.config, &producerConnDelegate{w})
    w.conn.SetLogger(logger, logLvl, fmt.Sprintf("%3d (%%s)", w.id))
    _, err := w.conn.Connect()
    w.wg.Add(1)
    go w.router()

这里需要注意一下, go-nsq/conn.go是对底层连接的一个抽象,他是不关心你是生产者还是消费者,这里使用到了
delegate 模式,conn.go收到消息的处理放到了producerConnDelegate和consumerConnDelegate中,然后通知到具体的
消费者活着生产者。

消费者

回过头我们再来看下消费者部分的代码,client端我们以nsq/apps/nsq_tail/nsq_tail.go为例,代码的基本逻辑如下:


// 1. new comsunmer instanace 
consumer, err := nsq.NewConsumer(topics[i], *channel, cfg)
// 2. add handler
consumer.AddHandler(&TailHandler{topicName: topics[i], totalMessages: *totalMessages})
// 3. connect to nsqd
consumer.ConnectToNSQDs(nsqdTCPAddrs)
if err != nil {
    log.Fatal(err)
}
// 4. connect to lookupd
err = consumer.ConnectToNSQLookupds(lookupdHTTPAddrs)
if err != nil {
    log.Fatal(err)
}
consumers = append(consumers, consumer)

下面来看下每个部分的实际代码:

func (r *Consumer) AddHandler(handler Handler) {
    r.AddConcurrentHandlers(handler, 1)
}
func (r *Consumer) AddConcurrentHandlers(handler Handler, concurrency int) {
    if atomic.LoadInt32(&r.connectedFlag) == 1 {
        panic("already connected")
    }

    atomic.AddInt32(&r.runningHandlers, int32(concurrency))
    for i := 0; i < concurrency; i++ {
        go r.handlerLoop(handler)
    }
}

至此handler添加完成,起一个单独的go routine来等待消息的到了。

func (r *Consumer) handlerLoop(handler Handler) {
    r.log(LogLevelDebug, "starting Handler")

    for {
        message, ok := <-r.incomingMessages // 有新的消息的到来
        if !ok {
            goto exit
        }

        if r.shouldFailMessage(message, handler) {
            message.Finish()
            continue
        }

        err := handler.HandleMessage(message) // 调用之前注册的handler
        if err != nil {
            r.log(LogLevelError, "Handler returned error (%s) for msg %s", err, message.ID)
            if !message.IsAutoResponseDisabled() {
                message.Requeue(-1)
            }
            continue
        }

        if !message.IsAutoResponseDisabled() {
            message.Finish()
        }
    }

exit:
    r.log(LogLevelDebug, "stopping Handler")
    if atomic.AddInt32(&r.runningHandlers, -1) == 0 {
        r.exit()
    }
}

官方是不推荐只部署nqd而不部署lookupd的,我们直接看下lookup的连接过程:

func (r *Consumer) ConnectToNSQLookupd(addr string) error {
    ...
    r.lookupdHTTPAddrs = append(r.lookupdHTTPAddrs, addr)
    numLookupd := len(r.lookupdHTTPAddrs)
    r.mtx.Unlock()

    // if this is the first one, kick off the go loop
    if numLookupd == 1 {
        r.queryLookupd()
        r.wg.Add(1)
        go r.lookupdLoop()
    }
    return nil
}

在queryLookupd中先去查询lookupd获取最新的nqd地址,然后connect to nsqd.

func (r *Consumer) lookupdLoop() {
    // add some jitter so that multiple consumers discovering the same topic,
    // when restarted at the same time, dont all connect at once.
    ticker = time.NewTicker(r.config.LookupdPollInterval)
    // 每个ticker interval更新nqd的地址信息
    for {
        select {
        case <-ticker.C:
            r.queryLookupd()
        case <-r.lookupdRecheckChan:
            r.queryLookupd()
        case <-r.exitChan:
            goto exit
        }
    }
}
func (r *Consumer) ConnectToNSQD(addr string) error {
    // 1. new connection
    conn := NewConn(addr, &r.config, &consumerConnDelegate{r})
    conn.SetLogger(logger, logLvl,
    fmt.Sprintf("%3d [%s/%s] (%%s)", r.id, r.topic, r.channel))

    // 2. connection list
    _, pendingOk := r.pendingConnections[addr]
    _, ok := r.connections[addr]

    r.pendingConnections[addr] = conn
    if idx := indexOf(addr, r.nsqdTCPAddrs); idx == -1 {
        r.nsqdTCPAddrs = append(r.nsqdTCPAddrs, addr)
    }

    r.log(LogLevelInfo, "(%s) connecting to nsqd", addr)
    // 3. new connect
    //   3.1 go c.readLoop()
    //   3.2 go c.writeLoop()
    resp, err := conn.Connect()
    
    // 4. sub to nsqd
    cmd := Subscribe(r.topic, r.channel)
    err = conn.WriteCommand(cmd)
}

以上就是客户端初始化的一个流程,然后就是接受消息处理了。

->NewConsumer() // 新建一个consumer
->ConnectToNSQLookupds() // 连接到lookupd
  |->ConnectToNSQLookupd() // 连接到lookupd
     |->r.queryLookupd() // 查询lookupd的
         |->apiRequestNegotiateV1() // 调用lookupd的rest api获取nsqd消息
         |->ConnectToNSQD() // 连接到具体nsq
            |->NewConn() // 连接instance
            |->conn.Connect() // 开始连接
                  |->c.readLoop() // 与nqd连接read loop
                  |->c.writeLoop() // 与nqd连接write loop
            |->Subscribe() // consumer发送SUB command
     |->lookupdLoop() // 定时查询lookupd并更新nsqd信息

注:

[1]. 关于delegate模式参考 这里

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/107169.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • Jlink-10 pin 的定义(stm32使用)官方定义

    Jlink-10 pin 的定义(stm32使用)官方定义因为在网上找了好久才找到正确的接法,所以专门记载了下来,因为stm32芯片这几个功能引脚会内置上拉电阻,所以不需要再外接电阻了。转载于:https://www.cnblogs.com/dzswise/p/7523420.html…

  • 从单一到融合,扫地机器人导航技术的“最优解”?

    从单一到融合,扫地机器人导航技术的“最优解”?人工智能浪潮下,智能家居产品层出不穷,但纵观行业发展能真正走入家庭中的产品屈指可数,而扫地机器人却是其中的“网红产品”。根据中怡康数据显示,2013年,我国扫地机器人市场规模仅为8.4亿元,而到了2020年,市场规模已达到94亿元。快速增长的市场同时也在倒逼扫地机器人的性能不断提升“下限”,尤其智能化已成为扫地机器人的高附加值特征。循次渐进,扫地机器人正逐步“完善”在经历2019年的市场寒潮后,2020年扫地机市场快速回暖,除了受疫情影响之外,最关键的原因仍是扫地机器人技术的创新和进步。尤其是各

  • mysql 全文索引 使用_MySql全文索引

    mysql 全文索引 使用_MySql全文索引使用索引是数据库性能优化的必备技能之一。在MySQL数据库中,有四种索引:聚集索引(主键索引)、普通索引、唯一索引以及我们这里将要介绍的全文索引(FULLTEXTINDEX)。全文索引(也称全文检索)是目前搜索引擎使用的一种关键技术。它能够利用「分词技术「等多种算法智能分析出文本文字中关键字词的频率及重要性,然后按照一定的算法规则智能地筛选出我们想要的搜索结果。在这里,我们就不追根究底其底层实现…

  • laravel 多个项目共享SESSION

    laravel 多个项目共享SESSION

  • Linux解压tar.gz和tar.bz2的命令「建议收藏」

    Linux解压tar.gz和tar.bz2的命令「建议收藏」两者的命令主要是参数的不同,解压tar.gz和tar.bz2不同压缩文件的命令如下:  1解压tar.gz文件tar-zxvf×××.tar.gz  2解压tar.bz2文件tar-jxvf×××.tar.bz2  -z:是否同时具有gzip的属性?亦即是否需要用gzip压缩?  -j:是否同时具有bzip2的属性?亦即是否需要用bzip2压缩?

  • 武后:像输入英文一样输入中文

    武后:像输入英文一样输入中文原文:TypingChineselikeEnglish作者:tedclancy翻译:赖信涛责编:仲培艺很长时间以来,我都希望输入中文能像输入英文一样。我的意思是,当你第一次看到中文键盘的时候,就知道如何输入中文,无论你的母语是什么。无需学习输入法(如仓颉)。无需知道怎么说普通话,无需学习如何写中文——我希望中文就像英文一样,所见即所得。15年来,我都在为这个问题寻找一个解…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号