1：使用TCP协议获取

 1V2
 2
 3IDENTIFY
 4
 5SUB is_test2 ch
 6
 7RDY 1
 8
 9RDY 1
10
11FIN 07ee6e602a3fa000

 1func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error) {
 2    ......
 3    topic := p.ctx.nsqd.GetTopic(topicName)
 4    channel := topic.GetChannel(channelName)
 5    channel.AddClient(client.ID, client)
 6    
 7    atomic.StoreInt32(&client.State, stateSubscribed)
 8    client.Channel = channel
 9    // update message pump
10    client.SubEventChan <- channel
11    ......
12}
13

 1func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
 2
 3    for {
 4        ......
 5    	select {
 6        ......
 7    	case subChannel = <-subEventChan:
 8    		// you can't SUB anymore
 9    		subEventChan = nil
10        ......
11    	}
12    }
13    ......
14}

 1func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
 2
 3    for {
 4    
 5        if subChannel == nil || !client.IsReadyForMessages() {
 6        .....
 7        } else if flushed {
 8            // last iteration we flushed...
 9            // do not select on the flusher ticker channel
10            // 监听 channel
11            memoryMsgChan = subChannel.memoryMsgChan
12            backendMsgChan = subChannel.backend.ReadChan()
13            flusherChan = nil
14        } else {
15            ......
16        }
17    
18        ......
19    	select {
20    	......
21    	//没有看出作用
22    	case <-client.ReadyStateChan:
23        ......
24    	case msg := <-memoryMsgChan:
25            if sampleRate > 0 && rand.Int31n(100) > sampleRate {
26            	continue
27            }
28            msg.Attempts++
29            
30            subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
31            client.SendingMessage()
32            err = p.SendMessage(client, msg, &buf)
33            if err != nil {
34            	goto exit
35            }
36            flushed = false
37            ......
38    	}
39    }
40    ......
41}