上一節(jié)我們試著寫了一個RPC的請求實例线定，通過分析源碼知道了RPC服務(wù)的創(chuàng)建流程娜谊，以及Http RPC server創(chuàng)建過程，Http RPC Client的請求流程斤讥。
這一節(jié)纱皆，先分析一下Http RPC server如何處理client的請求。然后再分析一下IPC RPC的處理流程芭商。
一派草，Http RPC server處理Client的請求。
回到上一節(jié)startHTTP() 里面HTTPServer初始化的方法

func NewHTTPServer(cors []string, vhosts []string, srv *Server) *http.Server {
    // Wrap the CORS-handler within a host-handler
    handler := newCorsHandler(srv, cors)
    handler = newVHostHandler(vhosts, handler)
    return &http.Server{Handler: handler}
}

// ServeHTTP serves JSON-RPC requests over HTTP.
func (srv *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // Permit dumb empty requests for remote health-checks (AWS)
    if r.Method == http.MethodGet && r.ContentLength == 0 && r.URL.RawQuery == "" {
        return
    }
    if code, err := validateRequest(r); err != nil {
        http.Error(w, err.Error(), code)
        return
    }
    // All checks passed, create a codec that reads direct from the request body
    // untilEOF and writes the response to w and order the server to process a
    // single request.
    codec := NewJSONCodec(&httpReadWriteNopCloser{r.Body, w})
    defer codec.Close()

    w.Header().Set("content-type", contentType)
    srv.ServeSingleRequest(codec, OptionMethodInvocation)
}

實現(xiàn)了http.server的 ServeHTTP(w http.ResponseWriter, r *http.Request)方法铛楣。
先過濾掉非法的請求近迁，對接收到的請求body體，進行JSONCodec封裝簸州。
然后交由 srv.ServeSingleRequest(codec, OptionMethodInvocation)處理鉴竭。
接著調(diào)用 s.serveRequest(codec, true, options)
singleShot參數(shù)是控制請求時同步還是異步。如果singleShot為true岸浑，那么請求的處理是同步的拓瞪，需要等待處理結(jié)果之后才能退出。 singleShot為false助琐，把處理請求的方法交由goroutine異步處理祭埂。
Http RPC的處理是使用同步方式。

func (s *Server) serveRequest(codec ServerCodec, singleShot bool, options CodecOption) error {
    var pend sync.WaitGroup

    defer func() {
        if err := recover(); err != nil {
            const size = 64 << 10
            buf := make([]byte, size)
            buf = buf[:runtime.Stack(buf, false)]
            log.Error(string(buf))
        }
        s.codecsMu.Lock()
        s.codecs.Remove(codec)
        s.codecsMu.Unlock()
    }()

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // if the codec supports notification include a notifier that callbacks can use
    // to send notification to clients. It is thight to the codec/connection. If the
    // connection is closed the notifier will stop and cancels all active subscriptions.
    if options&OptionSubscriptions == OptionSubscriptions {
        ctx = context.WithValue(ctx, notifierKey{}, newNotifier(codec))
    }
    s.codecsMu.Lock()
    if atomic.LoadInt32(&s.run) != 1 { // server stopped
        s.codecsMu.Unlock()
        return &shutdownError{}
    }
    s.codecs.Add(codec)
    s.codecsMu.Unlock()

    // test if the server is ordered to stop
    for atomic.LoadInt32(&s.run) == 1 {
        reqs, batch, err := s.readRequest(codec)
        if err != nil {
            // If a parsing error occurred, send an error
            if err.Error() != "EOF" {
                log.Debug(fmt.Sprintf("read error %v\n", err))
                codec.Write(codec.CreateErrorResponse(nil, err))
            }
            // Error or end of stream, wait for requests and tear down
            pend.Wait()
            return nil
        }

        // check if server is ordered to shutdown and return an error
        // telling the client that his request failed.
        if atomic.LoadInt32(&s.run) != 1 {
            err = &shutdownError{}
            if batch {
                resps := make([]interface{}, len(reqs))
                for i, r := range reqs {
                    resps[i] = codec.CreateErrorResponse(&r.id, err)
                }
                codec.Write(resps)
            } else {
                codec.Write(codec.CreateErrorResponse(&reqs[0].id, err))
            }
            return nil
        }
        // If a single shot request is executing, run and return immediately
        if singleShot {
            if batch {
                s.execBatch(ctx, codec, reqs)
            } else {
                s.exec(ctx, codec, reqs[0])
            }
            return nil
        }
        // For multi-shot connections, start a goroutine to serve and loop back
        pend.Add(1)

        go func(reqs []*serverRequest, batch bool) {
            defer pend.Done()
            if batch {
                s.execBatch(ctx, codec, reqs)
            } else {
                s.exec(ctx, codec, reqs[0])
            }
        }(reqs, batch)
    }
    return nil
}

