nsqd入口分析

[loadlj]

Main入口分析

nsqd的入口调用了go-svc的Run方法，这里的go-svc有点类似于watchdog监测用户signal的行为，可以很好的兼容Linux和Windows。svc_other里面声明的signalNotify是从signal.Notify这里拿到的，这个函数的主要作用是监听用户输入的signal。
Main还声明了一个tcpListener和一个httpListener，tcpListener是用来处理跟client的连接的。

	tcpServer := &tcpServer{ctx: ctx}
	n.waitGroup.Wrap(func() {
		protocol.TCPServer(n.tcpListener, tcpServer, n.logf)
	})

tcpServer有一个handle方法，用来处理tcp的连接的。里面只会接受" V2"的protocol，然后对每个连接进行IOLoop。
IOLoop会对每个client先进行messagePump

messagePumpStartedChan := make(chan bool)
go p.messagePump(client, messagePumpStartedChan)
<-messagePumpStartedChan

messagePump主要做的事：发送心跳包给客户端，获取channel中的消息，发送给客户端。

case <-heartbeatChan:
			err = p.Send(client, frameTypeResponse, heartbeatBytes)
			if err != nil {
				goto exit
			}
case b := <-backendMsgChan:
case msg := <-memoryMsgChan:

然后在IOLoop中，有新的for循环去监听客户端的消息并进行处理。这里大概理一下，后面详细分析

回到Main中，后面新开了两个goroutine，一个queueScanLoop处理消息，一个lookupLoop处理和nsqlookupd的连接的。

n.waitGroup.Wrap(func() { n.queueScanLoop() })
n.waitGroup.Wrap(func() { n.lookupLoop() })

queueScanLoop

• workTicker
  workerTicker是一个主动去刷新queue的ticker
• refreshTicker
  refreshTicker是一个被动的ticker，每间隔5s去刷新一次，参照了redis的过期刷新算法。

随机挑选若干个key，删除所有过期的，如果过期的key的占比大于25%，则继续循环删除
在nsqd里面的定义是如果一个channel还有work在处理则认为是"dirty",判断dirty的占比如果大于QueueScanDirtyPercent就会继续循环给workCh发送channel。

resizePool

func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) {
	idealPoolSize := int(float64(num) * 0.25)
	if idealPoolSize < 1 {
		idealPoolSize = 1
	} else 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++
		}
	}
}

resizePool是用来动态调整queueScanWorker的，idealPoolSize是当前的num(channels) / 4，会跟n.poolSize去做比较，这里表示的是queueScanWorker的数量。如果idealPoolSize要比当前的workpool小，再去新起一个queueScanWorker

queueScanWorker

在nsq中inFlight指的是正在投递但还没确认投递成功的消息，defferred指的是投递失败，等待重新投递的消息。 initPQ创建的字典和队列主要用于索引和存放这两类消息。其中两个字典使用消息ID作索引。