nsq源码学习

87374491 2019-06-26

nsq源码学习

简介

nsq 是用go语言实现的分布式队列。阅读源码对go语言的chanel，分布式有着更好的理解

代码结构

核心代码分位3部分：

  • nsqd：队列数据存储
  • nsqlookup：管理nsqd节点，服务发现
  • nsqadmin：nsq的可视化

nsqd

官方的介绍为

nsqd is the daemon that receives, queues, and delivers messages to clients.

It can be run standalone but is normally configured in a cluster with nsqlookupd instance(s) (in which case it will announce topics and channels for discovery).

It listens on two TCP ports, one for clients and another for the HTTP API. It can optionally listen on a third port for HTTPS.

大意为：nsqd是接收，分发队列信息的守护进程。一般集群化运行，也可以独自部署。

下面对nsqd的2个逻辑做一次学习

  1. 启动逻辑
  2. 数据存储

启动逻辑

在Makefile中，写到

$(BLDDIR)/nsqd:        $(wildcard apps/nsqd/*.go       nsqd/*.go       nsq/*.go internal/*/*.go)

可以找到nsqd的代码入口在apps/nsqd/nsqd.go

apps/nsqd/nsqd.go

这个文件作为程序入口，主要做了几件事情：

  • 接收命令行参数
  • 根据命令行参数，新建nsqd结构
  • 启动nsqd

首先作者使用svc包来控制程序的启动：

type program struct {
    nsqd *nsqd.NSQD
}

func main() {
    prg := &program{}
    if err := svc.Run(prg, syscall.SIGINT, syscall.SIGTERM); err != nil {
        log.Fatal(err)
    }
}

func (p *program) Init(env svc.Environment) error {...}

func (p *program) Start() error {...}

func (p *program) Stop() error {...}

使用svc 能更简洁的保证程序干净的退出。在nsqd中，退出信号有两个：SIGINT(输入任意健) 和 SIGTERM(kill)。

Start()函数是主要逻辑的入口，在函数中引用了NewOptions()，它会创建一个默认的Options 结构。Options 后续会作为nsqd启动的参数来源

opts := nsqd.NewOptions()

作者通过flag包实现了命令行参数接收，如果命令行中执行配置文件，会同时读取配置文件。根据配置文件，命令行参数，来创建一个nsqd结构

options.Resolve(opts, flagSet, cfg)
nsqd := nsqd.New(opts)

接下来会加载数据

err := nsqd.LoadMetadata()
err = nsqd.PersistMetadata()

LoadMetadata()过程为：

  1. 先使用atomic库加锁
  2. 读取以node id的文件，以及默认文件，比对二者，并从文件中获取数据
  3. 将数据json 解析出meta 结构
  4. 遍历meta，获取topic name以及chanel name，对需要暂停的topic/chanel 进行暂停操作

PersistMetadata()过程为：

  1. 根据nsqd 结构获取对应的topic和channel
  2. 将topic和channel 持久化到文件中

接下来调用启动nsqd的主逻辑nsqd.Main()，主要完成以下过程

  1. 根据options 参数监听tcp端口,http端口,https端口
  2. 启动4个goroutines分别实启动http api，queueScanLoop,lookupLoop,statsdLoop
n.waitGroup.Wrap(func() {
        http_api.Serve(n.httpListener, httpServer, "HTTP", n.logf)
    })

    n.waitGroup.Wrap(func() { n.queueScanLoop() })
    n.waitGroup.Wrap(func() { n.lookupLoop() })
    if n.getOpts().StatsdAddress != "" {
        n.waitGroup.Wrap(func() { n.statsdLoop() })
    }

这里使用到了waitGroup，它是一个groutines 的控制包，能上线类似python 的join()功能。可以实现所有groutines都执行完再退出。

作者封装了waitGroup库

type WaitGroupWrapper struct {
    sync.WaitGroup
}

func (w *WaitGroupWrapper) Wrap(cb func()) {
    w.Add(1)
    go func() {
        cb()
        w.Done()
    }()
}

Add() 会计数器加1，Done（）使得计数器减一。此外WaitGroup提供Wait()函数：当计数器归0时，继续执行，否则阻塞。等待线程执行完再退出的作用。

此外，将函数作为参数，再在内部groutines执行，和python的装饰器的用法类似。

