nsq源码阅读(6)-全局唯一MessageID

//客户端推送消息
func (p *protocolV2) PUB(client *clientV2, params [][]byte) ([]byte, error) {
	//创建 Message 对象
	msg := NewMessage(<-p.ctx.nsqd.idChan, messageBody)

	//topic 发送消息
	err = topic.PutMessage(msg)
}

func (n *NSQD) Main() {
	n.waitGroup.Wrap(func() {
		n.idPump()
	})
}

// 一直生产 序列化 id
func (n *NSQD) idPump() {
	factory := &guidFactory{}
	lastError := time.Unix(0, 0)
	//因为可以做成 nsqd 集群, 所以 msgid 关联 nsq id
	workerID := n.getOpts().ID
	for {
		id, err := factory.NewGUID(workerID)
		if err != nil {
			now := time.Now()
			if now.Sub(lastError) > time.Second {
				// only print the error once/second
				n.logf("ERROR: %s", err)
				lastError = now
			}
			runtime.Gosched()
			continue
		}

		//这里利用了 golang 的chan 满了阻塞的特性,
		//如果n.idChan 这个id缓存池满了, 就阻塞不会继续生产
		select {
		case n.idChan <- id.Hex():
		case <-n.exitChan:
			goto exit
		}
	}

	exit:
	n.logf("ID: closing")
}

func (f *guidFactory) NewGUID(workerID int64) (guid, error) {
	// divide by 1048576, giving pseudo-milliseconds
	ts := time.Now().UnixNano() >> 20

	if ts < f.lastTimestamp {
		return 0, ErrTimeBackwards
	}

	if f.lastTimestamp == ts {
		f.sequence = (f.sequence + 1) & sequenceMask
		if f.sequence == 0 {
			return 0, ErrSequenceExpired
		}
	} else {
		f.sequence = 0
	}

	f.lastTimestamp = ts

	// id = [ 37位ts + ?位 workerId + 12位 sequence ]
	id := guid(((ts - twepoch) << timestampShift) |
		(workerID << workerIDShift) |
		f.sequence)

	if id <= f.lastID {
		return 0, ErrIDBackwards
	}

	f.lastID = id

	return id, nil
}

if opts.ID < 0 || opts.ID >= 1024 {
	n.logf("FATAL: --worker-id must be [0,1024)")
	os.Exit(1)
}

for {
	id, err := factory.NewGUID(workerID)
	if err != nil {
		// 生成的太快了, 一个ts 中 生成了 4096个 id(sequence)
		// ts + sequence 满了, 休息一会, 让ts 发生改变
		runtime.Gosched()
		continue
	}
}