當(dāng)server啟動 s.run的值就為1，直到server stop蛆橡。
將codec add進s.codecs舌界，codecs是一個set。
處理完請求數(shù)據(jù)泰演，返回時需要從s.codecs remove 這個codec
對s.codecs的add 和 remove需要添加互斥鎖呻拌，保證s.codecs的線程安全。

s.readRequest(codec) 處理請求的codec數(shù)據(jù)睦焕。

func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, Error) {
    reqs, batch, err := codec.ReadRequestHeaders()
    if err != nil {
        return nil, batch, err
    }

    requests := make([]*serverRequest, len(reqs))

    // verify requests
    for i, r := range reqs {
        var ok bool
        var svc *service

        if r.err != nil {
            requests[i] = &serverRequest{id: r.id, err: r.err}
            continue
        }

        if r.isPubSub && strings.HasSuffix(r.method, unsubscribeMethodSuffix) {
            requests[i] = &serverRequest{id: r.id, isUnsubscribe: true}
            argTypes := []reflect.Type{reflect.TypeOf("")} // expect subscription id as first arg
            if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil {
                requests[i].args = args
            } else {
                requests[i].err = &invalidParamsError{err.Error()}
            }
            continue
        }

        if svc, ok = s.services[r.service]; !ok { // rpc method isn't available
            requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}}
            continue
        }

        if r.isPubSub { // eth_subscribe, r.method contains the subscription method name
            if callb, ok := svc.subscriptions[r.method]; ok {
                requests[i] = &serverRequest{id: r.id, svcname: svc.name, callb: callb}
                if r.params != nil && len(callb.argTypes) > 0 {
                    argTypes := []reflect.Type{reflect.TypeOf("")}
                    argTypes = append(argTypes, callb.argTypes...)
                    if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil {
                        requests[i].args = args[1:] // first one is service.method name which isn't an actual argument
                    } else {
                        requests[i].err = &invalidParamsError{err.Error()}
                    }
                }
            } else {
                requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}}
            }
            continue
        }

        if callb, ok := svc.callbacks[r.method]; ok { // lookup RPC method
            requests[i] = &serverRequest{id: r.id, svcname: svc.name, callb: callb}
            if r.params != nil && len(callb.argTypes) > 0 {
                if args, err := codec.ParseRequestArguments(callb.argTypes, r.params); err == nil {
                    requests[i].args = args
                } else {
                    requests[i].err = &invalidParamsError{err.Error()}
                }
            }
            continue
        }

        requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.service, r.method}}
    }

    return requests, batch, nil
}

codec.ReadRequestHeaders()解析了請求數(shù)據(jù)

func (c *jsonCodec) ReadRequestHeaders() ([]rpcRequest, bool, Error) {
    c.decMu.Lock()
    defer c.decMu.Unlock()

    var incomingMsg json.RawMessage
    if err := c.d.Decode(&incomingMsg); err != nil {
        return nil, false, &invalidRequestError{err.Error()}
    }

    if isBatch(incomingMsg) {
        return parseBatchRequest(incomingMsg)
    }

    return parseRequest(incomingMsg)
}