回到Main()函数中，启动http_api利用到了github.com/nsqio/nsq/internal/http_api包， 设置router等参数后，启动。

queueScanLoop() 是管道扫进程，他的逻辑是将tpic，channel中的数据读入到worker channel, 并每隔一定的时间更新worker数量，扫描chanel中的数据。

select {
        case <-workTicker.C:
            if len(channels) == 0 {
                continue
            }
        case <-refreshTicker.C:
            channels = n.channels()
            n.resizePool(len(channels), workCh, responseCh, closeCh)
            continue
        case <-n.exitChan:
            goto exit
        }

这里使用select来监听io操作，每隔扫描间隔时，判断channel中的是否存在数据需要处理，如果没有，则略过本次扫描。

每隔刷新间隔判断worker数量是否发生变化。

loop:
        numDirty := 0
        for i := 0; i < num; i++ {
            if <-responseCh {
                numDirty++
            }
        }

        if float64(numDirty)/float64(num) > n.getOpts().QueueScanDirtyPercent {
            goto loop
        }

这里还有dirty比率的概念，channel中有数据就认为是dirty，当该比率超过配置中的值时，则继续处理调用worker来处理，而不是等待固定间隔才进行扫描。

启动lookupLoop()和statsdLoop()；这两个函数的作用初步看和nsqdlookup通信用，细节还未了解。

上面阐述了nsqd的启动逻辑。nsqd使用http api和用户交互

数据存储

在api文档中，看到pub接口用来发布信息：

使用示例
curl -d "<message>" http://127.0.0.1:4151/pub?topic=name

在nsqd/http.go中，定义了路由规则

func newHTTPServer(ctx *context, tlsEnabled bool, tlsRequired bool) *httpServer {
    ...
    s := &httpServer{
        ctx:         ctx,
        tlsEnabled:  tlsEnabled,
        tlsRequired: tlsRequired,
        router:      router,
    }
    router.Handle("POST", "/pub", http_api.Decorate(s.doPUB, http_api.V1))
    ...
}

在doPUB()函数中，可以看到数据存储时，最终调用了opic.PutMessage(msg)

err = topic.PutMessage(msg)
func (t *Topic) PutMessage(m *Message) error {
    t.RLock()
    defer t.RUnlock()
    if atomic.LoadInt32(&t.exitFlag) == 1 {
        return errors.New("exiting")
    }
    err := t.put(m)
    if err != nil {
        return err
    }
    atomic.AddUint64(&t.messageCount, 1)
    return nil
}

PutMessage的逻辑是做并发控制（加锁）后，调Topic.put(*Message) 来写入信息。

这里有两个锁控制机制：

  1. RLock
  2. atomic

RLoclk

go语言中，sync包有两种锁，分别是互斥锁sync.Mutex和读写锁sync.RWMutex。

type Mutex
    func (m *Mutex) Lock()
    func (m *Mutex) Unlock()

type RWMutex
    func (rw *RWMutex) Lock()
    func (rw *RWMutex) RLock()
    func (rw *RWMutex) RLocker() Locker
    func (rw *RWMutex) RUnlock()
    func (rw *RWMutex) Unlock()

互斥锁倾向于在全局使用，一旦加锁，就必须解锁之后才能访问。不二次加锁、二次解锁都会报错。

读写锁用在读远多于写的场景。

Lock()表示写加锁，加写锁前，如果已经存在写锁，或者其他读锁，会阻塞住，直到锁可用。已阻塞的 Lock 调用会从获得的锁中排除新的读取器，即写锁权限高于读锁，有写锁时优先进行写锁定。

RLock()表示读加锁，当有写锁时，无法加载读锁，当只有读锁或者没有锁时，可以加载读锁，读锁可以加载多个，所以适用于＂读多写少＂的场景。

关于读写锁的具体例子请参考golang中sync.RWMutex和sync.Mutex区别

atomic

atomic是sync包中的另一种锁机制，在实现上，它比互斥锁层级更低：互斥锁调用的是golang的api，而atomic是在内核层面实现。因此它比互斥锁效率更高，但是使用上也存在一定的限制。如果使用存储相关接口，存入的是nil，或者类型不对，会报错。

此外，在一些文章中，以及stack overflow中都提到尽量少用atomic，具体原因还不知道。

