内容简介:看完了nsqlookupd我们继续往下看, nsqd才是他的核心. 里面大量的使用到了go channel, 相信看完之后对你学习go有很大的帮助.相较于lookupd部分无论在代码逻辑和实现上都要复杂很多.不过基本的代码结构基本上都是一样的, 进程使用go-srv来管理, Main里启动一个http sever和一个tcp server, 这里可以参考下之前文章的进程模型小节, 不过在nsqd中会启动另外的两个goroutine queueScanLoop和lookupLoop。下面是一个具体的进程模型
看完了nsqlookupd我们继续往下看, nsqd才是他的核心. 里面大量的使用到了go channel, 相信看完之后对你学习 go 有很大的帮助.相较于lookupd部分无论在代码逻辑和实现上都要复杂很多.
不过基本的代码结构基本上都是一样的, 进程使用go-srv来管理, Main里启动一个http sever和一个tcp server, 这里可以参考下之前文章的进程模型小节, 不过在nsqd中会启动另外的两个goroutine queueScanLoop和lookupLoop。下面是一个
具体的进程模型。
后面的分析都是基于这个进程模型。
NSQD的启动
启动时序这块儿大体上和lookupd中的一致, 我们下面来看看lookupLoop和queueScanLoop.
lookupLoop代码见nsqd/lookup.go中 主要做以下几件事情:
- 和lookupd建立连接(这里是一个长连接)
- 每隔15s ping一下lookupd
- 新增或者删除topic的时候通知到lookupd
- 新增或者删除channel的时候通知到lookupd
- 动态的更新options
由于设计到了nsq里的in-flight/deferred message, 我们把queueScanLoop放到最后来看.
一条message的LifeLine
下面我们就通过一条message的生命周期来看下nsqd的工作原理. 根据官方的QuickStart, 我们可以通过curl来pub一条消息.
curl -d 'hello world 1' 'http://127.0.0.1:4151/pub?topic=test'
http handler
我们就跟着代码看一下, 首先是http对此的处理:
// nsq/nsqd/http.go func (s *httpServer) doPUB(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) { ... reqParams, topic, err := s.getTopicFromQuery(req) // 从http query中拿到topic信息 ... }
// nsq/nsqd/http.go func (s *httpServer) getTopicFromQuery(req *http.Request) (url.Values, *Topic, error) { reqParams, err := url.ParseQuery(req.URL.RawQuery) topicNames, ok := reqParams["topic"] return reqParams, s.ctx.nsqd.GetTopic(topicName), nil }
// nsq/nsqd/nsqd.go // GetTopic performs a thread safe operation // to return a pointer to a Topic object (potentially new) func (n *NSQD) GetTopic(topicName string) *Topic { // 1. 首先查看n.topicMap,确认该topic是否已经存在(存在直接返回) t, ok := n.topicMap[topicName] // 2. 否则将新建一个topic t = NewTopic(topicName, &context{n}, deleteCallback) n.topicMap[topicName] = t // 3. 查看该nsqd是否设置了lookupd, 从lookupd获取该tpoic的channel信息 // 这个topic/channel已经通过nsqlookupd的api添加上去的, 但是nsqd的本地 // 还没有, 针对这种情况我们需要创建该channel对应的deffer queue和inFlight // queue. lookupdHTTPAddrs := n.lookupdHTTPAddrs() if len(lookupdHTTPAddrs) > 0 { channelNames, err := n.ci.GetLookupdTopicChannels(t.name, lookupdHTTPAddrs) } // now that all channels are added, start topic messagePump // 对该topic的初始化已经完成下面就是message t.Start() return t }
topic messagePump
在上面消息初始化完成之后就启动了tpoic对应的messagePump
// nsq/nsqd/topic.go // messagePump selects over the in-memory and backend queue and // writes messages to every channel for this topic func (t *Topic) messagePump() { // 1. do not pass messages before Start(), but avoid blocking Pause() // or GetChannel() // 等待channel相关的初始化完成,GetTopic中最后的t.Start()才正式启动该Pump // 2. main message loop // 开始从Memory chan或者disk读取消息 // 如果topic对应的channel发生了变化,则更新channel信息 // 3. 往该tpoic对应的每个channel写入message(如果是deffermessage // 的话放到对应的deffer queue中 // 否则放到该channel对应的memoryMsgChan中)。 }
至此也就完成了从tpoic memoryMsgChan收到消息投递到channel memoryMsgChan的投递, 我们先看下http
收到消息到通知pump处理的过程。
// nsq/nsqd/http.go func (s *httpServer) doPUB(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) { ... msg := NewMessage(topic.GenerateID(), body) msg.deferred = deferred err = topic.PutMessage(msg) if err != nil { return nil, http_api.Err{503, "EXITING"} } return "OK", nil }
// nsq/nsqd/topic.go // PutMessage writes a Message to the queue func (t *Topic) PutMessage(m *Message) error { t.RLock() defer t.RUnlock() if atomic.LoadInt32(&t.exitFlag) == 1 { return errors.New("exiting") } err := t.put(m) if err != nil { return err } atomic.AddUint64(&t.messageCount, 1) return nil }
func (t *Topic) put(m *Message) error { select { case t.memoryMsgChan <- m: default: 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 }
这里memoryMsgChan的大小我们可以通过--mem-queue-size参数来设置,上面这段代码的流程是如果memoryMsgChan还没有满的话
就把消息放到memoryMsgChan中,否则就放到backend(disk)中。topic的mesasgePump检测到有新的消息写入的时候就开始工作了,
从memoryMsgChan/backend(disk)读取消息投递到channel对应的chan中。 还有一点请注意就是messagePump中
if len(chans) > 0 && !t.IsPaused() { memoryMsgChan = t.memoryMsgChan backendChan = t.backend.ReadChan() }
这段代码只有channel(此channel非golang里的channel而是nsq的channel类似nsq_to_file)存在的时候才会去投递。上面部分就是
msg从producer生产消息到吧消息写到memoryChan/Disk的过程,下面我们来看下consumer消费消息的过程。
首先是consumer从nsqlookupd查询到自己所感兴趣的topic/channel的nsqd信息, 然后就是来连接了。
tcp handler
对新的client的处理
//nsq/internal/protocol/tcp_server.go func TCPServer(listener net.Listener, handler TCPHandler, logf lg.AppLogFunc) { go handler.Handle(clientConn) }
//nsq/nsqd/tcp.go func (p *tcpServer) Handle(clientConn net.Conn) { prot.IOLoop(clientConn) }
针对每个client起一个messagePump吧msg从上面channel对应的chan 写入到consumer侧
//nsq/nsqd/protocol_v2.go func (p *protocolV2) IOLoop(conn net.Conn) error { client := newClientV2(clientID, conn, p.ctx) p.ctx.nsqd.AddClient(client.ID, client) messagePumpStartedChan := make(chan bool) go p.messagePump(client, messagePumpStartedChan) // read the request line, err = client.Reader.ReadSlice('\n') response, err = p.Exec(client, params) p.Send(client, frameTypeResponse, response) }
//nsq/nsqd/protocol_v2.go func (p *protocolV2) Exec(client *clientV2, params [][]byte) ([]byte, error) { switch { case bytes.Equal(params[0], []byte("FIN")): return p.FIN(client, params) case bytes.Equal(params[0], []byte("RDY")): return p.RDY(client, params) case bytes.Equal(params[0], []byte("REQ")): return p.REQ(client, params) case bytes.Equal(params[0], []byte("PUB")): return p.PUB(client, params) case bytes.Equal(params[0], []byte("MPUB")): return p.MPUB(client, params) case bytes.Equal(params[0], []byte("DPUB")): return p.DPUB(client, params) case bytes.Equal(params[0], []byte("NOP")): return p.NOP(client, params) case bytes.Equal(params[0], []byte("TOUCH")): return p.TOUCH(client, params) case bytes.Equal(params[0], []byte("SUB")): return p.SUB(client, params) case bytes.Equal(params[0], []byte("CLS")): return p.CLS(client, params) case bytes.Equal(params[0], []byte("AUTH")): return p.AUTH(client, params) } }
//nsq/nsqd/protocol_v2.go func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error) { var channel *Channel topic := p.ctx.nsqd.GetTopic(topicName) channel = topic.GetChannel(channelName) channel.AddClient(client.ID, client) // 通知messagePump开始工作 client.SubEventChan <- channel
通知topic的messagePump开始工作
func (t *Topic) GetChannel(channelName string) *Channel { t.Lock() channel, isNew := t.getOrCreateChannel(channelName) t.Unlock() if isNew { // update messagePump state select { case t.channelUpdateChan <- 1: case <-t.exitChan: } } return channel }
message 对应的Pump
func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) { for { if subChannel == nil || !client.IsReadyForMessages() { // the client is not ready to receive messages... // 等待client ready,并且channel的初始化完成 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 } else { // we're buffered (if there isn't any more data we should flush)... // select on the flusher ticker channel, too memoryMsgChan = subChannel.memoryMsgChan backendMsgChan = subChannel.backend.ReadChan() flusherChan = outputBufferTicker.C } select { case <-flusherChan: // if this case wins, we're either starved // or we won the race between other channels... // in either case, force flush case <-client.ReadyStateChan: case subChannel = <-subEventChan: // you can't SUB anymore // channel初始化完成,pump开始工作 subEventChan = nil case identifyData := <-identifyEventChan: // you can't IDENTIFY anymore case <-heartbeatChan: // heartbeat的处理 case b := <-backendMsgChan: // 1. decode msg // 2. 把msg push到Flight Queue里 // 3. send msg to client case msg := <-memoryMsgChan: // 1. 把msg push到Flight Queue里 // 2. send msg to client case <-client.ExitChan: // exit the routine } }
至此我们看的代码就是一条消息从pub到nsqd中到被消费者处理的过程。不过得注意一点,我们在上面的代码分析中,创建
topic/channel的部分放到了message Pub的链上, 如果是没有lookupd的模式的话这部分是在client SUB链上的。
topic/hannel的管理
在NSQ内部通过
type NSQD struct { topicMap map[string]*Topic } 和 type Topic struct { channelMap map[string]*Channel }
来维护一个内部的topic/channel状态,然后在提供了如下的接口来管理topic和channel
/topic/create - create a new topic /topic/delete - delete a topic /topic/empty - empty a topic /topic/pause - pause message flow for a topic /topic/unpause - unpause message flow for a topic /channel/create - create a new channel /channel/delete - delete a channel /channel/empty - empty a channel /channel/pause - pause message flow for a channel /channel/unpause - unpause message flow for a channel
create topic/channel的话我们在之前的代码看过了,这里可以重点看下topic/channel delete的时候怎样保证数据优雅的删除的,以及
messagePump的退出机制。
queueScanLoop的工作
// queueScanLoop runs in a single goroutine to process in-flight and deferred // priority queues. It manages a pool of queueScanWorker (configurable max of // QueueScanWorkerPoolMax (default: 4)) that process channels concurrently. // // It copies Redis's probabilistic expiration algorithm: it wakes up every // QueueScanInterval (default: 100ms) to select a random QueueScanSelectionCount // (default: 20) channels from a locally cached list (refreshed every // QueueScanRefreshInterval (default: 5s)). // // If either of the queues had work to do the channel is considered "dirty". // // If QueueScanDirtyPercent (default: 25%) of the selected channels were dirty, // the loop continues without sleep.
这里的注释已经说的很明白了,queueScanLoop就是通过动态的调整queueScanWorker的数目来处理
in-flight和deffered queue的。在具体的算法上的话参考了 redis 的随机过期算法。
总结
阅读源码就是走走停停的过程,从一开始的无从下手到后面的一点点的把它啃透。一开始都觉得很困难,无从下手。以前也是尝试着去看一些
经典的开源代码,但都没能坚持下来,有时候人大概是会高估自己的能力的,好多东西自以为看个一两遍就能看懂,其实不然,
好多知识只有不断的去研究你才能参透其中的原理。
一定要持续的读,不然过几天之后就忘了前面读的内容 一定要多总结, 总结就是在不断的读的过程,从第一遍读通到你把它表述出来至少需要再读5-10次 多思考,这段时间在地铁上/跑步的时候我会回向一下其中的流程 分享(读懂是一个层面,写出来是一个层面,讲给别人听是另外一个层面)
后面我会先看下go-nsqd部分的代码,之后会研究下gnatsd, 两个都是cloud native的消息系统,看下有啥区别。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:- 【源码阅读】AndPermission源码阅读
- ReactNative源码解析-初识源码
- 【源码阅读】Gson源码阅读
- Spring源码系列:BeanDefinition源码解析
- istio 源码 – Citadel 源码分析 (原创)
- istio 源码 – pilot 源码分析(原创)
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
Java数据结构和算法
拉佛 / 计晓云 / 中国电力出版社 / 2004-02-01 / 55.00元
《Java数据结构和算法》(第2版)以一种易懂的方式教授如何安排和操纵数据的问题,其中不乏一些难题:了解这些知识以期使计算机的应用获得最好的表现。不管使用何种语言或平台,掌握了数据结构和算法将改进程序的质量和性能。 《Java数据结构和算法》(第2版)提供了一套独创的可视讨论专题用以阐明主要的论题:它使用Java语言说明重要的概念,而避免了C/C++语言的复杂性,以便集中精力论述数据结构和算法。经......一起来看看 《Java数据结构和算法》 这本书的介绍吧!