如果請求的數(shù)據(jù)是一組req數(shù)組用parseBatchRequest(incomingMsg)解析藐握，否則用 parseRequest(incomingMsg)。兩者處理大同小異垃喊。

func parseRequest(incomingMsg json.RawMessage) ([]rpcRequest, bool, Error) {
    var in jsonRequest
    if err := json.Unmarshal(incomingMsg, &in); err != nil {
        return nil, false, &invalidMessageError{err.Error()}
    }

    if err := checkReqId(in.Id); err != nil {
        return nil, false, &invalidMessageError{err.Error()}
    }

    // subscribe are special, they will always use `subscribeMethod` as first param in the payload
    if strings.HasSuffix(in.Method, subscribeMethodSuffix) {
        reqs := []rpcRequest{{id: &in.Id, isPubSub: true}}
        if len(in.Payload) > 0 {
            // first param must be subscription name
            var subscribeMethod [1]string
            if err := json.Unmarshal(in.Payload, &subscribeMethod); err != nil {
                log.Debug(fmt.Sprintf("Unable to parse subscription method: %v\n", err))
                return nil, false, &invalidRequestError{"Unable to parse subscription request"}
            }

            reqs[0].service, reqs[0].method = strings.TrimSuffix(in.Method, subscribeMethodSuffix), subscribeMethod[0]
            reqs[0].params = in.Payload
            return reqs, false, nil
        }
        return nil, false, &invalidRequestError{"Unable to parse subscription request"}
    }

    if strings.HasSuffix(in.Method, unsubscribeMethodSuffix) {
        return []rpcRequest{{id: &in.Id, isPubSub: true,
            method: in.Method, params: in.Payload}}, false, nil
    }

    elems := strings.Split(in.Method, serviceMethodSeparator)
    if len(elems) != 2 {
        return nil, false, &methodNotFoundError{in.Method, ""}
    }

    // regular RPC call
    if len(in.Payload) == 0 {
        return []rpcRequest{{service: elems[0], method: elems[1], id: &in.Id}}, false, nil
    }

    return []rpcRequest{{service: elems[0], method: elems[1], id: &in.Id, params: in.Payload}}, false, nil
}

解析出service名字猾普，方法名，id本谜，請求參數(shù)組裝成rpcRequest對象初家，并返回。
readRequest(codec ServerCodec)方法對rpcRequest再處理加工一下乌助，然后返回溜在。

回到serveRequest方法，繼續(xù)分析s.exec(ctx, codec, reqs[0])的實現(xiàn)

func (s *Server) exec(ctx context.Context, codec ServerCodec, req *serverRequest) {
    var response interface{}
    var callback func()
    if req.err != nil {
        response = codec.CreateErrorResponse(&req.id, req.err)
    } else {
        response, callback = s.handle(ctx, codec, req)
    }

    if err := codec.Write(response); err != nil {
        log.Error(fmt.Sprintf("%v\n", err))
        codec.Close()
    }

    // when request was a subscribe request this allows these subscriptions to be actived
    if callback != nil {
        callback()
    }
}

交由s.handle(ctx, codec, req)處理