atomic有几种常见的函数：

  1. CAS:比较和存储，如果是等于旧的值，就将新的值写入
  2. 增加或减少
  3. 读取或写入

具体查看atomic介绍

在上面的PutMessage逻辑中，增加topic读锁和topic中的部分值的原子操作锁后，调用了put()函数来实现写入。

func (t *Topic) put(m *Message) error {
    select {
    case t.memoryMsgChan <- m:
    default:
        b := bufferPoolGet()
        err := writeMessageToBackend(b, m, t.backend)
        bufferPoolPut(b)
        t.ctx.nsqd.SetHealth(err)
        if err != nil {
            t.ctx.nsqd.logf(LOG_ERROR,
                "TOPIC(%s) ERROR: failed to write message to backend - %s",
                t.name, err)
            return err
        }
    }
    return nil
}

put函数的操作是，将Message写入channel，如果该topic的memoryMsgChan长度满了，则通过default逻辑，写入buffer中.

buffer的实现是利用了sync.Pool包，相当于是一块缓存，在gc前释放，存储的长度受限于内存大小。

这里有两个问题：

  1. 存入memoryMsgChan就算完成topic写入了吗
  2. buffer中的数据怎么办

经过查找，发现处理上述两个channel的函数是messagePump，而messagePump在创建一个新Topic时会被后台调用：

func NewTopic(topicName string, ctx *context, deleteCallback func(*Topic)) *Topic {
    ...
    t.waitGroup.Wrap(func() {t.messagePump()})
    ...
}
func (t *Topic) messagePump() {
    ...
    if len(chans) > 0 {
        memoryMsgChan = t.memoryMsgChan
        backendChan = t.backend.ReadChan()
    }
    select {
        case msg = <-memoryMsgChan:
        case buf = <-backendChan:
            msg, err = decodeMessage(buf)
            if err != nil {
                t.ctx.nsqd.logf(LOG_ERROR, "failed to decode message - %s", err)
                continue
            }    
        ...
    }
    ...
    for i, channel := range chans {
        chanMsg := msg 
        if i > 0 {
                chanMsg = NewMessage(msg.ID, msg.Body)
                chanMsg.Timestamp = msg.Timestamp
                chanMsg.deferred = msg.deferred
            }        
        ...
        err := channel.PutMessage(chanMsg)
        ...
    }
    ...
}

上述调用了channel的PutMessage()完成了message写入channel的memoryMsgChan中，写入逻辑和写入topic逻辑类似。到这里完成了数据的写入流程分析。

nsqlookup

官方的介绍如下

nsqlookupd is the daemon that manages topology information. Clients query nsqlookupd to discover nsqd producers for a specific topic and nsqd nodes broadcasts topic and channel information.

There are two interfaces: A TCP interface which is used by nsqd for broadcasts and an HTTP interface for clients to perform discovery and administrative actions.

大意为：nsqlookup是管理nsqd集群拓补信息的守护进程。nsqlookup用于

  1. 供客户端查询，获得具体的topic和channel
  2. nsqd节点将自己的信息广播给nsqloookup。

下面梳理一下nsqllookup的两个逻辑：

  1. 供客户端查询具体的topic数据
  2. 接收nsqd的广播。

查询topic和channel

根据查询数据的过程进行梳理，nsq提供了几个封装好的查询接口，如果nsq_tail、nsq_to_file 等。此处从nsq_til 举例查看。

nsq_tail中主要逻辑如下：

consumers := []*nsq.Consumer{}
    for i := 0; i < len(topics); i += 1 {
        fmt.Printf("Adding consumer for topic: %s\n", topics[i])

        consumer, err := nsq.NewConsumer(topics[i], *channel, cfg)
        if err != nil {
            log.Fatal(err)
        }

        consumer.AddHandler(&TailHandler{topicName: topics[i], totalMessages: *totalMessages})

        err = consumer.ConnectToNSQDs(nsqdTCPAddrs)
        if err != nil {
            log.Fatal(err)
        }

        err = consumer.ConnectToNSQLookupds(lookupdHTTPAddrs)
        if err != nil {
            log.Fatal(err)
        }

        consumers = append(consumers, consumer)
    }

nsq_tail的逻辑是针对每个topic，分别初始化一个消费者consumer， 此处consumer实现的库是go-nsq。
并实现一个nsq_tail逻辑的handler，初始化在consumer中。

