NSQ源码剖析(一)：NSQD主要结构方法和消息生产消费过程
Jo_ZSM 人气:0目录
- 1 概述
- 2 主要结构体及方法
- 2.1 NSQD
- 2.2 tcpServer
- 2.3 protocolV2
- 2.4 clientV2
- 2.5 Topic
- 2.6 channel
- 3 启动过程
- 4 消费和生产过程
- 4.1 消息生产
- 4.2 消息消费
- 4.2 延迟消费
1 概述
NSQ包含3个组件:
- nsqd：每个nsq实例运行一个nsqd进程，负责接收生产者消息、向nsqlookupd注册、向消费者推送消息
- nsqlookupd：集群注册中心，可以有多个，负责接收nsqd的注册信息，向消费者提供服务发现
- nsqadmin：用于监控和管理的web ui
生产者将消息写入到指定的主题Topic，同一个Topic下则可以关联多个管道Channel，每个Channel都会传输对应Topic的完整副本。
消费者则订阅Channel的消息。于是多个消费者订阅不同的Channel的话，他们各自都能拿到完整的消息副本；但如果多个消费者订阅同一个Channel，则是共享的，即消息会随机发送给其中一个消费者。
接下来我们来分析下nsq的源码：
- 源码地址：https://github.com/nsqio/nsq
- 2020年1月18日最新的master分支
nsq各组件均使用上述代码仓库，通过apps目录下的不同的main包启动。比如nsqd的main函数在apps/nsqd目录下，其他类同。
本文档主要分析nsqd的主要结构体和方法，及消息生产和消费的过程。主要以TCP api为例来分析，HTTP/HTTPS的api类同。
2 主要结构体及方法
2.1 NSQD
nsqd/nsqd.go文件，NSQD是主实例，一个nsqd进程创建一个nsqd结构体实例，并通过此结构体的Main()方法启动所有的服务。
type NSQD struct {
clientIDSequence int64 // 递增的客户端ID，每个客户端连接均从这里取一个递增后的ID作为唯一标识
sync.RWMutex
opts atomic.Value // 参数选项，真实类型是apps/nsqd/option.go:Options结构体
dl *dirlock.DirLock
isLoading int32
errValue atomic.Value
startTime time.Time
topicMap map[string]*Topic // 保存当前所有的topic
clientLock sync.RWMutex
clients map[int64]Client
lookupPeers atomic.Value
tcpServer *tcpServer
tcpListener net.Listener
httpListener net.Listener
httpsListener net.Listener
tlsConfig *tls.Config
poolSize int // 当前工作协程组的协程数量
notifyChan chan interface{}
optsNotificationChan chan struct{}
exitChan chan int
waitGroup util.WaitGroupWrapper
ci *clusterinfo.ClusterInfo
}主要方法
/*
程序启动时调用本方法，执行下面的动作：
- 启动TCP/HTTP/HTTPS服务
- 启动工作协程组：NSQD.queueScanLoop
- 启动服务注册：NSQD.lookupLoop
*/
func (n *NSQD) Main() error
// 负责管理工作协程组的数量，每调用一次NSQD.queueScanWorker()方法启动一个工作协程
func (n *NSQD) queueScanLoop()
// 由queueScanLoop()调用，负责启动工作协程组并动态调整协程数量。工作协程的数量为当前的channel数 * 0.25
func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int)
// 这是具体的工作协程，监听workCh，对收到的待处理Channel做两个动作，一是将超时的消息重新入队；二是将到时间的延时消息入队
func (n *NSQD) queueScanWorker(workCh chan *Channel, responseCh chan bool, closeCh chan int)
/*
lookupLoop()方法与nsqlookupd建立连接，负责向nsqlookupd注册topic，并定时发送心跳包
*/
func (n *NSQD) lookupLoop()2.2 tcpServer
nsqd/tcp.go文件，tcpServer通过Handle()方法接收TCP请求。
tcpServer是nsqd结构的成员，全局也就只有一个实例，但在protocol包的TCPServer方法中，每创建一个新的连接，均会调用一次tcpServer.Handle()
type tcpServer struct {
ctx *context
conns sync.Map
}主要方法
/*
p.nsqd.Main()启动protocol.TCPServer()，这个方法里会为每个客户端连接创建一个新协程，协程执行tcpServer.Handle()方法
本方法首先对新连接读取4字节校验版本，新连接必须首先发送4字节" V2"。
然后阻塞调用nsqd.protocolV2.IOLoop()处理客户端接下来的请求。
*/
func (p *tcpServer) Handle(clientConn net.Conn)2.3 protocolV2
nsqd/protocol_v2.go文件，protocolV2负责处理过来的具体的用户请求。
每个连接均会创建一个独立的protocolV2实例（由tcpServer.Handle()创建）
type protocolV2 struct {
ctx *context
}主要方法
/*
tcpServer.Handle()阻塞调用本方法
1. 启用一个独立协程向消费者推送消息protocolV2.messagePump()
2. for循环接收并处理客户端请求protocolV2.Exec()
*/
func (p *protocolV2) IOLoop(conn net.Conn) error
// 组装消息并调用protocolV2.Send()发送给消费者
func (p *protocolV2) SendMessage(client *clientV2, msg *Message) error
// 向客户端发送数据帧
func (p *protocolV2) Send(client *clientV2, frameType int32, data []byte) error
// 解析客户端请求的指令，调用对应的指令方法
func (p *protocolV2) Exec(client *clientV2, params [][]byte) ([]byte, error)
// 负责向消费者推送消息
func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool)
// 下面这组方法是NSQD支持的指令对应的处理方法
func (p *protocolV2) IDENTIFY(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) AUTH(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) RDY(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) FIN(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) REQ(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) CLS(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) NOP(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) PUB(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) MPUB(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) DPUB(client *clientV2, params [][]byte) ([]byte, error)
func (p *protocolV2) TOUCH(client *clientV2, params [][]byte) ([]byte, error)2.4 clientV2
nsqd/client_v2.go文件，保存每个客户端的连接信息。
clientV2实例由protocolV2.IOLoop()创建，每个连接均有一个独立的实例。
type clientV2 struct {
// 64bit atomic vars need to be first for proper alignment on 32bit platforms
ReadyCount int64
InFlightCount int64
MessageCount uint64
FinishCount uint64
RequeueCount uint64
pubCounts map[string]uint64
writeLock sync.RWMutex
metaLock sync.RWMutex
ID int64
ctx *context
UserAgent string
// original connection
net.Conn
// connections based on negotiated features
tlsConn *tls.Conn
flateWriter *flate.Writer
// reading/writing interfaces
Reader *bufio.Reader
Writer *bufio.Writer
OutputBufferSize int
OutputBufferTimeout time.Duration
HeartbeatInterval time.Duration
MsgTimeout time.Duration
State int32
ConnectTime time.Time
Channel *Channel
ReadyStateChan chan int
ExitChan chan int
ClientID string
Hostname string
SampleRate int32
IdentifyEventChan chan identifyEvent
SubEventChan chan *Channel
TLS int32
Snappy int32
Deflate int32
// re-usable buffer for reading the 4-byte lengths off the wire
lenBuf [4]byte
lenSlice []byte
AuthSecret string
AuthState *auth.State
}2.5 Topic
nsqd/topic.go文件，对应于每个topic实例，由系统启动时创建或者发布消息/消费消息时自动创建。
type Topic struct {
messageCount uint64 // 累计消息数
messageBytes uint64 // 累计消息体的字节数
sync.RWMutex
name string // topic名，生产和消费时需要指定此名称
channelMap map[string]*Channel // 保存每个channel name和channel指针的映射
backend BackendQueue // 磁盘队列，当内存memoryMsgChan满时，写入硬盘队列
memoryMsgChan chan *Message // 消息优先存入这个内存chan
startChan chan int
exitChan chan int
channelUpdateChan chan int
waitGroup util.WaitGroupWrapper
exitFlag int32
idFactory *guidFactory
ephemeral bool
deleteCallback func(*Topic)
deleter sync.Once
paused int32
pauseChan chan int
ctx *context
}主要方法
/*
下面两个方法负责将消息写入topic，底层均调用topic.put()方法
1. topic.memoryMsgChan未满时，优先写入内存memoryMsgChan
2. 否则，写入磁盘topic.backend
*/
func (t *Topic) PutMessage(m *Message) error
func (t *Topic) PutMessages(msgs []*Message) error
/*
NewTopic创建新的topic时会为每个topic启动一个独立线程来处理消息推送，即messagePump()
此方法循环随机从内存memoryMsgChan和磁盘队列backend中取消息写入到topic下每一个chnnel中
*/
func (t *Topic) messagePump()2.6 channel
nsqd/channel.go文件，对应于每个channel实例
type Channel struct {
requeueCount uint64
messageCount uint64
timeoutCount uint64
sync.RWMutex
topicName string
name string
ctx *context
backend BackendQueue // 磁盘队列，当内存memoryMsgChan满时，写入硬盘队列
memoryMsgChan chan *Message // 消息优先存入这个内存chan
exitFlag int32
exitMutex sync.RWMutex
// state tracking
clients map[int64]Consumer
paused int32
ephemeral bool
deleteCallback func(*Channel)
deleter sync.Once
// Stats tracking
e2eProcessingLatencyStream *quantile.Quantile
// TODO: these can be DRYd up
deferredMessages map[MessageID]*pqueue.Item // 保存尚未到时间的延迟消费消息
deferredPQ pqueue.PriorityQueue // 保存尚未到时间的延迟消费消息，最小堆
deferredMutex sync.Mutex
inFlightMessages map[MessageID]*Message // 保存已推送尚未收到FIN的消息
inFlightPQ inFlightPqueue // 保存已推送尚未收到FIN的消息，最小堆
inFlightMutex sync.Mutex
}主要方法
/*
将消息写入channel，逻辑与topic的一致，内存未满则优先写内存chan，否则写入磁盘队列
*/
func (c *Channel) PutMessage(m *Message) error
func (c *Channel) put(m *Message) error
// 消费超时相关
func (c *Channel) StartInFlightTimeout(msg *Message, clientID int64, timeout time.Duration) error
func (c *Channel) pushInFlightMessage(msg *Message) error
func (c *Channel) popInFlightMessage(clientID int64, id MessageID) (*Message, error)
func (c *Channel) addToInFlightPQ(msg *Message)
func (c *Channel) removeFromInFlightPQ(msg *Message)
func (c *Channel) processInFlightQueue(t int64) bool
// 延时消费相关
func (c *Channel) StartDeferredTimeout(msg *Message, timeout time.Duration) error
func (c *Channel) pushDeferredMessage(item *pqueue.Item) error
func (c *Channel) popDeferredMessage(id MessageID) (*pqueue.Item, error)
func (c *Channel) addToDeferredPQ(item *pqueue.Item)
func (c *Channel) processDeferredQueue(t int64) bool3 启动过程
nsqd的main函数在apps/nsqd/main.go文件。
启动时调用了一个第三方包svc，主要作用是拦截syscall.SIGINT/syscall.SIGTERM这两个信号，最终还是调用了main.go下的3个方法：
- program.Init()：windows下特殊操作
- program.Start()：加载参数和配置文件、加载上一次保存的Topic信息并完成初始化、创建nsqd并调用p.nsqd.Main()启动
- program.Stop()：退出处理
p.nsqd.Main()的逻辑也很简单，代码不贴了，依次启动了TCP服务、HTTP服务、HTTPS服务这3个服务。除此之外，还启动了以下两个协程：
- queueScanLoop：消息延时/超时处理
- lookupLoop：服务注册
TCPServer
protocol包的TCPServer的核心代码就是下面这几行，循环等待客户端连接，并为每个连接创建一个独立的协程：
func TCPServer(listener net.Listener, handler TCPHandler, logf lg.AppLogFunc) error {
for {
// 等待生产者或消费者连接
clientConn, err := listener.Accept()
// 每创建一个连接wg +1
wg.Add(1)
go func() {
// 每个连接均启动一个独立的协程来接收处理请求
handler.Handle(clientConn)
wg.Done()
}()
}
// 等待所有协程退出
wg.Wait()
return nil
}TCPServer的核心是为每个连接启动的协程处理方法handler.Handle(clientConn)，实际调用的是下面这个方法，连接建立时先读取4字节，必须是" V2"，然后启动prot.IOLoop(clientConn)处理接下来的客户端请求：
func (p *tcpServer) Handle(clientConn net.Conn) {
// 无论是生产者还是消费者，建立连接时，必须先发送4字节的" V2"进行版本校验
buf := make([]byte, 4)
_, err := io.ReadFull(clientConn, buf)
protocolMagic := string(buf)
var prot protocol.Protocol
switch protocolMagic {
case " V2":
prot = &protocolV2{ctx: p.ctx}
default:
protocol.SendFramedResponse(clientConn, frameTypeError, []byte("E_BAD_PROTOCOL"))
clientConn.Close()
p.ctx.nsqd.logf(LOG_ERROR, "client(%s) bad protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)
return
}
// 版本校验通过，保存连接信息，key-是ADDR，value是当前连接指针
p.conns.Store(clientConn.RemoteAddr(), clientConn)
// 启动
err = prot.IOLoop(clientConn)
if err != nil {
p.ctx.nsqd.logf(LOG_ERROR, "client(%s) - %s", clientConn.RemoteAddr(), err)
}
p.conns.Delete(clientConn.RemoteAddr())
}4 消费和生产过程
4.1 消息生产
生产者pub消息时，消息会首先写入对应topic的队列（内存优先，内存满了写磁盘），topic的messagePump()方法再将消息拷贝给每个channel。
每个channel均各执一份完整的消息。
1.消息写入topic
消息生产由生产者调用PUB/MPUB/DPUB这类指令实现，底层都是调用topic.PutMessage(msg)，进一步调用topic.put(msg)：
func (t *Topic) put(m *Message) error {
select {
case t.memoryMsgChan <- m: // 优先写入内存memoryMsgChan
default: // 当内存case失败即memoryMsgChan满时，走default，将msg以字节形式写入磁盘队列topic.backend
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
}消息写入topic的逻辑比较简单，优先写memoryMsgChan，如果memoryMsgChan满了，则写入磁盘队列topic.backend。
这里留个思考题：NSQ是否支持不写内存，全部写磁盘队列？
2.topic将消息复制给每个channel
第二章介绍结构体和方法时，介绍了topic结构体的messagePump()方法，正是这个方法将第1步写入的消息复制给每个channel的：
func (t *Topic) messagePump() {
/* 准备工作有代码我们略过 */
// 主消息处理循环
for {
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
}
case <-t.channelUpdateChan:
chans = chans[:0]
t.RLock()
for _, c := range t.channelMap {
chans = append(chans, c)
}
t.RUnlock()
if len(chans) == 0 || t.IsPaused() {
memoryMsgChan = nil
backendChan = nil
} else {
memoryMsgChan = t.memoryMsgChan
backendChan = t.backend.ReadChan()
}
continue
case <-t.pauseChan:
if len(chans) == 0 || t.IsPaused() {
memoryMsgChan = nil
backendChan = nil
} else {
memoryMsgChan = t.memoryMsgChan
backendChan = t.backend.ReadChan()
}
continue
case <-t.exitChan:
goto exit
}
for i, channel := range chans {
chanMsg := msg
/* channel消费消息时，需要处理延时/超时等问题，所以这里复制了消息，给每个channel传递的是独立的消息实例 */
if i > 0 {
chanMsg = NewMessage(msg.ID, msg.Body)
chanMsg.Timestamp = msg.Timestamp
chanMsg.deferred = msg.deferred
}
if chanMsg.deferred != 0 {
channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
continue
}
err := channel.PutMessage(chanMsg)
if err != nil {
t.ctx.nsqd.logf(LOG_ERROR,
"TOPIC(%s) ERROR: failed to put msg(%s) to channel(%s) - %s",
t.name, msg.ID, channel.name, err)
}
}
}
}topic.messagePump()方法代码还蛮长的，前面是些准备工作，主要就是后面的for循环。其中for循环中select的前两项，memoryMsgChan来源于topic.memoryMsgChan，而backendChan则是topic.backend.ReadChan()，分别对应于内存和磁盘队列。注意只有这两个case会往下传递消息，其他的case处理退出和更新机制的，会continue或exit外层的for循环。
虽然通道channel是有序的，但select的case具有随机性，这就决定了每轮循环读的是内存还是磁盘是随机的，消息的消费顺序是不可控的。
select语句获取的消息，交给第2层for循环处理，逻辑比较简单，遍历每一个chan，调用channel.PutMessage()写入。由于每个channel对应于不同的消费者，有不同的延时/超时和消费机制，所以这里拷贝了message实例。
4.2 消息消费
每个连接均会启动一个运行protocolV2.messagePump()方法的协程，这个协程负责监听channel的消息队列并向客户端推送消息。客户端只有触发SUB指令之后，才会将channel传递给protocolV2.messagePump()，这之后消费推送才会正式开启。
启动消息推送
前面讲Tcpserver时有提到，客户端创建连接时，会调用tcpserver.Handle()，里面再调用protocolV2.IOLoop()。protocolV2.IOLoop()方法开头有下面这行：
go p.messagePump(client, messagePumpStartedChan)这行创建了一个独立线程，调用的protocolV2.messagePump()负责向消费者推送消息。
有个小细节是无论是生产者还是消费者，都会创建这个协程，protocolV2.messagePump()创建后并不会立即推送消息，而是需要调用SUB指令，以protocolV2.SUB()方法为例，方法末尾有这么一行：
client.SubEventChan <- channel将当前消费者订阅的channel传入client.SubEventChan，这个会由protocolV2.messagePump()接收，这个方法核心是下面这个for循环（限于篇幅，我省略了大量无关代码）：
func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
for {
if subChannel == nil || !client.IsReadyForMessages() {
// the client is not ready to receive messages...
memoryMsgChan = nil
backendMsgChan = nil
flusherChan = nil
// force flush
client.writeLock.Lock()
err = client.Flush()
client.writeLock.Unlock()
if err != nil {
goto exit
}
flushed = true
} else if flushed {
// last iteration we flushed...
// do not select on the flusher ticker channel
memoryMsgChan = subChannel.memoryMsgChan
backendMsgChan = subChannel.backend.ReadChan()
flusherChan = nil
}
select {
case subChannel = <-subEventChan:
// you can't SUB anymore
subEventChan = nil
case b := <-backendMsgChan:
if sampleRate > 0 && rand.Int31n(100) > sampleRate {
continue
}
msg, err := decodeMessage(b)
if err != nil {
p.ctx.nsqd.logf(LOG_ERROR, "failed to decode message - %s", err)
continue
}
msg.Attempts++
subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
client.SendingMessage()
err = p.SendMessage(client, msg)
if err != nil {
goto exit
}
flushed = false
case msg := <-memoryMsgChan:
if sampleRate > 0 && rand.Int31n(100) > sampleRate {
continue
}
msg.Attempts++
subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
client.SendingMessage()
err = p.SendMessage(client, msg)
if err != nil {
goto exit
}
flushed = false
case <-client.ExitChan:
goto exit
}
}
}客户端建立连接初始，subChannel为空，循环一直走第1个if语句。直到客户端调用SUB指令，select语句执行"case subChannel = <-subEventChan:"，此时subChannel非空，接下来backendMsgChan和memoryMsgChan被赋值，此后开始推送消息：
- 消息会随机从内存和磁盘队列取，因为如果内存和磁盘都有数据，select是随机的
- 消息通过protocolV2.SendMessage()推送给消费者
当多个消费者订阅同一个channel时情况会如何？
上面我们提到消费者发起SUB指令订阅消息，protocolV2.SUB()会将chan传给protocolV2.messagePump()，即这一行“client.SubEventChan <- channel”，那么我们来看下这个channel变量怎么来的：
func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error) {
...
topic := p.ctx.nsqd.GetTopic(topicName)
channel = topic.GetChannel(channelName)
...
client.SubEventChan <- channelSUB方法包含多种逻辑：
- 当channel不存在时，topic.GetChannel()方法自动创建并与这个消费者绑定
- 当channel存在，比如事先通过http-api创建好了，但没有消费者订阅，则当前消费者独立绑定这个channel
- 当channel存在，且已经有消费者订阅了，topic.GetChannel()方法依然会返回这个channel，这时就有多个消费者同时订阅了这个channel，大家共用一个通道chan变量
由于是多个消费者共用一个通道chan变量，每个消费者都有一个for select在循环监听这个通道，根据chan变量的特性，消费会随机发送给一位消费者，且一条消息只会推送给一个消费者。
消费超时处理
protocolV2.messagePump()方法，无论是“case b := <-backendMsgChan:”还是“case msg := <-memoryMsgChan:”，在向消费者推送消息前都调用了下面这行代码：
func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout) // 省略其他代码
}
func (c *Channel) StartInFlightTimeout(msg *Message, clientID int64, timeout time.Duration) error {
msg.pri = now.Add(timeout).UnixNano() // pri成员保存本消息超时时间
err := c.pushInFlightMessage(msg)
c.addToInFlightPQ(msg)
}channel.StartInFlightTimeout()将消息保存到channel的inFlightMessages和inFlightPQ队列中，这两个缓存是用来处理消费超时的。
值得注意的一个小细节是c.addToInFlightPQ(msg)将msg压入最小堆时，将msg在数组的偏移量保存到了msg.index成员中（最小堆底层是数组实现）
我们先简单看下FIN指令会做啥：
func (p *protocolV2) FIN(client *clientV2, params [][]byte) ([]byte, error) {
err = client.Channel.FinishMessage(client.ID, *id) // 省略其他代码
}
func (c *Channel) FinishMessage(clientID int64, id MessageID) error {
// 省略其他代码
msg, err := c.popInFlightMessage(clientID, id)
c.removeFromInFlightPQ(msg)
}FIN的动作比较简单，主要就是调用channel.FinishMessage()方法把上面写入超时缓存的msg给删除掉。
FIN从inFlightMessages中删除消息比较容易，这是个map，key是msg.id。客户端发送FIN消息时附带了msg.id。但如何从最小堆inFlightPQ中删除对应的msg呢？前面提到在入堆时的一个细节，即保存了msg的偏移量，此时正好用上。通过msg.index直接定位到msg的位置并调整堆即可。
说了这么多，最小堆的作用是啥？别急，接下来我们看下超时逻辑：
超时逻辑由程序启动时开启的工作线程组来处理，即NSQD.queueScanLoop()方法：
func (n *NSQD) queueScanLoop() {
n.resizePool(len(channels), workCh, responseCh, closeCh)
for {
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
}
num := n.getOpts().QueueScanSelectionCount
if num > len(channels) {
num = len(channels)
}
loop:
for _, i := range util.UniqRands(num, len(channels)) {
workCh <- channels[i] // 触发协程组工作
}
numDirty := 0
for i := 0; i < num; i++ {
if <-responseCh {
numDirty++
}
}
if float64(numDirty)/float64(num) > n.getOpts().QueueScanDirtyPercent {
goto loop
}
}
}NSQD.queueScanLoop()方法主要有一个for循环，内层是一个select和一个loop循环。select中，第1个定时器case <-workTicker.C的作用是定时触发工作，只有这个case会跳出select走到下面的loop。第2个定时器负责启动工作协程组并动态调整协程数量，我们来看下第2个定时器调用的resizePool()方法：
func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) {
idealPoolSize := int(float64(num) * 0.25) // 协程数量设定为channel数的1/4
if idealPoolSize < 1 {
idealPoolSize = 1
} else if idealPoolSize > n.getOpts().QueueScanWorkerPoolMax {
idealPoolSize = n.getOpts().QueueScanWorkerPoolMax
}
for {
if idealPoolSize == n.poolSize { // 当协程数量达到协程数量设定为channel数的1/4时，退出
break
} else if idealPoolSize < n.poolSize { // 否则如果当前协程数大于目标值，则通过closeCh通知部分协程退出
// contract
closeCh <- 1
n.poolSize--
} else { // 否则协程数不够，则启动新的协程
// expand
n.waitGroup.Wrap(func() {
n.queueScanWorker(workCh, responseCh, closeCh)
})
n.poolSize++
}
}
}resizePool()方法上面的注释已经说的很清楚了，作用就是保持工作协程数量为当前channel数的1/4。
接下来我们看具体的工作逻辑，queueScanWorker()方法：
func (n *NSQD) queueScanWorker(workCh chan *Channel, responseCh chan bool, closeCh chan int) {
for {
select {
case c := <-workCh:
now := time.Now().UnixNano()
dirty := false
if c.processInFlightQueue(now) {
dirty = true
}
if c.processDeferredQueue(now) {
dirty = true
}
responseCh <- dirty
case <-closeCh:
return
}
}
}queueScanWorker()方法的代码很短，一是监听closeCh的退出信号；二是监听workCh的工作信号。workCh会将需要处理的channel传入，然后调用processInFlightQueue()清理超时的消息，调用processDeferredQueue()清理到时间的延时消息：
func (c *Channel) processInFlightQueue(t int64) bool {
dirty := false
for {
c.inFlightMutex.Lock()
msg, _ := c.inFlightPQ.PeekAndShift(t)
c.inFlightMutex.Unlock()
if msg == nil {
goto exit
}
dirty = true
_, err := c.popInFlightMessage(msg.clientID, msg.ID)
if err != nil {
goto exit
}
atomic.AddUint64(&c.timeoutCount, 1)
c.RLock()
client, ok := c.clients[msg.clientID]
c.RUnlock()
if ok {
client.TimedOutMessage()
}
c.put(msg)
}
exit:
return dirty
}
func (pq *inFlightPqueue) PeekAndShift(max int64) (*Message, int64) {
if len(*pq) == 0 {
return nil, 0
}
x := (*pq)[0]
if x.pri > max {
return nil, x.pri - max
}
pq.Pop()
return x, 0
}前面提到msg.pri成员保存本消息超时时间，所以PeekAndShift()返回的是最小堆里已经超时且超时时间最长的那条消息。processInFlightQueue()则将消息从超时队列中删，同时将消息重新put进channel。注意此时超时的消息put进channel后实际是排在队尾的，消费顺序将发生改变。
processInFlightQueue()方法如果存在超时消息，返回值dirty标识true。queueScanWorker()将dirty写入responseCh。再往回看，queueScanLoop()方法统计了dirty的数量，超过一定比例会继续执行loop，而不是等待下一次定时执行。
4.2 延迟消费
生产者调用DPUB发布的消息，可以指定延时多少再推送给消费者。
我们来看下DPUB的逻辑：
func (p *protocolV2) DPUB(client *clientV2, params [][]byte) ([]byte, error) {
timeoutMs, err := protocol.ByteToBase10(params[2])
timeoutDuration := time.Duration(timeoutMs) * time.Millisecond
msg := NewMessage(topic.GenerateID(), messageBody)
msg.deferred = timeoutDuration
err = topic.PutMessage(msg)
}从上面截取的PUB()方法代码可以看出，DPUB的消息会将延时时间写入msg.deferred成员。4.1章节第2部分介绍的Topic.messagePump()方法有下面这段：
func (t *Topic) messagePump() {
if chanMsg.deferred != 0 {
channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
continue
}
}当chanMsg.deferred != 0时表示延时消息，此时不是直接调用putMessage()方法写入channel，而是调用channel.PutMessageDeferred(chanMsg, chanMsg.deferred)，消息被写入了延时队列Channel.deferredMessages和Channel.deferredPQ。之后的逻辑是在工作协程组NSQD.queueScanLoop()中被识别并put进channel，这与超时的处理逻辑是一样的，不展开说。
加载全部内容