func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverRequest) (interface{}, func()) {
    if req.err != nil {
        return codec.CreateErrorResponse(&req.id, req.err), nil
    }

    if req.isUnsubscribe { // cancel subscription, first param must be the subscription id
        if len(req.args) >= 1 && req.args[0].Kind() == reflect.String {
            notifier, supported := NotifierFromContext(ctx)
            if !supported { // interface doesn't support subscriptions (e.g. http)
                return codec.CreateErrorResponse(&req.id, &callbackError{ErrNotificationsUnsupported.Error()}), nil
            }

            subid := ID(req.args[0].String())
            if err := notifier.unsubscribe(subid); err != nil {
                return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}), nil
            }

            return codec.CreateResponse(req.id, true), nil
        }
        return codec.CreateErrorResponse(&req.id, &invalidParamsError{"Expected subscription id as first argument"}), nil
    }

    if req.callb.isSubscribe {
        subid, err := s.createSubscription(ctx, codec, req)
        if err != nil {
            return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}), nil
        }

        // active the subscription after the sub id was successfully sent to the client
        activateSub := func() {
            notifier, _ := NotifierFromContext(ctx)
            notifier.activate(subid, req.svcname)
        }

        return codec.CreateResponse(req.id, subid), activateSub
    }

    // regular RPC call, prepare arguments
    if len(req.args) != len(req.callb.argTypes) {
        rpcErr := &invalidParamsError{fmt.Sprintf("%s%s%s expects %d parameters, got %d",
            req.svcname, serviceMethodSeparator, req.callb.method.Name,
            len(req.callb.argTypes), len(req.args))}
        return codec.CreateErrorResponse(&req.id, rpcErr), nil
    }

    arguments := []reflect.Value{req.callb.rcvr}
    if req.callb.hasCtx {
        arguments = append(arguments, reflect.ValueOf(ctx))
    }
    if len(req.args) > 0 {
        arguments = append(arguments, req.args...)
    }

    // execute RPC method and return result
    reply := req.callb.method.Func.Call(arguments)
    if len(reply) == 0 {
        return codec.CreateResponse(req.id, nil), nil
    }

    if req.callb.errPos >= 0 { // test if method returned an error
        if !reply[req.callb.errPos].IsNil() {
            e := reply[req.callb.errPos].Interface().(error)
            res := codec.CreateErrorResponse(&req.id, &callbackError{e.Error()})
            return res, nil
        }
    }
    return codec.CreateResponse(req.id, reply[0].Interface()), nil
}

跳過對訂閱和取消訂閱的請求處理他托。
reply := req.callb.method.Func.Call(arguments) 執(zhí)行了RPC方法并返回結(jié)果reply掖肋。
codec.CreateResponse(req.id, reply[0].Interface())是rpc.json.go對返回結(jié)果的封裝。
回到exec(ctx context.Context, codec ServerCodec, req *serverRequest)方法赏参。codec.Write(response)對返回結(jié)果json序列化培遵。
如果請求方法是訂閱執(zhí)行有回調(diào)callback()。

// 往client寫resp
func (c *jsonCodec) Write(res interface{}) error {
    c.encMu.Lock()
    defer c.encMu.Unlock()

    return c.e.Encode(res)
}

c.e.Encode(res)會調(diào)用enc.w.Write(b)登刺，這個w就是func (srv *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)方法傳入的http.ResponseWriter籽腕。借用這個writer來實現(xiàn)server和client的通信。

二纸俭，其他RPC 撥號的實現(xiàn)方法
RPC Client撥號的過程實質(zhì)是建立client和server的讀寫通道皇耗。
1，上一節(jié)分析的DialHTTPWithClient()方法揍很，RPC的Http服務(wù)郎楼，創(chuàng)建了一個httpConn通道。
2窒悔，RPC的WebSocket服務(wù)呜袁，撥號的實現(xiàn)方法：

func wsDialContext(ctx context.Context, config *websocket.Config) (*websocket.Conn, error) {
    var conn net.Conn
    var err error
    switch config.Location.Scheme {
    case "ws":
        conn, err = dialContext(ctx, "tcp", wsDialAddress(config.Location))
    case "wss":
        dialer := contextDialer(ctx)
        conn, err = tls.DialWithDialer(dialer, "tcp", wsDialAddress(config.Location), config.TlsConfig)
    default:
        err = websocket.ErrBadScheme
    }
    if err != nil {
        return nil, err
    }
    ws, err := websocket.NewClient(config, conn)
    if err != nil {
        conn.Close()
        return nil, err
    }
    return ws, err
}

dialContext創(chuàng)建了一個ws的net.conn，tls.DialWithDialer創(chuàng)建了一個wss的net.conn
3简珠，RPC的InProc服務(wù)阶界，撥號的實現(xiàn)方法

func DialInProc(handler *Server) *Client {
    initctx := context.Background()
    c, _ := newClient(initctx, func(context.Context) (net.Conn, error) {
        p1, p2 := net.Pipe()
        go handler.ServeCodec(NewJSONCodec(p1), OptionMethodInvocation|OptionSubscriptions)
        return p2, nil
    })
    return c
}

