nsq源码分析（2）：nsqlookup之tcp服务

nsq源码分析（2）：nsqlookup之tcp服务

本章涉及db的读写操作，请参考 nsqlookup之RegistrationDB数据库

通信协议

本章内容涉及tcp协议的封包解包内容，请参考 nsq tcp协议规范

[ ][ ][V][1]

client连接后，客户端必须发送一个4 字节的 “magic” 标识码来选择通讯协议的版本。

启动tcp服务

nsqlookupd/nsqlookupd.go

// 启动tcp服务
tcpListener, err := net.Listen("tcp", l.opts.TCPAddress)
if err != nil {
l.logf("FATAL: listen (%s) failed - %s", l.opts.TCPAddress, err)
os.Exit(1)
}
l.Lock()
l.tcpListener = tcpListener
l.Unlock()
tcpServer := &tcpServer{ctx: ctx}
// 封装的waitGroup，内部使用goroutine启动该服务，使用waitGroup守护改协程直到退出
l.waitGroup.Wrap(func() {
protocol.TCPServer(tcpListener, tcpServer, l.opts.Logger)
})

接收client连接并回调handle方法

internal/protocol/tcp_server.go

type TCPHandler interface {
Handle(net.Conn)
}

func TCPServer(listener net.Listener, handler TCPHandler, l app.Logger) {
l.Output(2, fmt.Sprintf("TCP: listening on %s", listener.Addr()))

// 接收client连接并回调handle方法
for {
clientConn, err := listener.Accept()
if err != nil {
if nerr, ok := err.(net.Error); ok && nerr.Temporary() {
l.Output(2, fmt.Sprintf("NOTICE: temporary Accept() failure - %s", err))
// runtime.Gosched让出cpu时间片，让另一时间片处理
runtime.Gosched()
continue
}
// theres no direct way to detect this error because it is not exposed
if !strings.Contains(err.Error(), "use of closed network connection") {
l.Output(2, fmt.Sprintf("ERROR: listener.Accept() - %s", err))
}
break
}
// 这里的handle方法虽然是TCPHandler的接口类型，实际回调的是nsqlookupd/tcp.go中Handle方法
go handler.Handle(clientConn)
}

l.Output(2, fmt.Sprintf("TCP: closing %s", listener.Addr()))
}

解析协议版本和命令

解析协议版本

nsqlookupd/tcp.go

// 回调handle：解析协议版本
func (p *tcpServer) Handle(clientConn net.Conn) {
p.ctx.nsqlookupd.logf("TCP: new client(%s)", clientConn.RemoteAddr())

// The client should initialize itself by sending a 4 byte sequence indicating
// the version of the protocol that it intends to communicate, this will allow us
// to gracefully upgrade the protocol away from text/line oriented to whatever...
// 读取tcp流协议的前四个字节，用于选择使用哪种协议
buf := make([]byte, 4)
_, err := io.ReadFull(clientConn, buf)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: failed to read protocol version - %s", err)
return
}
protocolMagic := string(buf)

p.ctx.nsqlookupd.logf("CLIENT(%s): desired protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)

// 作者为了nsq的可扩展性，可支持多种协议，目前只支持V1版本
var prot protocol.Protocol
switch protocolMagic {
case "  V1":
prot = &LookupProtocolV1{ctx: p.ctx}
default:
// 其他协议则发送E_BAD_PROTOCOL并关闭连接并
protocol.SendResponse(clientConn, []byte("E_BAD_PROTOCOL"))
clientConn.Close()
p.ctx.nsqlookupd.logf("ERROR: client(%s) bad protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)
return
}

// 使用对应的协议处理client请求
err = prot.IOLoop(clientConn)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: client(%s) - %s", clientConn.RemoteAddr(), err)
return
}
}

解析命令

nsqlookupd/lookup_protocol_v1.go

