本篇讲nsqlookupd中tcp.go、tcp_server.go
tcp_server.go位于util目录下。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | package util import ( "log" "net" "runtime" "strings" ) type TCPHandler interface { Handle(net.Conn) } /** *本方法开始接受客户端连接，并注册处理方法 */ func TCPServer(listener net.Listener, handler TCPHandler) { log.Printf("TCP: listening on %s", listener.Addr().String()) for { clientConn, err := listener.Accept() if err != nil { if nerr, ok := err.(net.Error); ok && nerr.Temporary() { log.Printf("NOTICE: temporary Accept() failure - %s", err.Error()) 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") { log.Printf("ERROR: listener.Accept() - %s", err.Error()) } break } //handler由调用时通过参数传入,作为工具方法，将可能的变化放在外部，以保证本方法可以有尽可能多的适用范围 go handler.Handle(clientConn) } log.Printf("TCP: closing %s", listener.Addr().String()) } |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | package nsqlookupd import ( "io" "log" "net" "github.com/bitly/nsq/util" ) type tcpServer struct { context *Context } // // nsqlookupd接收到tcp数据时，会调用本方法处理。 //实现了/util/tcp_server.go中定义的TCPHandler接口 // func (p *tcpServer) Handle(clientConn net.Conn) { log.Printf("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... //客户端在初始化自己的时候，需要发送4字节的数据用来标识它自己所有使用协议版本。将来升级协议的时候可以避免使用旧协议的客户端不能使用。 buf := make([]byte, 4) _, err := io.ReadFull(clientConn, buf) if err != nil { log.Printf("ERROR: failed to read protocol version - %s", err.Error()) return } //获取协议内容 protocolMagic := string(buf) log.Printf("CLIENT(%s): desired protocol magic '%s'", clientConn.RemoteAddr(), protocolMagic) //utilProtocol.go中定义了一个Protocol的接口 var prot util.Protocol switch protocolMagic { case " V1": //当前只支持" V1"协议（前两有两个空格，所以总共是4字节），协议在nsqlookupdlookup_protocol_v1.go文件中定义，这里创建了LookupProtocolV1的实例 //LookupProtocolV1实现了Protocol接口 prot = &LookupProtocolV1{context: p.context} default: //如果不是" V1"协议，则协议出错，断开链接，返回。 util.SendResponse(clientConn, []byte("E_BAD_PROTOCOL")) clientConn.Close() log.Printf("ERROR: client(%s) bad protocol magic '%s'", clientConn.RemoteAddr(), protocolMagic) return } //如果是" V1"f协议，这里就进入了LookupProtocolV1的IOLoop方法。此方法里有for死循环运行，直到出现error时，才会执行下面的代码。 err = prot.IOLoop(clientConn) if err != nil { log.Printf("ERROR: client(%s) - %s", clientConn.RemoteAddr(), err.Error()) return } } |
从以上代码可看出，tcp.go和tcp_server.go主要功能就是监听并接受TCP请求，收到请求后，将数据转给lookup_protocol_v1.go来处理。这个下篇再讲。