創(chuàng)建了一個net.Pipe通道
4虹钮，RPC的IPC服務(wù)，撥號的實現(xiàn)方法

func DialIPC(ctx context.Context, endpoint string) (*Client, error) {
    return newClient(ctx, func(ctx context.Context) (net.Conn, error) {
        return newIPCConnection(ctx, endpoint)
    })
}
//unix
func newIPCConnection(ctx context.Context, endpoint string) (net.Conn, error) {
    return dialContext(ctx, "unix", endpoint)
}
//windows
// newIPCConnection will connect to a named pipe with the given //endpoint as name.
func newIPCConnection(ctx context.Context, endpoint string) (net.Conn, error) {
    timeout := defaultPipeDialTimeout
    if deadline, ok := ctx.Deadline(); ok {
        timeout = deadline.Sub(time.Now())
        if timeout < 0 {
            timeout = 0
        }
    }
    return npipe.DialTimeout(endpoint, timeout)
}

如果是unix系統(tǒng)走的是websocket的創(chuàng)建方式膘融，創(chuàng)建一個net.conn通道芙粱，
如果是windows系統(tǒng)用第三方防范，創(chuàng)建了一個net.conn通道

三氧映，其他RPC Client如何發(fā)送請求
Rpc/client.go 的CallContext()方法春畔，如果不是http請求，選擇走c.send(ctx, op, msg)方法岛都。之所以會這樣是因為律姨，http是一個短連接痪蝇，每次請求是同步的绿语，直接返回請求結(jié)果甫窟。其他的比如IPC费坊、InProc、 websocket都是長連接缆娃，每次請求都是異步的，需要在網(wǎng)絡(luò)線程外監(jiān)聽請求返回的結(jié)果。

