nsq源码-nsqd总体流程
2023-05-03 10:48:41 # Go # 源码阅读

nsq源码-nsqd总体流程

看了一阵子nsq源码,出去细节大体的流程基本算是看明白了
下面来总结一下,看下面的图

个人觉得其实只要搞清楚几个Loop,我称为“守护”协程,也就是一直在for… select… 里面跑着不退出的几个关键函数。
看上面的图,有几个呢?

从底层依次往上看

  1. diskqueue.IoLoop(): 我觉得这个第一个要看懂,topic和channel都会用到他来作为磁盘存储消息,关键消息在里面怎么进怎么出,这里也要搞清楚。

  2. Topic.MessagePump(): 不断的将topci收到的消息,分发给底下所有的Channel。

  3. protocolV2.IOLoop(): 处理客户端通过socket传上来的指令,然后分发处理指令。比如收到PUB指令,将消息丢给topic

  4. protocolV2.messagePump(): 不断的将Channel.memoryMsgChan或者diskqueue里面的消息通过socket推给客户端订阅者

  5. nsqd.queueScanLoop(): 其实看上面的协程都是成对出现的,也就是消息有进去协程,就应该有消息出来的协程。但是Channel有消息进去的协程,也就是topic的MessagePump这个协程负责将消息给Channel,那么Channel消息出来的协程就是这个nsqd.queueScanLoop()了。他负责将Inflight也就是处理中的消息,推到channel.memoryMsgChan或者diskqueue中。

一个完整的消息进出流程大概就是这样子

客户端生产者 -> protocolV2.IOLoop() 收到消息
-> Topic.PutMessage() 消息传到Topic
-> Topic.MessagePump() 消息传到Channel.InFlight
-> nsqd.queueScanLoop()
-> 消息传到Channel.memoryMsgChan
-> protocolV2.messagePump()
-> 客户端订阅者

这里有一个很容易搞晕的地方,就是Channel中放消息其实有两块

1
2
3
4
5
6
7
8
9
10
11
12
13
type Channel struct {
... //这一块是存储和持久化,消息最终是从这里出去
backend BackendQueue
memoryMsgChan chan *Message

... //这一块可以理解成一个缓存,意思在处理中消息,消息首先到这里
deferredMessages map[MessageID]*pqueue.Item
deferredPQ pqueue.PriorityQueue
deferredMutex sync.Mutex
inFlightMessages map[MessageID]*Message
inFlightPQ inFlightPqueue
inFlightMutex sync.Mutex
}