go-ethereum中rpc源码学习
背景
RPC全称Remote Procedure Call，即远程过程调用是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程。
运行时一次客户机对服务器的RPC调用，基本过程如下图：
go语言中的rpc
go语言的标准包中以及有了对RPC的支持，分别在三个层面上即TCP、HTTP和JSONRPC上提供了支持。除了JSONRPC其余两种都只支持以go语言开发的客户端与服务器。
首先，在Go中一个正确的RPC函数应该满足下面要求
- 函数必须是可导出的（首字母大写）
- 必须有两个可导出类型的参数
- 第一个是接受的参数，第二个是返回参数
- 函数必须有一个error类型的返回值
例子如下：
func (t *T) MethodName(argType T1, replyType *T2) errorHTTP RPC
服务端代码
type HelloService struct {
}
func (p *HelloService)Hello(requset string,reply *string)error{
*reply = "hello " + requset
return nil
}
func main() {
rpc.Register(new(HelloService))
rpc.HandleHTTP()
err:=http.ListenAndServe(":1234",nil)
if err!=nil {
fmt.Println(err.Error())
}
}关键有两点，一个是定义供客户端调用的方法，另一个是注册服务，并使RPC托管http服务，最后正常启动http服务
客户端代码
func main() {
client,err:=rpc.DialHTTP("tcp",":1234")
if err!=nil {
log.Fatal("dial",err.Error())
}
var reply string
err = client.Call("HelloService.Hello","world",&reply)
fmt.Println(reply)
}TPC RPC
服务端代码
func main() {
rpc.Register(new(HelloService))
lis,err:=net.Listen("tcp",":1234")
if err!=nil {
log.Fatal("listen:",err.Error())
}
conn,err:=lis.Accept()
if err!=nil {
log.Fatal("accept:",err.Error())
}
rpc.ServeConn(conn)
}供远程调用的方法和前面的例子一样，不在重复，tcprpc和httprpc的不同点就是在于首先正常启动tcp监听，然后接收到的连接交给rpc即可
客户端代码
func main() {
client,err:=rpc.Dial("tcp",":1234")
if err!=nil {
log.Fatal("dial",err.Error())
}
var reply string
err = client.Call("HelloService.Hello","world",&reply)
fmt.Println(reply)
}和httprpc的唯一区别就是初始化客户端对象的方法不同
jsonRPC
服务端代码
func main() {
rpc.Register(new(HelloService))
lis,err:=net.Listen("tcp",":1234")
if err!=nil {
log.Fatal("listen:",err.Error())
}
conn,err:=lis.Accept()
if err!=nil {
log.Fatal("accept:",err.Error())
}
jsonrpc.ServeConn(conn)
}相比较tcprpc只是在处理tcp连接时使用rpcjson提供的方法。
客户端代码
func main() {
client,err:=jsonrpc.Dial("tcp",":1234")
if err!=nil {
log.Fatal("dial",err.Error())
}
var reply string
err = client.Call("HelloService.Hello","world",&reply)
fmt.Println(reply)
}
唯一区别也只是在创建client实例时的不同，使用jsonrpc提供的方法
由于tcprpc和httprpc都采用gob编码，所以不能跨语言使用，而jsonrpc采用的json格式编码，可以很方便的跨语言，这里提供一个go语言和java语言交互的例子，其中服务端由go语言编写，如上面所示，java编写的客户端如下：
public class ClientRequest {
public String method;
public String[] params;
public int id;
}
public class ServerResponse {
public int id;
public String result;
public String error;
}
public static void main(String[] args){
try(Socket client = new Socket("127.0.0.1",1234)){
OutputStream out = client.getOutputStream();
Gson gson = new Gson();
ClientRequest request = new ClientRequest();
request.id = 1;
request.method = "HelloService.Hello";
request.params = new String[]{"hello"};
out.write(gson.toJson(request).getBytes());
client.shutdownOutput();
InputStream in = client.getInputStream();
int len;
byte[] buffer = new byte[1024];
StringBuilder sb = new StringBuilder();
while ((len = in.read(buffer))!=-1)
sb.append(new String(buffer,0,len));
ServerResponse response = gson.fromJson(sb.toString(),ServerResponse.class);
client.shutdownInput();
System.out.println(response.result);
}catch (Exception e){
e.printStackTrace();
}
}还是基于TCP通信的，由于是借助json传输的，所以需要有json序列化和反序列化操作，且json格式要符合go语言的规定。go语言中规定请求的格式如下：
//客户端请求
type clientRequest struct{
Method string `json:"method"`
Params [1]interface{} `json:"params"`
Id uint64 `json:"id"`
}
//服务端请求
type serverRequest struct{
Method string `json:"method"`
Params *json.RawMessage `json:"params"`
Id *json.RawMessage `json:"id"`
}规定响应的格式如下
//客户端响应
type clientResponse struct{
Id uint64 `json:"id"`
Result *json.RawMessage `json:"result"`
Error interface{} `json:"error"`
}
//服务端响应
type serverResponse struct{
Id uint64 `json:"id"`
Result interface{} `json:"result"`
Error interface{} `json:"error"`
}在跨语言交互时，要根据上面的定义，编写对应的json文件
go-ethereum中的rpc
主要集中在rpc目录下
server.go
服务创建与注册
首先从服务端开始看，server的结构体以及创建如下
// go-ethereum\rpc\server.go
type Server struct {
services serviceRegistry
idgen func() ID
run int32
codecs mapset.Set
}
type serviceRegistry struct {
mu sync.Mutex
services map[string]service
}
func NewServer() *Server {
server := &Server{idgen: randomIDGenerator(), codecs: mapset.NewSet(), run: 1}
rpcService := &RPCService{server}
//MetadataApi = "rpc"
server.RegisterName(MetadataApi, rpcService)
return server
}结构体中services的类型是serviceRegistry实际上就是记录了所有注册的服务。run用来控制server的运行。codecs是一个set类型，存储所有编解码器。
在创建server时，randomIDGenerator()是一个id的随机生成器。随后调用RegisterName把自己的实例注册进去（RPCService包装了server类）
func (s *Server) RegisterName(name string, receiver interface{}) error {
return s.services.registerName(name, receiver)
}
// go-ethereum\rpc\service.go
func (r *serviceRegistry) registerName(name string, rcvr interface{}) error {
rcvrVal := reflect.ValueOf(rcvr)
if name == "" {
return fmt.Errorf("no service name for type %s", rcvrVal.Type().String())
}
callbacks := suitableCallbacks(rcvrVal)
if len(callbacks) == 0 {
return fmt.Errorf("service %T doesn't have any suitable methods/subscriptions to expose", rcvr)
}
r.mu.Lock()
defer r.mu.Unlock()
if r.services == nil {
r.services = make(map[string]service)
}
svc, ok := r.services[name]
if !ok {
svc = service{
name: name,
callbacks: make(map[string]*callback),
subscriptions: make(map[string]*callback),
}
r.services[name] = svc
}
for name, cb := range callbacks {
if cb.isSubscribe {
svc.subscriptions[name] = cb
} else {
svc.callbacks[name] = cb
}
}
return nil
}suitableCallbacks方法如下
// go-ethereum\rpc\service.go
func suitableCallbacks(receiver reflect.Value) map[string]*callback {
typ := receiver.Type()
callbacks := make(map[string]*callback)
for m := 0; m < typ.NumMethod(); m++ {
method := typ.Method(m)
if method.PkgPath != "" {
continue // method not exported
}
cb := newCallback(receiver, method.Func)
if cb == nil {
continue // function invalid
}
name := formatName(method.Name)
callbacks[name] = cb
}
return callbacks
}
func newCallback(receiver, fn reflect.Value) *callback {
fntype := fn.Type()
c := &callback{fn: fn, rcvr: receiver, errPos: -1, isSubscribe: isPubSub(fntype)}
c.makeArgTypes()
if !allExportedOrBuiltin(c.argTypes) {
return nil
}
outs := make([]reflect.Type, fntype.NumOut())
for i := 0; i < fntype.NumOut(); i++ {
outs[i] = fntype.Out(i)
}
if len(outs) > 2 || !allExportedOrBuiltin(outs) {
return nil
}
switch {
case len(outs) == 1 && isErrorType(outs[0]):
c.errPos = 0
case len(outs) == 2:
if isErrorType(outs[0]) || !isErrorType(outs[1]) {
return nil
}
c.errPos = 1
}
return c
}
type callback struct {
fn reflect.Value
rcvr reflect.Value
argTypes []reflect.Type
hasCtx bool
errPos int
isSubscribe bool
}在suitableCallbacks内先遍历了server的所有方法，对于可导出的，利用newCallback方法创建一个callback对象。一个callback对象中的几个成员意义如下：
- fn表示对于的方法
- rcvr表示方法所在的类
- argtypes存储所有输入参数类型（context除外）
- hasCtx表示参数中是否有Context类型，.
- errPos表示error类型的返回值是第几个
- isSubscribe表示该方法是否可订阅（满足三个条件：第二个输入参数是Context类型，第一个输出参数是Subscription类型，第二个输出参数是error类型）
最后suitableCallbacks返回一个map存储符合条件的callback。接下来初始化serviceRegistry的services，也是一个map。然后创建一个service，存入services。最后遍历刚才callbacks，将其中的可订阅的方法单独拿出来。
总结一下，所谓的serviceRegistry注册方法就是给一个类，然后遍历所有方法，根据方法的参数与返回值归类，最后以service的形式放到serviceRegistry的services中.
服务启动及rpc请求处理分析
节点会根据需求启动多种rpc服务，这里我们先以iprpc为例分析，ipcrpc在启动node时使用startIPC启动
// go-ethereum\rpc\ipc.go
func (s *Server) ServeListener(l net.Listener) error {
for {
conn, err := l.Accept()
if netutil.IsTemporaryError(err) {
log.Warn("RPC accept error", "err", err)
continue
} else if err != nil {
return err
}
log.Trace("Accepted RPC connection", "conn", conn.RemoteAddr())
go s.ServeCodec(NewJSONCodec(conn), OptionMethodInvocation|OptionSubscriptions)
}
}整个是一个无限循环，每次循环接受一个连接，然后启动一个goroutine去调用ServeCodec处理，第一个参数是ServerCodec类型：
func NewJSONCodec(conn Conn) ServerCodec {
enc := json.NewEncoder(conn)
dec := json.NewDecoder(conn)
dec.UseNumber()
return NewCodec(conn, enc.Encode, dec.Decode)
}
func NewCodec(conn Conn, encode, decode func(v interface{}) error) ServerCodec {
codec := &jsonCodec{
closed: make(chan interface{}),
encode: encode,
decode: decode,
conn: conn,
}
if ra, ok := conn.(ConnRemoteAddr); ok {
codec.remoteAddr = ra.RemoteAddr()
}
return codec
}具体来说是一个jsonCodec对象，用于读写jsonrpc请求与响应的。ServeCodec的第二个参数是一个选项，但是新版本的go-ethereum不在支持该选项，所以弃用。总的来说codec类型对象就代表的是一个个连接请求
具体来说ServeCodec方法是读取一个请求，然后使用合适的callback去处理，最后返回，来看具体代码
// go-ethereum\rpc\server.go
func (s *Server) ServeCodec(codec ServerCodec, options CodecOption) {
defer codec.Close()
if atomic.LoadInt32(&s.run) == 0 {
return
}
s.codecs.Add(codec)
defer s.codecs.Remove(codec)
c := initClient(codec, s.idgen, &s.services)
<-codec.Closed()
c.Close()
}前面说过run成员是控制运行的，如果等于0就不执行任何操作。然后将codec添加到set集合中，然后初始化一个Client去处理，最后等待codec关闭后，关闭Client。一个client就代表一个连接。初始化客户端代码如下
// go-ethereum\rpc\client.go
func initClient(conn ServerCodec, idgen func() ID, services *serviceRegistry) *Client {
_, isHTTP := conn.(*httpConn)
c := &Client{
idgen: idgen,
isHTTP: isHTTP,
services: services,
writeConn: conn,
close: make(chan struct{}),
closing: make(chan struct{}),
didClose: make(chan struct{}),
reconnected: make(chan ServerCodec),
readOp: make(chan readOp),
readErr: make(chan error),
reqInit: make(chan *requestOp),
reqSent: make(chan error, 1),
reqTimeout: make(chan *requestOp),
}
if !isHTTP {
go c.dispatch(conn)
}
return c
}就是简单的初始化，不过通过判断是否是http类型连接来决定是否分发。
// go-ethereum\rpc\client.go
func (c *Client) dispatch(codec ServerCodec) {
var (
lastOp *requestOp
reqInitLock = c.reqInit
conn = c.newClientConn(codec)
reading = true
)
defer func() {
close(c.closing)
if reading {
conn.close(ErrClientQuit, nil)
c.drainRead()
}
close(c.didClose)
}()
go c.read(codec)
for {
select {
case <-c.close:
return
case op := <-c.readOp:
if op.batch {
conn.handler.handleBatch(op.msgs)
} else {
conn.handler.handleMsg(op.msgs[0])
}
case err := <-c.readErr:
conn.handler.log.Debug("RPC connection read error", "err", err)
conn.close(err, lastOp)
reading = false
case newcodec := <-c.reconnected:
log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.RemoteAddr())
if reading {
conn.close(errClientReconnected, lastOp)
c.drainRead()
}
go c.read(newcodec)
reading = true
conn = c.newClientConn(newcodec)
conn.handler.addRequestOp(lastOp)
case op := <-reqInitLock:
reqInitLock = nil
lastOp = op
conn.handler.addRequestOp(op)
case err := <-c.reqSent:
if err != nil {
conn.handler.removeRequestOp(lastOp)
}
reqInitLock = c.reqInit
lastOp = nil
case op := <-c.reqTimeout:
conn.handler.removeRequestOp(op)
}
}
}大体上来看dispatch方法是一个无限循环，每次循环都借助channel机制实现对不同的情况做不同处理，在循环阻塞时，启动了一个goroutine，去读取rpc请求
func (c *Client) read(codec ServerCodec) {
for {
msgs, batch, err := codec.Read()
if _, ok := err.(*json.SyntaxError); ok {
codec.Write(context.Background(), errorMessage(&parseError{err.Error()}))
}
if err != nil {
c.readErr <- err
return
}
c.readOp <- readOp{msgs, batch}
}
}这里就很清楚，直接调用ServerCodec的read方法，我们前面说过ServerCodec是一个个连接请求的包装，而这里的ServerCodec实际上是个jsonCodec类型，我们看看它的read方法
func (c *jsonCodec) Read() (msg []*jsonrpcMessage, batch bool, err error) {
var rawmsg json.RawMessage
if err := c.decode(&rawmsg); err != nil {
return nil, false, err
}
msg, batch = parseMessage(rawmsg)
return msg, batch, nil
}
func parseMessage(raw json.RawMessage) ([]*jsonrpcMessage, bool) {
if !isBatch(raw) {
msgs := []*jsonrpcMessage{{}}
json.Unmarshal(raw, &msgs[0])
return msgs, false
}
dec := json.NewDecoder(bytes.NewReader(raw))
dec.Token() // skip '['
var msgs []*jsonrpcMessage
for dec.More() {
msgs = append(msgs, new(jsonrpcMessage))
dec.Decode(&msgs[len(msgs)-1])
}
return msgs, true
}这里的decode就是json.NewDecoder(conn)创建的，它将请求中的数据流转为一个json原始格式信息，为的是接下来反序列化。isBatch检查数据中第一个非空字符是否是“[”，如果是的话表示是一个json数据格式。总之最后解析为jsonrpcMessage对象，结构如下
type jsonrpcMessage struct {
Version string `json:"jsonrpc,omitempty"`
ID json.RawMessage `json:"id,omitempty"`
Method string `json:"method,omitempty"`
Params json.RawMessage `json:"params,omitempty"`
Error *jsonError `json:"error,omitempty"`
Result json.RawMessage `json:"result,omitempty"`
}我们继续回到read方法中，当正确解析请求的内容时，返回值打包为readOp类型，然后写入Client的readOp这个chan字段中，这时触发dispatch中的select，执行下面逻辑：
if op.batch {
conn.handler.handleBatch(op.msgs)
} else {
conn.handler.handleMsg(op.msgs[0])
}op.batch代表是否有多个jsonrpcMessage，以此执行不同逻辑。这里的conn是一个clientConn类型，包装了ServerCodec以及一个handler。我们下面看一下只有一个message的处理逻辑
// go-ethereum\rpc\handler.go
func (h *handler) handleMsg(msg *jsonrpcMessage) {
if ok := h.handleImmediate(msg); ok {
return
}
h.startCallProc(....)
}第一行中handleImmediate处理的是不需要回复的请求，如一个通知或一个响应。通过方法名，ID以及参数等综合判断。对于正常请求调用startCallProc开始处理。
func (h *handler) startCallProc(fn func(*callProc)) {
h.callWG.Add(1)
go func() {
ctx, cancel := context.WithCancel(h.rootCtx)
defer h.callWG.Done()
defer cancel()
fn(&callProc{ctx: ctx})
}()
}h.callWG是一个sync.WaitGroup类型，为的是线程同步，这里加一表示有一个goroutine在运行，后面启动一个goroutine，这里面的实际逻辑是先前传递进来的func，如下
func(cp *callProc) {
answer := h.handleCallMsg(cp, msg)
h.addSubscriptions(cp.notifiers)
if answer != nil {
h.conn.Write(cp.ctx, answer)
}
for _, n := range cp.notifiers {
n.activate()
}
}这里调用handleCallMsg去处理message：
func (h *handler) handleCallMsg(ctx *callProc, msg *jsonrpcMessage) *jsonrpcMessage {
start := time.Now()
switch {
case msg.isNotification():
h.handleCall(ctx, msg)
h.log.Debug("Served "+msg.Method, "t", time.Since(start))
return nil
case msg.isCall():
resp := h.handleCall(ctx, msg)
if resp.Error != nil {
h.log.Info("Served "+msg.Method, "reqid", idForLog{msg.ID}, "t", time.Since(start), "err", resp.Error.Message)
} else {
h.log.Debug("Served "+msg.Method, "reqid", idForLog{msg.ID}, "t", time.Since(start))
}
return resp
case msg.hasValidID():
return msg.errorResponse(&invalidRequestError{"invalid request"})
default:
return errorMessage(&invalidRequestError{"invalid request"})
}
}不管是通知类型还是调用类型，都调用的是handleCall方法：
func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage {
if msg.isSubscribe() {
return h.handleSubscribe(cp, msg)
}
var callb *callback
if msg.isUnsubscribe() {
callb = h.unsubscribeCb
} else {
callb = h.reg.callback(msg.Method)
}
if callb == nil {
return msg.errorResponse(&methodNotFoundError{method: msg.Method})
}
args, err := parsePositionalArguments(msg.Params, callb.argTypes)
if err != nil {
return msg.errorResponse(&invalidParamsError{err.Error()})
}
return h.runMethod(cp.ctx, msg, callb, args)
}分了三种情况：订阅请求，取消订阅请求以及一般请求。我们先看一般请求：h.reg是serviceRegistry，handler的serviceRegistry是前面dispatch方法创建clientConn时创建handle时传入的，来自于client的services字段，而client的services字段是在server的ServeCodec方法中调用initClient传入的，实际上就是Server的services字段。我们上节了解到在server会有一个registerName动作，会解析rpcserver的所有可导出方法，并用callback包装，而在handleCall中，就调用了serviceRegistry方法
func (r *serviceRegistry) callback(method string) *callback {
elem := strings.SplitN(method, serviceMethodSeparator, 2)
if len(elem) != 2 {
return nil
}
r.mu.Lock()
defer r.mu.Unlock()
return r.services[elem[0]].callbacks[elem[1]]
}首先解析方法名，然后从services中，这里保存了一系列注册的服务，在从各个服务的callbacks集合中寻找对应的callback。继续回到handleCall，若找的callback不为空，则尝试解析参数：
func parsePositionalArguments(rawArgs json.RawMessage, types []reflect.Type) ([]reflect.Value, error) {
dec := json.NewDecoder(bytes.NewReader(rawArgs))
var args []reflect.Value
tok, err := dec.Token()
switch {
case err == io.EOF || tok == nil && err == nil:
case err != nil:
return nil, err
case tok == json.Delim('['):
if args, err = parseArgumentArray(dec, types); err != nil {
return nil, err
}
default:
return nil, errors.New("non-array args")
}
for i := len(args); i < len(types); i++ {
if types[i].Kind() != reflect.Ptr {
return nil, fmt.Errorf("missing value for required argument %d", i)
}
args = append(args, reflect.Zero(types[i]))
}
return args, nil
}
func parseArgumentArray(dec *json.Decoder, types []reflect.Type) ([]reflect.Value, error) {
args := make([]reflect.Value, 0, len(types))
for i := 0; dec.More(); i++ {
if i >= len(types) {
return args, fmt.Errorf("too many arguments, want at most %d", len(types))
}
argval := reflect.New(types[i])
if err := dec.Decode(argval.Interface()); err != nil {
return args, fmt.Errorf("invalid argument %d: %v", i, err)
}
if argval.IsNil() && types[i].Kind() != reflect.Ptr {
return args, fmt.Errorf("missing value for required argument %d", i)
}
args = append(args, argval.Elem())
}
_, err := dec.Token()
return args, err
}主要逻辑是在parseArgumentArray中，根据方法的参数列表类型一个一个进行解析，对于没有的参数设为该类型的默认空值，最后返回一组参数。在handleCall的最后去执行方法：
func (h *handler) runMethod(ctx context.Context, msg *jsonrpcMessage, callb *callback, args []reflect.Value) *jsonrpcMessage {
result, err := callb.call(ctx, msg.Method, args)
if err != nil {
return msg.errorResponse(err)
}
return msg.response(result)
}
func (c *callback) call(ctx context.Context, method string, args []reflect.Value) (res interface{}, errRes error) {
fullargs := make([]reflect.Value, 0, 2+len(args))
if c.rcvr.IsValid() {
fullargs = append(fullargs, c.rcvr)
}
if c.hasCtx {
fullargs = append(fullargs, reflect.ValueOf(ctx))
}
fullargs = append(fullargs, args...)
defer func() {
if err := recover(); err != nil {
const size = 64 << 10
buf := make([]byte, size)
buf = buf[:runtime.Stack(buf, false)]
log.Error("RPC method " + method + " crashed: " + fmt.Sprintf("%v\n%s", err, buf))
errRes = errors.New("method handler crashed")
}
}()
results := c.fn.Call(fullargs)
if len(results) == 0 {
return nil, nil
}
if c.errPos >= 0 && !results[c.errPos].IsNil() {
err := results[c.errPos].Interface().(error)
return reflect.Value{}, err
}
return results[0].Interface(), nil
}基本上就是补全参数，然后调用callback中保存的func去执行，然后进行错误处理，最后返回结果也就是响应。回到runMethod中，调用response去包装结果
// go-ethereum\rpc\json.go
func (msg *jsonrpcMessage) response(result interface{}) *jsonrpcMessage {
enc, err := json.Marshal(result)
if err != nil {
// TODO: wrap with 'internal server error'
return msg.errorResponse(err)
}
return &jsonrpcMessage{Version: vsn, ID: msg.ID, Result: enc}
}这一步是将结果序列化，最后包装成为一个jsonrpcMessage。到这里handleCall流程结束，回到handleCallMsg，直接返回响应，再往回调，来到startCallProc方法，如果响应不为空，则调用write写入响应，一次rpc普通调用完成！负责写的还是ServerCodec类型对象，具体就是jsonCodec，最后写入响应流中。
总结一下，大致流程就是当接收到一个网络请求时，就启动一个goroutine，同时创建一个ServerCodec去包装过来的连接，在这个goroutine中会初始化一个client对象，代表一个连接，在初始化之后会进行连接的分发，分发就是有一个无限循环，同时再启动一个goroutine去解析请求信息，根据信息去开始注册的服务中查找合适的callback然后传入请求体重的参数执行响应逻辑，最后在利用ServerCodec去写会响应。
服务的关闭
func (s *Server) Stop() {
if atomic.CompareAndSwapInt32(&s.run, 1, 0) {
log.Debug("RPC server shutting down")
s.codecs.Each(func(c interface{}) bool {
c.(ServerCodec).Close()
return true
})
}
}首先将标志位run置0，这样后续新的连接就被放弃（参见ServeCodec方法），然后遍历set，对每个ServerCodec执行close方法。
func (c *jsonCodec) Close() {
c.closer.Do(func() {
close(c.closed)
c.conn.Close()
})
}主要就是关闭channel和连接。channel关闭后影响ServeCodec方法，原来阻塞地方开始执行，然后关闭client
func (c *Client) Close() {
if c.isHTTP {
return
}
select {
case c.close <- struct{}{}:
<-c.didClose
case <-c.didClose:
}
}这里首先close这个channel获得值，在dispatch的那个无限循环中的到触发，跳出循环，执行defer逻辑，
defer func() {
close(c.closing)
if reading {
conn.close(ErrClientQuit, nil)
c.drainRead()
}
close(c.didClose)
}()这里进行最后善后处理，关闭完didClose后，上面Close()方法中最后阻塞得到解除，方法正常执行完毕。
题图来自unsplash：https://unsplash.com/photos/zNNPSqKRR2c