// IOLoop中不断的for循环等待用户的消息
func (p *LookupProtocolV1) IOLoop(conn net.Conn) error {
var err error
var line string

// 实例化V1协议
client := NewClientV1(conn)
reader := bufio.NewReader(client)
for {
// 每次读取一行
line, err = reader.ReadString('\n')
if err != nil {
break
}

// 解析出数据中的params
line = strings.TrimSpace(line)
params := strings.Split(line, " ")

var response []byte
response, err = p.Exec(client, reader, params)
if err != nil {
// Exec过程中出现err则将err发送给client
ctx := ""
if parentErr := err.(protocol.ChildErr).Parent(); parentErr != nil {
ctx = " - " + parentErr.Error()
}
p.ctx.nsqlookupd.logf("ERROR: [%s] - %s%s", client, err, ctx)

_, sendErr := protocol.SendResponse(client, []byte(err.Error()))
if sendErr != nil {
p.ctx.nsqlookupd.logf("ERROR: [%s] - %s%s", client, sendErr, ctx)
break
}

// errors of type FatalClientErr should forceably close the connection
// 如果err为Fatal类型的错误则break后强制关闭该连接
if _, ok := err.(*protocol.FatalClientErr); ok {
break
}
continue
}

// 将Exec结果返回给client
if response != nil {
_, err = protocol.SendResponse(client, response)
if err != nil {
break
}
}
}

// for循环退出后则关闭client连接，并从db中删除client信息
conn.Close()
p.ctx.nsqlookupd.logf("CLIENT(%s): closing", client)
if client.peerInfo != nil {
registrations := p.ctx.nsqlookupd.DB.LookupRegistrations(client.peerInfo.id)
for _, r := range registrations {
if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed {
p.ctx.nsqlookupd.logf("DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",
client, r.Category, r.Key, r.SubKey)
}
}
}
return err
}

处理client指令

nsqlookupd/lookup_protocol_v1.go

v1协议支持如下命令

func (p *LookupProtocolV1) Exec(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
switch params[0] {
case "PING":
return p.PING(client, params)
case "IDENTIFY":
return p.IDENTIFY(client, reader, params[1:])
case "REGISTER":
return p.REGISTER(client, reader, params[1:])
case "UNREGISTER":
return p.UNREGISTER(client, reader, params[1:])
}
// 非以上命令则返回E_INVALID错误
return nil, protocol.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("invalid command %s", params[0]))
}

PING命令

// PING：nsqd每隔一段时间都会向nsqlookupd发送心跳
// nsqlookup接收到心跳，原子操作更新nsqd的lastUpdate字段并返回OK
func (p *LookupProtocolV1) PING(client *ClientV1, params []string) ([]byte, error) {
if client.peerInfo != nil {
// we could get a PING before other commands on the same client connection
cur := time.Unix(0, atomic.LoadInt64(&client.peerInfo.lastUpdate))
now := time.Now()
p.ctx.nsqlookupd.logf("CLIENT(%s): pinged (last ping %s)", client.peerInfo.id,
now.Sub(cur))
atomic.StoreInt64(&client.peerInfo.lastUpdate, now.UnixNano())
}
return []byte("OK"), nil
}

IDENTITY命令