之后从nsqd和nsqdlookup中获取数据，并调用handler处理。

在go-nsq/consumer.go中，ConnectToNSQLookupd()会调用queryLookupd()和lookupdLoop()，而lookupdLoop()又会定期调用queryLookupd()。代码如下：

func (r *Consumer) ConnectToNSQLookupd(addr string) error {
    ...
    if numLookupd == 1 {
        r.queryLookupd()
        r.wg.Add(1)
        go r.lookupdLoop()
    }
    ...
}
func (r *Consumer) lookupdLoop() {
    ...
    for {
        select {
        case <-ticker.C:
            r.queryLookupd()
        case <-r.lookupdRecheckChan:
            r.queryLookupd()
        case <-r.exitChan:
            goto exit
        }
    }
    ...
}

// make an HTTP req to one of the configured nsqlookupd instances to discover
// which nsqd's provide the topic we are consuming.
//
// initiate a connection to any new producers that are identified.
func (r *Consumer) queryLookupd() {
    ...
    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)
    }
    // apply filter
    if discoveryFilter, ok := r.behaviorDelegate.(DiscoveryFilter); ok {
        nsqdAddrs = discoveryFilter.Filter(nsqdAddrs)
    }
    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
        }
    }
    
}

在queryLookupd()中，获取到生产者信息后，调用ConnectToNSQD()连接每个nsqd server。用ConnectToNSQD()实现了读取message。

ConnectYpNSQD()调用了connection结果的函数readLoop()。

