【源码阅读】Nsqd
Nsqd源码阅读
简介
nsqd为nsq daemon的简写，是nsq组件最主要的服务。
nsqd提供一个tcp服务、一个http服务以及一个可选的https服务，tcp服务于客户端(生产者或消费者)，http则提供API(可用于创建、删除topic与channel，生产数据，清空数据等)。
初始化
nsqd的启动入口为apps/nsqd/nsqd.go文件里的main函数。
首先定义了一个program的结构体，用于对程序的控制。结构体内元素为指向NSQD的指针。
main函数里面定义了一个具体的prg，然后Run它。
Run函数负责启动prg并阻塞，直至接收到对应的信号（对于nsqd为SIGINT或者SIGTERM信号）。
type program struct {
nsqd *nsqd.NSQD
}
func main() {
//定义一个具体的prg，并启动它
prg := &program{}
if err := svc.Run(prg, syscall.SIGINT, syscall.SIGTERM); err != nil {
log.Fatal(err)
}
}
Run函数会调用program的Start方法，并调用Main()来启动nsqd服务。
func (p *program) Start() error {
opts := nsqd.NewOptions()
... //配置读取过程，会修改opt
nsqd := nsqd.New(opts)
... //Metadata的处理，以后再说
nsqd.Main() //启动nsqd服务
p.nsqd = nsqd
return nil
}
启动
nsqd首先启动一个Tcp服务、一个Http服务以及一个可选的Https服务，然后调用queueScanLoop函数来处理in-flight与defered数据。
func (n *NSQD) Main() {
var httpListener net.Listener
var httpsListener net.Listener
ctx := &context{n}
//连续启动Tcp、Https、Http服务
tcpListener, err := net.Listen("tcp", n.getOpts().TCPAddress)
...
if n.tlsConfig != nil && n.getOpts().HTTPSAddress != "" {
httpsListener, err = tls.Listen("tcp", n.getOpts().HTTPSAddress, n.tlsConfig)
...
}
httpListener, err = net.Listen("tcp", n.getOpts().HTTPAddress)
...
n.waitGroup.Wrap(func() { n.queueScanLoop() })
n.waitGroup.Wrap(func() { n.lookupLoop() })
if n.getOpts().StatsdAddress != "" {
n.waitGroup.Wrap(func() { n.statsdLoop() })
}
}
客户端连接
客户端连接nsqd的tcp server以后，nsqd会启动一个IOLoop，IOLoop里面首先启动messagePump，然后启动循环处理后续请求。
messagePump负责将Channel里面的消息取出来，并push给客户端。
func (p *protocolV2) IOLoop(conn net.Conn) error {
var err error
var line []byte
var zeroTime time.Time
clientID := atomic.AddInt64(&p.ctx.nsqd.clientIDSequence, 1)
client := newClientV2(clientID, conn, p.ctx)
// messagePump初始化会用到client的一些参数，这里的messagePumpStartedChan保证了初始化完成以后才会接收新的请求，避免了IDENTIFY请求对client的参数可能进行的修改。
messagePumpStartedChan := make(chan bool)
go p.messagePump(client, messagePumpStartedChan)
<-messagePumpStartedChan
for {
...
//读取下一次请求
line, err = client.Reader.ReadSlice('\n')
...
params := bytes.Split(line, separatorBytes)
p.ctx.nsqd.logf(LOG_DEBUG, "PROTOCOL(V2): [%s] %s", client, params)
//处理请求
response, err = p.Exec(client, params)
...
if response != nil {
//发送响应
err = p.Send(client, frameTypeResponse, response)
...
}
}
...
}
数据生产
topic创建
调用http的”/topic/create”接口、”/pub”接口，tcp的SUB/PUB请求等都会触发topic的创建。
创建topic位于NSQD的GetTopic方法。
首先使用读锁判断topic是否存在，如果存在则直接返回；如果不存在，则加写锁，然后调用NewTopic函数创建新的topic。
func (n *NSQD) GetTopic(topicName string) *Topic {
//读锁判断topic是否存在，如果存在则直接返回
// most likely, we already have this topic, so try read lock first.
n.RLock()
t, ok := n.topicMap[topicName]
n.RUnlock()
if ok {
return t
}
n.Lock()
//获取写锁后再次判断topic是否存在，如果存在则直接返回
t, ok = n.topicMap[topicName]
if ok {
n.Unlock()
return t
}
deleteCallback := func(t *Topic) {
n.DeleteExistingTopic(t.name)
}
//创建topic
t = NewTopic(topicName, &context{n}, deleteCallback)
n.topicMap[topicName] = t
n.logf(LOG_INFO, "TOPIC(%s): created", t.name)
// release our global nsqd lock, and switch to a more granular topic lock while we init our
// channels from lookupd. This blocks concurrent PutMessages to this topic.
t.Lock()
n.Unlock()
// 使用lookup的相关处理，如果不使用可以先忽略
...
t.Unlock()
// 触发messagePump更新channel状态
select {
case t.channelUpdateChan <- 1:
case <-t.exitChan:
}
return t
}
生产消息
调用http的”/pub”接口，或者tcp的PUB操作，都可以将消息发送给nsqd，nsqd首先将消息存入topic中作为过渡。
两种处理过程分别在(s httpServer)的doPUB方法与(p
protocolV2)的PUB方法，二者殊途同归，都会调用(t *Topic)的PutMessage方法，将消息写入topic中。
func (t *Topic) PutMessage(m *Message) error {
...
succ, err := t.put(m)
...
}
func (t *Topic) put(m *Message) (succ bool, err error) {
if t.putMode == PUTMODE_NORMAL{
select {
//将消息写入到Topic的memoryMsgChan
case t.memoryMsgChan <- m:
default:
t.put2Disk(m)
}
}else{
...
}
}
数据消费
channel创建
消费者的“订阅”(SUB)请求会触发channel的创建，在tcp服务的SUB处理里面。
func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error) {
...
topicName := string(params[1])
...
channelName := string(params[2])
...
// 防止该topic或者channel是正处于退出状态
var channel *Channel
for {
//获取或者创建topic
topic := p.ctx.nsqd.GetTopic(topicName)
//获取或者创建channel
channel = topic.GetChannel(channelName)
channel.AddClient(client.ID, client)
if (channel.ephemeral && channel.Exiting()) || (topic.ephemeral && topic.Exiting()) {
channel.RemoveClient(client.ID)
time.Sleep(1 * time.Millisecond)
continue
}
break
}
atomic.StoreInt32(&client.State, stateSubscribed)
client.Channel = channel
// 将channel告知client
client.SubEventChan <- channel
return okBytes, nil
}
topic的消息复制到channel
topic创建时调用的NewTopic函数会启动messagePump函数，负责更新channel，并将topic中的消息复制到所有channel。
// messagePump selects over the in-memory and backend queue and
// writes messages to every channel for this topic
func (t *Topic) messagePump() {
//获取所有channel
t.RLock()
for _, c := range t.channelMap {
chans = append(chans, c)
}
t.RUnlock()
if len(chans) > 0 {
memoryMsgChan = t.memoryMsgChan
limitedMsgChan = t.limitedMsgChan
backendChan = t.backend.ReadChan()
}
for {
select {
//接收消息
case msg = <-memoryMsgChan:
if msg == nil {
continue
}
...
//更新channel
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 {
limitedMsgChan = t.limitedMsgChan
memoryMsgChan = t.memoryMsgChan
backendChan = t.backend.ReadChan()
}
continue
...
case 0 {
chanMsg = NewMessage(msg.ID,msg.MsgType, msg.Body)
chanMsg.Timestamp = msg.Timestamp
chanMsg.deferred = msg.deferred
}
...
//将消息存储到channel
//此处的PutMessage与Topic的同名方法类似，也是将消息写到channel的memoryMsgChan
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)
}
}
}
exit:
t.ctx.nsqd.logf(LOG_INFO, "TOPIC(%s): closing ... messagePump", t.name)
}
//PutMessage会调用到c.put(msg)
func (c *Channel) put(m *Message) (succ bool, err error) {
succ = true
if c.putMode == PUTMODE_NORMAL {
select {
//将数据发送到Channel的memoryMsgChan
case c.memoryMsgChan <- m:
default:
...
}
}else{
...
}
}
数据push到消费者
回忆下前面介绍的两点：一是客户端连接时，会启动messagePump负责将Channel里面的消息取出来，并push给客户端；二是channel创建时，会将创建的channel告知client。
messagePump获取创建的这个channel，并从channel的memoryMsgChan接收消息，然后push给消费者。
func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
...
//channel创建成功后，通过SubEventChan告知client
subEventChan := client.SubEventChan
for {
if subChannel == nil || !client.IsReadyForMessages() {
...
} else if flushed {
...
} else {
// 获取channel的memoryMsgChan
memoryMsgChan = subChannel.memoryMsgChan
...
}
select {
...
case subChannel = <-subEventChan:
p.ctx.nsqd.logf(LOG_INFO, "get subEventChan:%+v", subChannel)
// you can't SUB anymore
subEventChan = nil
...
//从memoryMsgChan里接收消息，并push给客户端
case msg := 0 && rand.Int31n(100) > sampleRate {
continue
}
msg.Attempts++
//将消息放入in-flight队列
subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
client.SendingMessage()
p.ctx.nsqd.logf(LOG_INFO, "get memory msg:%+v", msg)
//将消息push给消费者
err = p.SendMessage(client, msg, &buf)
if err != nil {
goto exit
}
flushed = false
...
case <-client.ExitChan:
goto exit
}
}
exit:
...
}
in-flight数据与deferred数据处理
回一下前面讲到两点：一是queueScanLoop函数会处理in-flight数据与deferred数据；二是消息push给消费者之前会调用StartInFlightTimeout将该消息放入in-flight队列。
queueScanLoop管理一个queueScanWorker pool（默认大小为4），各个worker并发处理channel数据。
in-flight数据的存储与清理
in-flight数据存储时会记录下该消息的到期时间，以便到期后将该消息重新push给消费者。
func (c *Channel) StartInFlightTimeout(msg *Message, clientID int64, timeout time.Duration) error {
now := time.Now()
msg.clientID = clientID
msg.deliveryTS = now
//存储到期时间
msg.pri = now.Add(timeout).UnixNano()
err := c.pushInFlightMessage(msg)
if err != nil {
return err
}
c.addToInFlightPQ(msg)
return nil
}
如果消费者成功接收，则会回应一个”FIN”，nsqd收到”FIN”则将该消息从in-flight队列中清除。
func (p *protocolV2) FIN(client *clientV2, params [][]byte) ([]byte, error) {
...
id, err := getMessageID(params[1])
if err != nil {
return nil, protocol.NewFatalClientErr(nil, "E_INVALID", err.Error())
}
//将该消息从in-flight队列中清除
err = client.Channel.FinishMessage(client.ID, *id)
if err != nil {
return nil, protocol.NewClientErr(err, "E_FIN_FAILED",
fmt.Sprintf("FIN %s failed %s", *id, err.Error()))
}
client.FinishedMessage()
return nil, nil
}
defered数据的存储与清理
如果消费者收到消息以后，如果一时间自己处理不过来，可以通过”REQ”将该消息重新入队，并可以设定多长时间后重新消费，时间为0的话则立即消费，否则延迟消费。
延迟消费的处理方式与in-flight数据类似，也是先写入到一个队列，并设定到期时间，等待重新读取。
下面介绍这两部分数据时如何重新消费的，主要是queueScanLoop的处理逻辑。
worker的创建与销毁
worker的创建与销毁是在resizePool函数。
worker的完美个数为channel总数的四分之一，但是不能大于QueueScanWorkerPoolMax。
1 <= pool <= min(num * 0.25, QueueScanWorkerPoolMax)
所有的worker都会监听同一个workCh、closeCh，如果worker过多，则只需要向closeCh写入一个“通知”，收到这个“通知”的worker就会被销毁。
一次for循环只创建或销毁一个worker，直至worker数目达到idealPoolSize。
func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) {
idealPoolSize := int(float64(num) * 0.25)
if idealPoolSize n.getOpts().QueueScanWorkerPoolMax {
idealPoolSize = n.getOpts().QueueScanWorkerPoolMax
}
for {
if idealPoolSize == n.poolSize {
break
} else if idealPoolSize < n.poolSize {
// contract
closeCh <- 1
n.poolSize--
} else {
// expand
n.waitGroup.Wrap(func() {
n.queueScanWorker(workCh, responseCh, closeCh)
})
n.poolSize++
}
}
}
channel的选择
queueScanLoop的处理方法模仿了Redis的概率到期算法(probabilistic expiration algorithm)：每过一个QueueScanInterval(默认100ms)间隔，进行一次概率选择，从所有的channel缓存中随机选择QueueScanSelectionCount(默认20)个channel，如果某个被选中channel的任何一个queue有事可做，则认为该channel为“脏”channel。如果被选中channel中“脏”channel的比例大于QueueScanDirtyPercent(默认25%)，则不投入睡眠，直接进行下一次概率选择。
channel缓存每QueueScanRefreshInterval(默认5s)刷新一次。
queueScanLoop与worker的交互
queueScanLoop与worker之间通过workCh与responseCh来进行交互。
- workCh：queueScanLoop随机选择一定数目的channel后，通过workCh告诉worker。
- responseCh：worker处理完成后，通过responseCh反馈该channel是否为“脏”。
func (n *NSQD) queueScanLoop() {
workCh := make(chan *Channel, n.getOpts().QueueScanSelectionCount)
responseCh := make(chan bool, n.getOpts().QueueScanSelectionCount)
closeCh := make(chan int)
workTicker := time.NewTicker(n.getOpts().QueueScanInterval)
refreshTicker := time.NewTicker(n.getOpts().QueueScanRefreshInterval)
//根据channel数目，创建worker
channels := n.channels()
n.resizePool(len(channels), workCh, responseCh, closeCh)
for {
select {
case <-workTicker.C:
if len(channels) == 0 {
continue
}
case <-refreshTicker.C:
//更新channel缓存，并据此创建或者销毁worker
channels = n.channels()
n.resizePool(len(channels), workCh, responseCh, closeCh)
continue
case len(channels) {
num = len(channels)
}
loop:
//随机选择num个channel，并传入workCh
for _, i := range util.UniqRands(num, len(channels)) {
workCh <- channels[i]
}
//等待这num个channel的处理结果(是否为“脏”channel)
numDirty := 0
for i := 0; i < num; i++ {
if n.getOpts().QueueScanDirtyPercent {
goto loop
}
}
exit:
n.logf(LOG_INFO, "QUEUESCAN: closing")
close(closeCh)
workTicker.Stop()
refreshTicker.Stop()
}
worker处理
worker从queueScanLoop接收需要处理的channel，处理该channel的in-flight数据与deferred数据。processInFlightQueue与processDeferredQueue函数都会调用c.put(msg)，将数据发送到Channel的memoryMsgChan，进而重新被push到消费者。
func (n *NSQD) queueScanWorker(workCh chan *Channel, responseCh chan bool, closeCh chan int) {
for {
select {
case c := <-workCh:
now := time.Now().UnixNano()
dirty := false
//处理in-flight消息
if c.processInFlightQueue(now) {
dirty = true
}
//处理defered消息
if c.processDeferredQueue(now) {
dirty = true
}
responseCh <- dirty
case <-closeCh:
return
}
}
}
func (c *Channel) processInFlightQueue(t int64) bool {
c.exitMutex.RLock()
defer c.exitMutex.RUnlock()
if c.Exiting() {
return false
}
dirty := false
for {
c.inFlightMutex.Lock()
//获取超时的消息
msg, _ := c.inFlightPQ.PeekAndShift(t)
c.inFlightMutex.Unlock()
if msg == nil {
goto exit
}
dirty = true
//判断该消息是否属于这个client
_, 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()
}
//将消息重新写入channel
c.put(msg)
}
exit:
return dirty
}
processDeferredQueue的处理与此类似。
About The Author
