nsq源码阅读 nsqlookupd源码三 tcp.go tcp_server.go
2017-05-02 16:26
801 查看
nsqlookupd.go中对TCP的处理代码为
用户监听处理nsqd发送的信息
TCPServer在internal/protocol/tcp_server.go中定义:
package protocol
import (
"fmt"
"net"
"runtime"
"strings"
"github.com/nsqio/nsq/internal/app"
)
type TCPHandler interface {
Handle(net.Conn)
}
//启动TCP服务,接受客户端连接,并注册处理方法
func TCPServer(listener net.Listener, handler TCPHandler, l app.Logger) {
l.Output(2, fmt.Sprintf("TCP: listening on %s", listener.Addr()))
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()
continue
}
// there's 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
}
//golang接口的应用
go handler.Handle(clientConn)
}
l.Output(2, fmt.Sprintf("TCP: closing %s", listener.Addr()))
}
protocol.tcpServer()的第二个参数,是一个tcpServer实例,在文件nsqlookupd/tcp.go中定义,改实例实现了internal/protocol/tcp_server.go中的TCPHandler接口,
tcp.go:
package nsqlookupd
import (
"io"
"net"
"github.com/nsqio/nsq/internal/protocol"
)
type tcpServer struct {
ctx *Context
}
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...
//客户端初始化自己时,需要发送一个4字节序列数据指明它想使用的协议版本
buf := make([]byte, 4)
//读取协议内容
//函数原型
// func ReadFull(r Reader, buf []byte) (n int, err error)
//把对象 r 中的数据读出来,然后存入一个缓冲区 buf 中
_, err := io.ReadFull(clientConn, buf)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: failed to read protocol version - %s", err)
return
}
//将读取到的字节流转换成string
protocolMagic := string(buf)
p.ctx.nsqlookupd.logf("CLIENT(%s): desired protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)
//internal/protocol/protocol.go中定义了Protocol接口
//接口中方法原型 IOLoop(conn net.Conn) error
var prot protocol.Protocol
switch protocolMagic {
case " V1":
//当前只支持" V1"协议(前面两个空格,所以共4个字节)
//协议在nsqlookupd/lookup_protocol_v1.go中定义
//创建LookupProtocolV1实例
//LookupProtocolV1实现了Protocol接口
prot = &LookupProtocolV1{ctx: p.ctx}
default:
//如果不是" V1"协议,则断开链接,打印错误信息,返回
protocol.SendResponse(clientConn, []byte("E_BAD_PROTOCOL"))
clientConn.Close()
p.ctx.nsqlookupd.logf("ERROR: client(%s) bad protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)
return
}
//如果是" V1"协议,则调用LookupProtocolV1的IOLoop方法
err = prot.IOLoop(clientConn)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: client(%s) - %s", clientConn.RemoteAddr(), err)
return
}
}
protocol.TCPServer(tcpListener, tcpServer, l.opts.Logger)
用户监听处理nsqd发送的信息
TCPServer在internal/protocol/tcp_server.go中定义:
package protocol
import (
"fmt"
"net"
"runtime"
"strings"
"github.com/nsqio/nsq/internal/app"
)
type TCPHandler interface {
Handle(net.Conn)
}
//启动TCP服务,接受客户端连接,并注册处理方法
func TCPServer(listener net.Listener, handler TCPHandler, l app.Logger) {
l.Output(2, fmt.Sprintf("TCP: listening on %s", listener.Addr()))
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()
continue
}
// there's 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
}
//golang接口的应用
go handler.Handle(clientConn)
}
l.Output(2, fmt.Sprintf("TCP: closing %s", listener.Addr()))
}
protocol.tcpServer()的第二个参数,是一个tcpServer实例,在文件nsqlookupd/tcp.go中定义,改实例实现了internal/protocol/tcp_server.go中的TCPHandler接口,
tcp.go:
package nsqlookupd
import (
"io"
"net"
"github.com/nsqio/nsq/internal/protocol"
)
type tcpServer struct {
ctx *Context
}
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...
//客户端初始化自己时,需要发送一个4字节序列数据指明它想使用的协议版本
buf := make([]byte, 4)
//读取协议内容
//函数原型
// func ReadFull(r Reader, buf []byte) (n int, err error)
//把对象 r 中的数据读出来,然后存入一个缓冲区 buf 中
_, err := io.ReadFull(clientConn, buf)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: failed to read protocol version - %s", err)
return
}
//将读取到的字节流转换成string
protocolMagic := string(buf)
p.ctx.nsqlookupd.logf("CLIENT(%s): desired protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)
//internal/protocol/protocol.go中定义了Protocol接口
//接口中方法原型 IOLoop(conn net.Conn) error
var prot protocol.Protocol
switch protocolMagic {
case " V1":
//当前只支持" V1"协议(前面两个空格,所以共4个字节)
//协议在nsqlookupd/lookup_protocol_v1.go中定义
//创建LookupProtocolV1实例
//LookupProtocolV1实现了Protocol接口
prot = &LookupProtocolV1{ctx: p.ctx}
default:
//如果不是" V1"协议,则断开链接,打印错误信息,返回
protocol.SendResponse(clientConn, []byte("E_BAD_PROTOCOL"))
clientConn.Close()
p.ctx.nsqlookupd.logf("ERROR: client(%s) bad protocol magic '%s'",
clientConn.RemoteAddr(), protocolMagic)
return
}
//如果是" V1"协议,则调用LookupProtocolV1的IOLoop方法
err = prot.IOLoop(clientConn)
if err != nil {
p.ctx.nsqlookupd.logf("ERROR: client(%s) - %s", clientConn.RemoteAddr(), err)
return
}
}
相关文章推荐
- nsq源码阅读 nsqlookupd源码五 http.go http_server.go
- nsq源码阅读 nsqlookupd源码一 nsqlookupd.go
- nsq源码阅读 nsqlookupd源码二 registration_db.go
- nsq源码阅读 nsqlookupd源码四 lookup_protocol_v1.go
- nsq源码阅读 nsqd源码三 tcp.go
- nsq源码阅读 nsqd源码四 nsqd/lookup.go 与nsqlookupd服务的交互
- go语言 nsq源码解读三 nsqlookupd源码nsqlookupd.go
- Go语言Http Server源码阅读
- nsq源码阅读 nsqd源码一 apps/nsqd/nsqd.go
- NSQ系列之nsqlookupd代码分析二(初识nsqlookupd tcpServer)
- ping源码代阅读(三)面向连接tcp/ip程序
- 分布式文件系统KFS源码阅读与分析(三):RPC实现机制(MetaServer端)
- Tomcat源码阅读(四)Server
- go-tour源码阅读
- MongoDB源码阅读之Shard源码分析--CongfigServer启动
- (转)go语言nsq源码解读二 nsqlookupd、nsqd与nsqadmin
- 强大的TcpServer压力测试工具源码(附突破连接限制的方法和工具)
- webpy源码阅读(2)——CherryPyWSGIServer
- Go Revel - server.go 源码分析
- /src/pkg/math/abs.go源码阅读兼谈golang与汇编