func (c *Conn) readLoop() {
    for {
        ...
        frameType, data, err :=      ReadUnpackedResponse(c)
        ...
        switch frameType {
        case FrameTypeResponse:
            c.delegate.OnResponse(c, data)
        case FrameTypeMessage:
            msg, err := DecodeMessage(data)
            if err != nil {
                c.log(LogLevelError, "IO error - %s", err)
                c.delegate.OnIOError(c, err)
                goto exit
            }
            msg.Delegate = delegate
            msg.NSQDAddress = c.String()

            atomic.AddInt64(&c.rdyCount, -1)
            atomic.AddInt64(&c.messagesInFlight, 1)
            atomic.StoreInt64(&c.lastMsgTimestamp, time.Now().UnixNano())

            c.delegate.OnMessage(c, msg)
            ...
    }
}

在c.delegate.OnMessage(c, msg)中，会将message写入Consumer.incomingMessages。完成数据读取。

87374491

87374491

相关推荐

[Go] go-nsq 使用指南

首先你需要有一个 nsq 的服务端，nsq 由三部分构成：nsqd、nsqlookupd、nsqadmin。go-nsq 是众多客户端中的一个，使用 golang 编写，我们在逻辑代码中写 Producer 和 Comsumer。producer，把数据序

83374290 / 0评论 2020-06-18

NSQ

我们可以将原本耦合、同步执行的程序 解耦成 生产端+ 消息队列+消费端模型的异步程序，加上分布式的生产者和消费者架构就可以在一定程度上支撑大并发。NSQ是go语言开发的消息队列，所以对nsqd进行水平扩展时它的部署、配置也会相对简单。如果熟悉golang的

KOYO的云 / 0评论 2020-05-14

【代码篇】从零开始一步步搭建自己的golang框架（七）

这篇文章，我将会添加一个队列的功能。使用nsq需要知道以下几个概念：。nsqadmin：nsq的web管理组件；产生消息，需要指定该消息属于哪个topic的；nsq消费者与nsqd或者nsqlookupd建立一个channel，并监听channel中的消息

87374491 / 0评论 2020-02-17

golang nsq 同一个topic有多个channel,同时项目又互相引用事出现的问题

同一个topic下 不同channel分多个项目的时候,如果一个项目引用了别的项目包,那么被引用的项目就需要手动初始化mq,不能使用Init,因为会导致mq初始化两次,在一个channel下链接了两个消费者导致mq进行了负载均衡.

GoatSucker / 0评论 2020-01-11

NSQ 落地磁盘的消息队列源码解读

NSQ 消息队列实现消息落地使用的是 FIFO 队列。在NSQ 中， diskqueue 是一个实例化的 BackendQueue, 用于保存在内存放不下的消息 。是一个比较经典的队列的实现，值得学习。diskqueue 是 BackendQueue 的

83374290 / 0评论 2019-11-04

nsq 初学使用日记

注意其中tcp端口是给nsqd节点连接使用的,http则是给一些输出文件使用,比如数据经过节点会发送请求之类的。-address 表示lookupd的服务器ip地址在哪 端口号是lookupd 启动时开启的端口。// 除了第一个开启的节点只带lookupd

84394193 / 0评论 2019-11-02

nsq (三) 消息传输的可靠性和持久化[一]

上两篇帖子主要说了一下nsq的拓扑结构，如何进行故障处理和横向扩展,保证了客户端和服务端的长连接，连接保持了，就要传输数据了，nsq如何保证消息被订阅者消费，如何保证消息不丢失，就是今天要阐述的内容。消息处理的大概流程:当一个消息被nsq接收后，传给相应的

85354698 / 0评论 2019-10-31

深入NSQ 之旅

介绍NSQ是一个实时的分布式消息平台。它的设计目标是为在多台计算机上运行的松散服务提供一个现代化的基础设施骨架。可以说， 如果不是因为我们在bitly使用go语言，NSQ就不会存在。这里既会讲NSQ的功能也会涉及语言提供的特征。当然，语言会影响思维，这次也

83374290 / 0评论 2015-02-27

NSQ源码-Nsq客户端

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

84394193 / 0评论 2019-06-29

NSQ源码-NSQD

看完了nsqlookupd我们继续往下看, nsqd才是他的核心. 里面大量的使用到了go channel, 相信看完之后对你学习go有很大的帮助.相较于lookupd部分无论在代码逻辑和实现上都要复杂很多. 不过基本的代码结构基本上都是一样的, 进程使用

84394195 / 0评论 2019-06-29

NSQ源码-nsqlookupd

为什么选择nsq之前一直在用erlang做电信产品的开发，对erlang的一些生态也比较了解，和erlang相关的产品在互联网公司使用最多的应该就是rabbitmq了，也许很多人听说过erlang就是因为他们公司在使用rabbitmq。在之前也看过一点ra

83374290 / 0评论 2019-06-29

How we redesigned the NSQ- 其他特性及未来计划

在系列文章前面几篇中，介绍了 NSQ 改造的过程和几个基础特性，本文中我们继续介绍几个高级特性及其使用场景，这些都是结合有赞业务场景总结提炼出来的重要功能。相比较于 Kafka 等消息中间件，NSQ 的消息格式在内容和数量上较为简单。一条消息除了基本的元数

83357012 / 0评论 2019-06-27

nsq 优秀的消息队列

NSQ是Go语言编写的，开源的分布式消息队列中间件，其设计的目的是用来大规模地处理每天数以十亿计级别的消息。NSQ 具有分布式和去中心化拓扑结构，该结构具有无单点故障、故障容错、高可用性以及能够保证消息的可靠传递的特征，是一个成熟的、已在大规模生成环境下应

85354698 / 0评论 2019-06-26

NSQ 源码阅读 (四) diskqueue 文件读写

引言我们在这个系列第一篇文章中提到过，如果需要消息落地而对存储子系统的选择上，从速度上来说 文件系统>分布式KV（持久化）>分布式文件系统>数据库。而NSQ选择了文件系统作为存储子系统。这篇文章将重点介绍nsq 对于文件的操作。在内存的m

newtrekWang / 0评论 2019-06-25

NSQ 源码阅读(一) 相关概念理解

NSQ是由知名短链接服务商bitly用Go语言开发的实时消息处理系统，具有高性能、高可靠、无视单点故障等优点，是一个非常不错的新兴的消息队列解决方案。相关概念提到消息队列，我们先看一下消息队列的相关概念。以下主要参考 消息队列设计精要，我提取一些我认为重要

KOYO的云 / 0评论 2019-06-20

golang使用Nsq

最近在研究一些消息中间件，常用的MQ如RabbitMQ,ActiveMQ,Kafka等。NSQ是一个基于Go语言的分布式实时消息平台，它基于MIT开源协议发布，由bitly公司开源出来的一款简单易用的消息中间件。NSQ提供了分布式的，去中心化，且没有单点故

hjr / 0评论 2019-06-13