。

如果没看过的各个部件的大致作用。

如果没看过的建议大家去学习下不然理解后续的内容会有难度

这篇博客开始我们来看下每个部件的详细功能从源码入手分析其内部实现原理

建议大家也下载nsq的代码一边看博客一边看代码印象更深刻。

nsq的官方git代码地址GitHub

nsqio/nsq:

nsq代码结构如下图中有注释大家先有个整体印象知道各个模块的代码在哪就行

图中最上面的四个节点就是nsqd进程至少要有1个可以多开。

我们画了4个分别是nsq1nsq2nsq3nsq4

注意看nsqd的连接关系每个nsqd节点和所有客户端都有连接(tcphttp)且每个nsqd节点和所有的nsqlookupd节点也有连接(tcp)

topic的创建清空暂停重新激活删除持久化(保存到文件从文件加载)同步给nsqlookupd进程

channel的创建清空暂停重新激活删除持久化(保存到文件从文件加载)同步给nsqlookupd进程

message的监听中转持久化(保存到文件从文件加载)主动推送消息给各个客户端超时重发消息计数

配置修改运行状态协程、内存统计

统计和上报工作主要统计topicchannel消息各种队列客户端连接GC等信息通过UDP协议上报给指定地址的进程

可以说nsqd进程是整个nsq平台的核心消息队列架构简单的话只有一个nsqd进程就够了。

代码框架及main函数目录在

值得一提的是nsqdnsqlookupdnsqadmin这三个进程的框架都使用了go-svc包这个包很简单使用者只需实现它的三个函数即可

Init()

好了我们看nsqd的入口也就是main函数代码在nsq/apps/nsqd/main.go代码如下(已加注释)

type

Run内部会调用Init()Start()监听到这两个系统信号时会调用Stop()if

err

main()函数内部只有一个对象program也只有一处调用svc.Run()这个函数内部会调用program.Init()和program.Start()

其中program.Init()函数主要是创建并检测nsqd的配置然后根据配置创建出一个nsqd实例

func

加载元数据并创建初始化出所有的topic对象所有的channel对象err

err

再持久化元数据到文件不要觉得奇怪因为上面的LoadMetadata()函数可能会过滤掉一些无效的topicchannel这里再重写算是刷新了元数据err

err

启动一个新协程专门运行nsqd的Main()循环注意这个Main()是永不退出的(除非出错)go

func()

对上面的代码解释下program.Start()函数一共干了3件事

nsqd.LoadMetadata(),

这个函数根据配置加载旧nsqd元数据。

这些元数据包含版本号topicchannel过滤掉不合法的topic和channel合法的topic和channel都创建出对象并且为每个topic建立处理循环

把过滤后的topic和channel再保存到文件nsqd.dat算是把旧数据过滤了一遍

新启动一个协程调用nsqd.Main()这个Main()是nsqd的核心启动了nsqd的全部服务。

除非遇到错误否则永不退出

接下来看nsqd.Main()的内部实现代码在nsq/nsqd/nsqd.go代码如下(已加注释)

func

TCP服务独立协程n.waitGroup.Wrap(func()

{exitFunc(protocol.TCPServer(n.tcpListener,

n.tcpServer,

TLSRequired)n.waitGroup.Wrap(func()

{exitFunc(http_api.Serve(n.httpListener,

httpServer,

{exitFunc(http_api.Serve(n.httpsListener,

httpsServer,

独立协程抽检扫描各个队列n.waitGroup.Wrap(n.queueScanLoop)//

独立协程和nsqlookupd的循环连接和重连心跳维持topicchannel变化通知等n.waitGroup.Wrap(n.lookupLoop)if

{n.waitGroup.Wrap(n.statsdLoop)}err

-exitChreturn

开一个新协程启动tcp服务并一直监听为客户端提供tcp服务。

我们的客户端最常用因为生产消息中转消息处理消息都是这里实现的

开一个新协程启动http服务并一直监听为客户端提供htttp服务

开一个新协程启动https服务并一直监听为客户端提供htttps服务

开一个新协程建立并维持扫描池这些扫描协程会扫描所有channel的延迟队列飞行队列如果消息超时了就重新入队。

很有意思的是nsqd作者很大方地承认他抄袭了redis的抽检策略内部实现也确实是类redis操作这个我们后面再讲todo

开一个新协程和nsqlookupd建立循环主要是连接和重连心跳维持实时报告自己的topic和channel变化

如果你配置了信息上报的地址nsqd会再开一个新协程做统计上报操作统计topicchannel消息内存GC等信息通过UDP协议上报过去

本篇博客总结

给大家看了nsq平台下代码整体结构建议大家下载源码自己看下加强印象

跟踪了nsqd进程启动流程最核心的nsqd.Main()建议大家仔细看后面讲的nsqd内容也都是这几个协程里面干的活

todo