// IDENTITY: 当nsqd第一次连接nsqlookupd时，发送IDENTITY验证自己身份
// nsqd将自身的相关信息（ip、hostname等）提交给nsqlookup写入到RegistrationDB
func (p *LookupProtocolV1) IDENTIFY(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
var err error

if client.peerInfo != nil {
return nil, protocol.NewFatalClientErr(err, "E_INVALID", "cannot IDENTIFY again")
}

var bodyLen int32
err = binary.Read(reader, binary.BigEndian, &bodyLen)
if err != nil {
return nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to read body size")
}

body := make([]byte, bodyLen)
_, err = io.ReadFull(reader, body)
if err != nil {
return nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to read body")
}

// body is a json structure with producer information
peerInfo := PeerInfo{id: client.RemoteAddr().String()}
err = json.Unmarshal(body, &peerInfo)
if err != nil {
return nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to decode JSON body")
}

peerInfo.RemoteAddress = client.RemoteAddr().String()

// require all fields
if peerInfo.BroadcastAddress == "" || peerInfo.TCPPort == 0 || peerInfo.HTTPPort == 0 || peerInfo.Version == "" {
return nil, protocol.NewFatalClientErr(nil, "E_BAD_BODY", "IDENTIFY missing fields")
}

atomic.StoreInt64(&peerInfo.lastUpdate, time.Now().UnixNano())

p.ctx.nsqlookupd.logf("CLIENT(%s): IDENTIFY Address:%s TCP:%d HTTP:%d Version:%s",
client, peerInfo.BroadcastAddress, peerInfo.TCPPort, peerInfo.HTTPPort, peerInfo.Version)

client.peerInfo = &peerInfo
if p.ctx.nsqlookupd.DB.AddProducer(Registration{"client", "", ""}, &Producer{peerInfo: client.peerInfo}) {
p.ctx.nsqlookupd.logf("DB: client(%s) REGISTER category:%s key:%s subkey:%s", client, "client", "", "")
}

// build a response
data := make(map[string]interface{})
data["tcp_port"] = p.ctx.nsqlookupd.RealTCPAddr().Port
data["http_port"] = p.ctx.nsqlookupd.RealHTTPAddr().Port
data["version"] = version.Binary
hostname, err := os.Hostname()
if err != nil {
log.Fatalf("ERROR: unable to get hostname %s", err)
}
data["broadcast_address"] = p.ctx.nsqlookupd.opts.BroadcastAddress
data["hostname"] = hostname

response, err := json.Marshal(data)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: marshaling %v", data)
return []byte("OK"), nil
}
return response, nil
}

REGISTER命令

// REGISTER：当nsqd创建一个topic或者channel时，向nsqlookupd发送REGISTER请求
// nsqlookupd更新当前nsqd的topic或者channel信息
func (p *LookupProtocolV1) REGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
if client.peerInfo == nil {
return nil, protocol.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")
}

// 获取topci和channel
topic, channel, err := getTopicChan("REGISTER", params)
if err != nil {
return nil, err
}

// 更新channel信息
if channel != "" {
key := Registration{"channel", topic, channel}
if p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) {
p.ctx.nsqlookupd.logf("DB: client(%s) REGISTER category:%s key:%s subkey:%s",
client, "channel", topic, channel)
}
}
// 更新topic信息
key := Registration{"topic", topic, ""}
if p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) {
p.ctx.nsqlookupd.logf("DB: client(%s) REGISTER category:%s key:%s subkey:%s",
client, "topic", topic, "")
}

return []byte("OK"), nil
}

UNREGISTER命令

// UNREGISTER：当nsqd删除一个topic或者channel时，向nsqlookupd发送UNREGISTER请求
// nsqlookupd更新当前nsqd的topic或者channel信息
func (p *LookupProtocolV1) UNREGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
if client.peerInfo == nil {
return nil, protocol.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")
}

// 获取topci和channel
topic, channel, err := getTopicChan("UNREGISTER", params)
if err != nil {
return nil, err
}

if channel != "" {
// 删除channel信息
key := Registration{"channel", topic, channel}
removed, left := p.ctx.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id)
if removed {
p.ctx.nsqlookupd.logf("DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",
client, "channel", topic, channel)
}
// for ephemeral channels, remove the channel as well if it has no producers
// 如果存在临时channel同样删除
// channel名称前缀带#ephemeral则表示临时channel
if left == 0 && strings.HasSuffix(channel, "#ephemeral") {
p.ctx.nsqlookupd.DB.RemoveRegistration(key)
}
} else {
// 删除topic，并删除topic下的所有channel
// no channel was specified so this is a topic unregistration
// remove all of the channel registrations...
// normally this shouldn't happen which is why we print a warning message
// if anything is actually removed
// 查找出topci下的所有channel并删除
registrations := p.ctx.nsqlookupd.DB.FindRegistrations("channel", topic, "*")
for _, r := range registrations {
if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed {
p.ctx.nsqlookupd.logf("WARNING: client(%s) unexpected UNREGISTER category:%s key:%s subkey:%s",
client, "channel", topic, r.SubKey)
}
}

// 最后删除topic
key := Registration{"topic", topic, ""}
if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id); removed {
p.ctx.nsqlookupd.logf("DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",
client, "topic", topic, "")
}
}

return []byte("OK"), nil
}