// send registers op with the dispatch loop, then sends msg on the connection.
// if sending fails, op is deregistered.
func (c *Client) send(ctx context.Context, op *requestOp, msg interface{}) error {
    select {
    case c.requestOp <- op:
        log.Trace("", "msg", log.Lazy{Fn: func() string {
            return fmt.Sprint("sending ", msg)
        }})
        err := c.write(ctx, msg)
        c.sendDone <- err
        return err
    case <-ctx.Done():
        // This can happen if the client is overloaded or unable to keep up with
        // subscription notifications.
        return ctx.Err()
    case <-c.didQuit:
        return ErrClientQuit
    }

這時候請求被select阻塞住塔逃，直到c.requestOp receive到op，或者receive 到 ctx.Done()料仗，或receive到 c.didQuit湾盗。c.requestOp拿到op，調(diào)用write方法把請求的內(nèi)容寫到conn通道去立轧。然后發(fā)送給sendDone chan格粪，client的dispactch方法會收到這個結(jié)果。

func (c *Client) dispatch(conn net.Conn) {
    // Spawn the initial read loop.
    go c.read(conn)

    var (
        lastOp        *requestOp    // tracks last send operation
        requestOpLock = c.requestOp // nil while the send lock is held
        reading       = true        // if true, a read loop is running
    )
    defer close(c.didQuit)
    defer func() {
        c.closeRequestOps(ErrClientQuit)
        conn.Close()
        if reading {
            // Empty read channels until read is dead.
            for {
                select {
                case <-c.readResp:
                case <-c.readErr:
                    return
                }
            }
        }
    }()

    for {
        select {
        case <-c.close:
            return

        // Read path.
        case batch := <-c.readResp:
            for _, msg := range batch {
                switch {
                case msg.isNotification():
                    log.Trace("", "msg", log.Lazy{Fn: func() string {
                        return fmt.Sprint("<-readResp: notification ", msg)
                    }})
                    c.handleNotification(msg)
                case msg.isResponse():
                    log.Trace("", "msg", log.Lazy{Fn: func() string {
                        return fmt.Sprint("<-readResp: response ", msg)
                    }})
                    c.handleResponse(msg)
                default:
                    log.Debug("", "msg", log.Lazy{Fn: func() string {
                        return fmt.Sprint("<-readResp: dropping weird message", msg)
                    }})
                    // TODO: maybe close
                }
            }

        case err := <-c.readErr:
            log.Debug(fmt.Sprintf("<-readErr: %v", err))
            c.closeRequestOps(err)
            conn.Close()
            reading = false

        case newconn := <-c.reconnected:
            log.Debug(fmt.Sprintf("<-reconnected: (reading=%t) %v", reading, conn.RemoteAddr()))
            if reading {
                // Wait for the previous read loop to exit. This is a rare case.
                conn.Close()
                <-c.readErr
            }
            go c.read(newconn)
            reading = true
            conn = newconn

        // Send path.
        case op := <-requestOpLock:
            // Stop listening for further send ops until the current one is done.
            requestOpLock = nil
            lastOp = op
            for _, id := range op.ids {
                c.respWait[string(id)] = op
            }

        case err := <-c.sendDone:
            if err != nil {
                // Remove response handlers for the last send. We remove those here
                // because the error is already handled in Call or BatchCall. When the
                // read loop goes down, it will signal all other current operations.
                for _, id := range lastOp.ids {
                    delete(c.respWait, string(id))
                }
            }
            // Listen for send ops again.
            requestOpLock = c.requestOp
            lastOp = nil
        }
    }
}

這個dispatch()方法也是配套給非http請求用的氛改。通過goroutine c.read(conn)帐萎。來讀server通過conn返回的數(shù)據(jù)。

func (c *Client) read(conn net.Conn) error {
    var (
        buf json.RawMessage
        dec = json.NewDecoder(conn)
    )
    readMessage := func() (rs []*jsonrpcMessage, err error) {
        buf = buf[:0]
        if err = dec.Decode(&buf); err != nil {
            return nil, err
        }
        if isBatch(buf) {
            err = json.Unmarshal(buf, &rs)
        } else {
            rs = make([]*jsonrpcMessage, 1)
            err = json.Unmarshal(buf, &rs[0])
        }
        return rs, err
    }

    for {
        resp, err := readMessage()
        if err != nil {
            c.readErr <- err
            return err
        }
        c.readResp <- resp
    }
}

然后把server返回數(shù)據(jù)send 到c.readResp chan胜卤。
dispatch的 select case batch := <-c.readResp: receive到c.readResp疆导。如果這個請求的是通知，走通知的響應(yīng)葛躏，否則走c.handleResponse(msg)

func (c *Client) handleResponse(msg *jsonrpcMessage) {
    op := c.respWait[string(msg.ID)]
    if op == nil {
        log.Debug(fmt.Sprintf("unsolicited response %v", msg))
        return
    }
    delete(c.respWait, string(msg.ID))
    // For normal responses, just forward the reply to Call/BatchCall.
    if op.sub == nil {
        op.resp <- msg
        return
    }
    // For subscription responses, start the subscription if the server
    // indicates success. EthSubscribe gets unblocked in either case through
    // the op.resp channel.
    defer close(op.resp)
    if msg.Error != nil {
        op.err = msg.Error
        return
    }
    if op.err = json.Unmarshal(msg.Result, &op.sub.subid); op.err == nil {
        go op.sub.start()
        c.subs[op.sub.subid] = op.sub
    }
}

這時候把返回數(shù)據(jù)send給op.resp <- msg澈段。 后續(xù)處理和http RPC的處理一致，走到CallContext方法的 resp, err := op.wait(ctx)舰攒。

四败富，總結(jié)

go-ethereum有四種RPC。HTTP RPC摩窃、Inproc RPC兽叮、IPC RPC、WS RPC。它們主要的實現(xiàn)邏輯都在rpc/server.go和rpc/client.go充择。各自根據(jù)自己的實現(xiàn)方式派生自己的client實例德玫，建立各自的net.conn通道。由于HTTP RPC是基于短鏈接請求椎麦，實現(xiàn)方式和其他的不太一樣宰僧。