精华！20k Start的消息队列NSQ源码剖析

什么是NSQ

NSQ is a realtime distributed messaging platform designed to operate at scale, handling billions of messages per day    

与kafka、MQ为同一类消息中间件，基于Go语言实现而且因为轻量化，备受好评
它基于MIT开源协议发布，由bitly公司开源出来的一款简单易用的消息中间件
复制代码

术语介绍

Topic ：一个topic就是程序发布消息的一个逻辑键，当程序第一次发布消息时就会建立topic
Channels ：channel与消费者相关，是消费者之间的负载均衡，channel在某种意义上来讲是一个“队列”

nsqd：nsqd 是一个守护进程，负责接收，排队，投递消息给客户端
nsqlookupd：nsqlookupd 是守护进程负责管理拓扑信息
nsqadmin：nsqadmin 是一套 WEB UI，用来聚集集群的实时统计，并执行不一样的管理任务
nsq_to _file：消费指定的话题（topic）/通道（channel），并写到文件中，有选择的滚动和/或压缩文件
nsq_to _http：消费指定的话题（topic）/通道（channel）和执行 HTTP requests (GET/POST) 到指定的端点
nsq_to _nsq：消费者指定的话题/通道和重发布消息到目的地 nsqd 经过 TCP
复制代码

NSQ的应用场景

流程异步化、代码解耦合、流量削峰、高可用、高吞吐量、广播分发，达到数据的最终一致性，知足具体的业务场景需求
复制代码

从该项目中你能得到什么？

Golang的入门    
复制代码

主题

• NSQ系统架构

  会由三个组件构成，分别扮演不一样的角色
  重要性： NSQ > NSQLOOKUP > NSD Admin

  1. NSQ
  2. NSQLOOKUP
  3. NSD Admin

  为了便于你们理解，能够当作坐飞机。
  把NSQ看做飞机，NSQLOOKUP当作飞机的调度台，NSQ Admin为订票的app，而咱们本身则是整个过程中的message。NSQ会搭载着咱们去想去的地方，航班号就是topic。

• NSQ的代码结构

  这里也大体分为3块
  1.入口函数，build app
  2.公共的抽象
  3.nsq、nsqlookup、admin的service层，实现实际的func，后续将对于这部分进行重点展开

• NSQ的组件剖析

• nsqd

  nsqd 是一个守护进程，负责接收（生产者 producer ）、排队（最小堆实现）、投递（消费者 consumer ）消息给客户端。它能够独立运行，不过一般它是由 nsqlookupd 实例所在集群配置的

  nsqd的执行逻辑

  Producer --> 消息 --> topic --> channels --> Consumer

  nsqd对象

  TCP 服务    
    HTTP 服务
  
    图中能够看到，核心对象有2个，topicMap 和 Listener    
    topicMap存储topic信息，tcpListener 和httpListener监听tcp服务和http服务    
    其余的对象则是围绕此展开，例如：有保证上下文的context，有原子性的sync.RWMutex，有消息池管理的poolSize等
  复制代码

  nsqd 的启动流程

  1，会启动tcp server 和http server ，并以daemon的方式进行监听 2，会不断给nsqlookupd发送注册心跳，保证在nsqdlookup注册成功且存活

  tips:这里nsqd借助了第三方的svc包，对后台进程进行了管理
  "github.com/judwhite/go-svc" 这个是一个服务运行框架

  消息的传递

  nsqd是如何进行消息的分发传递的？ 分发的过程当中应该注意什么问题？

  咱们先来看看nsqd对于消息的处理过程
  生产者，先将消息传递给Topics，达到A，而后会将指定的消息分发给特定的Channel，而后就会经过这个channel进行消息的消费，正因如此，也实现了消费者的负载均衡

  1, 消息的内存优化
  经过sync.Pool 保存复用 临时对象，避免存储从新分配，下降程序GC的压力，从而避免：并发⼤－占⽤内存⼤－GC 缓慢－处理并发能⼒下降－并发更⼤

  2，消息的传递过程对象
  使用byte来进行消息的传递，避免频繁格式转化致使的性能下降

  3，消息的格式
  时间戳（8byte）+msgID（16byte）+body（*byte）

  4，消息的高可靠性
  超出阈值的会有内存转为磁盘进行持久化

  5,经过最小堆(min heap)进行channel的排队

  nsqd 的接收消息过程

  1，由一个协程来监听tcp server
    2，由一个协程来处理请求来的client，并保持长连接，当conn failed的时候，还会主动让出cpu
    3，由一个协程来监听tcp socket，并持续对消息进行处理
  复制代码

  [tcp server 的监听]

  [持续处理来自client的请求]

• NSQLOOKUP

  nsqlookupd 是守护进程负责管理拓扑信息。客户端经过查询 nsqlookupd 来发现指定话题（ topic ）的生产者，而且 nsqd 节点广播话题（topic）和通道（ channel ）信息。有两个接口： TCP 接口， nsqd 用它来广播。 HTTP 接口，客户端用它来发现和管理

  NSQLOOKUP对象则较为简单
  1个原子性和6个对象，其中分别监听tcp和http 服务

  • TCP服务：
  • HTTP服务：

• NSQAdmin

  nsqadmin 是一套 WEB UI，用来聚集集群的实时统计，并执行不一样的管理任务
    因为这块内容比较单一，不进行过多的解析
  复制代码

亮点

性能优化手段

- sync.pool 进行内存复用,内存分片持久化，避免频繁回收致使的GC太高    
   - byte流传递，避免频繁序列化致使的性能降低    
   - cpu让出，例如，当conn err时候，主动让出cpu
   - 最小堆排序
   - 并发动态调整
   - 几率过时算法
   - 自定义数据报文
    
复制代码

开发手段

- 设计模式：工厂方法、装饰器等    
   - Go 并发处理，动态调整goroutiune
   - 优雅的OOP思路
复制代码

总结

虽然NSQ并不是完美，总体架构也并不是复杂，本篇介绍的内容也并不全面
可是，细节之处请细品

附录

NSQ的 Git：github.com/nsqio/nsq ，做为一个拥有20k star的消息中间件