【深度知识】RPC原理及以太坊RPC的实现

辉哥
更新于 2019-08-07 17:46
阅读 7930

RPC原理及以太坊RPC的实现

# 1.摘要

本文介绍RPC协议的原理和调用流程，同时介绍以太坊RPC的实现机制。

# 2\. 内容

## 2.1 RPC协议和调用流程

### 2.1.1 远程过程调用 (RPC)

Remote Procedure Calls 远程过程调用 (RPC) 是一种协议，就是从一台机器（客户端）上通过参数传递的方式调用另一台机器（服务器）上的一个函数或方法（可以统称为服务）并得到返回的结果。
通常的实现有 **XML-RPC , JSON-RPC ,** 通信方式基本相同, 所不同的只是传输数据的格式。

RPC是分布式架构的核心，按响应方式分如下两种：
**同步调用：** 客户端调用服务方方法，等待直到服务方返回结果或者超时，再继续自己的操作；
**异步调用：** 客户端把消息发送给中间件，不再等待服务端返回，直接继续自己的操作；

同步调用的实现方式有WebService和RMI。
Web Service提供的服务是基于web容器的，底层使用http协议，因而适合不同语言异构系统间的调用。
RMI（Remote Method Invocation，远程方法调用）实际上是Java语言的RPC实现，允许方法返回 Java 对象以及基本数据类型，适合用于JAVA语言构建的不同系统间的调用。

异步调用的JAVA实现版就是JMS(Java Message Service)，目前开源的的JMS中间件有Apache社区的ActiveMQ、Kafka消息中间件，另外有阿里的RocketMQ。

### 2.1.2 RPC框架

一个完整的RPC架构里面包含了四个核心的组件，分别是Client，Client Stub，Server以及Server Stub，这个Stub可以理解为存根。

* 客户端(Client)，服务的调用方。
* 客户端存根(Client Stub)，存放服务端的地址消息，再将客户端的请求参数打包成网络消息，然后通过网络远程发送给服务方。
* 服务端(Server)，真正的服务提供者。
* 服务端存根(Server Stub)，接收客户端发送过来的消息，将消息解包，并调用本地的方法。

![](https://img.learnblockchain.cn/2020/02/01_/117571352.png)

### 2.1.3 RPC调用流程

![](https://img.learnblockchain.cn/2020/02/01_/44853336.png)

**具体实现步骤：**
1、 服务调用方（client）(客户端)以本地调用方式调用服务；
2、 client stub接收到调用后负责将方法、参数等组装成能够进行网络传输的消息体；在Java里就是序列化的过程;
3、 client stub找到服务地址，并将消息通过网络发送到服务端；
4、 server stub收到消息后进行解码,在Java里就是反序列化的过程；
5、 server stub根据解码结果调用本地的服务；
6、 本地服务执行处理逻辑；
7、 本地服务将结果返回给server stub；
8、 server stub将返回结果打包成消息，Java里的序列化；
9、 server stub将打包后的消息通过网络并发送至消费方;
10、 client stub接收到消息，并进行解码, Java里的反序列化；
11、 服务调用方（client）得到最终结果。

RPC框架的目标就是把2-10步封装起来，把调用、编码/解码的过程封装起来，让用户像调用本地服务一样的调用远程服务。要做到对客户端（调用方）透明化服务， RPC框架需要考虑解决如下问题：
**1、通讯问题** : 主要是通过在客户端和服务器之间建立TCP连接，远程过程调用的所有交换的数据都在这个连接里传输。连接可以是按需连接，调用结束后就断掉，也可以是长连接，多个远程过程调用共享同一个连接。
**2、寻址问题** ： A服务器上的应用怎么告诉底层的RPC框架，如何连接到B服务器（如主机或IP地址）以及特定的端口，方法的名称是什么，这样才能完成调用。比如基于Web服务协议栈的RPC，就要提供一个endpoint URI，或者是从UDDI服务上查找。如果是RMI调用的话，还需要一个RMI Registry来注册服务的地址。
**3、序列化与反序列化** ： 当A服务器上的应用发起远程过程调用时，方法的参数需要通过底层的网络协议如TCP传递到B服务器，由于网络协议是基于二进制的，内存中的参数的值要序列化成二进制的形式，也就是序列化（Serialize）或编组（marshal），通过寻址和传输将序列化的二进制发送给B服务器。
同理，B服务器接收参数要将参数反序列化。B服务器应用调用自己的方法处理后返回的结果也要序列化给A服务器，A服务器接收也要经过反序列化的过程。

## 2.2 以太坊RPC实现

JSON-RPC是区块链外部调用的标配了。以太坊同样也实现了这个功能。底层支持四种协议：InProc，IPC，HTTP，WEBSOCKED。上层除了常规的方法调用之外还实现了Pub/Sub功能。本文主要分析以太坊是如何支持这些个功能的。

### 2.2.1 api发布

api接口分布在各个模块，主要分为两种

* 1：直接code在Node中的几个service(admin,web3j,debug etc)
* 2: 实现了Service接口的服务结构，已经注册的服务会调用APIs()方法获得其中的api。

```
//file go-ethereum/node/node.go
func (n *Node) startRPC(services map[reflect.Type]Service) error {
    apis := n.apis()
    for _, service := range services {
        apis = append(apis, service.APIs()...)
    }
}

```

node中写死的接口

```
    // node中写死的接口
    func (n *Node) apis() []rpc.API {
        return []rpc.API{
            {
                Namespace: "admin",
                Version:   "1.0",
                Service:   NewPrivateAdminAPI(n),
            }, {
                Namespace: "admin",
                Version:   "1.0",
                Service:   NewPublicAdminAPI(n),
                Public:    true,
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   debug.Handler,
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   NewPublicDebugAPI(n),
                Public:    true,
            }, {
                Namespace: "web3",
                Version:   "1.0",
                Service:   NewPublicWeb3API(n),
                Public:    true,
            },
        }
    }

```

Ethereum 服务实现的APIs()接口 类似的还有其他的服务（dashboard，ethstats）

```
    //Ethereum 服务实现的APIs()接口
    func (s *Ethereum) APIs() []rpc.API {
        apis := ethapi.GetAPIs(s.ApiBackend)

// Append any APIs exposed explicitly by the consensus engine
        apis = append(apis, s.engine.APIs(s.BlockChain())...)

// Append all the local APIs and return
        return append(apis, []rpc.API{
            {
                Namespace: "eth",
                Version:   "1.0",
                Service:   NewPublicEthereumAPI(s),
                Public:    true,
            }, {
                Namespace: "eth",
                Version:   "1.0",
                Service:   NewPublicMinerAPI(s),
                Public:    true,
            }, {
                Namespace: "eth",
                Version:   "1.0",
                Service:   downloader.NewPublicDownloaderAPI(s.protocolManager.downloader, s.eventMux),
                Public:    true,
            }, {
                Namespace: "miner",
                Version:   "1.0",
                Service:   NewPrivateMinerAPI(s),
                Public:    false,
            }, {
                Namespace: "eth",
                Version:   "1.0",
                Service:   filters.NewPublicFilterAPI(s.ApiBackend, false),
                Public:    true,
            }, {
                Namespace: "admin",
                Version:   "1.0",
                Service:   NewPrivateAdminAPI(s),
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   NewPublicDebugAPI(s),
                Public:    true,
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   NewPrivateDebugAPI(s.chainConfig, s),
            }, {
                Namespace: "net",
                Version:   "1.0",
                Service:   s.netRPCService,
                Public:    true,
            },
        }...)
    }

```

这里的Service只是类型，还要注册到Server里面，原理就是反射出结构体里的类型，解析出函数方法名称（转小写），参数名称，返回类型等信息，最终每个合格的方法都会生成service实例。

```
    type service struct {
        name          string        // name for service
        typ           reflect.Type  // receiver type
        callbacks     callbacks     // registered handlers
        subscriptions subscriptions // available subscriptions/notifications
    }

```

```
    //反射除Service Api的结构方法
    //file go-ethereum/rpc/utils.go
    func suitableCallbacks(rcvr reflect.Value, typ reflect.Type) (callbacks, subscriptions) {
        callbacks := make(callbacks)
        subscriptions := make(subscriptions)

METHODS:
        for m := 0; m < typ.NumMethod(); m++ {
            method := typ.Method(m)
            mtype := method.Type
            //转小写
            mname := formatName(method.Name)
            if method.PkgPath != "" { // method must be exported
                continue
            }

var h callback
            //订阅事件类型判断 主要根据签名的入参第二位和返回参数第一位
            h.isSubscribe = isPubSub(mtype)  
            h.rcvr = rcvr
            h.method = method
            h.errPos = -1

firstArg := 1
            numIn := mtype.NumIn()
            if numIn >= 2 && mtype.In(1) == contextType {
                h.hasCtx = true
                firstArg = 2
            }

if h.isSubscribe {
                //订阅类型
                h.argTypes = make([]reflect.Type, numIn-firstArg) // skip rcvr type
                for i := firstArg; i < numIn; i++ {
                    argType := mtype.In(i)
                    if isExportedOrBuiltinType(argType) {
                        h.argTypes[i-firstArg] = argType
                    } else {
                        continue METHODS
                    }
                }

subscriptions[mname] = &h
                continue METHODS
            }

// determine method arguments, ignore first arg since it's the receiver type
            // Arguments must be exported or builtin types
            h.argTypes = make([]reflect.Type, numIn-firstArg)
            for i := firstArg; i < numIn; i++ {
                argType := mtype.In(i)
                if !isExportedOrBuiltinType(argType) {
                    continue METHODS
                }
                h.argTypes[i-firstArg] = argType
            }

// check that all returned values are exported or builtin types
            for i := 0; i < mtype.NumOut(); i++ {
                if !isExportedOrBuiltinType(mtype.Out(i)) {
                    continue METHODS
                }
            }

// when a method returns an error it must be the last returned value
            h.errPos = -1
            for i := 0; i < mtype.NumOut(); i++ {
                if isErrorType(mtype.Out(i)) {
                    h.errPos = i
                    break
                }
            }

if h.errPos >= 0 && h.errPos != mtype.NumOut()-1 {
                continue METHODS
            }

switch mtype.NumOut() {
            case 0, 1, 2:
                if mtype.NumOut() == 2 && h.errPos == -1 { // method must one return value and 1 error
                    continue METHODS
                }
                callbacks[mname] = &h
            }
        }

return callbacks, subscriptions
    }

```

### 2.2.2 底层协议

底层支持了InProc，IPC，HTTP，WEBSOCKED 四种传输协议

* 1 InProc 直接生成RPCService实例，挂在Node上面可以直接调用。
* 2 IPC 监听管道，收到消息后解析成ServerCodec对象，扔给Server的ServeCodec方法使用；

```
    //file ipc.go
    func (srv *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 connection", "addr", conn.RemoteAddr())
            go srv.ServeCodec(NewJSONCodec(conn), OptionMethodInvocation|OptionSubscriptions)
        }
    }

```

* 3 HTTP 生成两个中间件，第二个中间件接收消息生成ServerCOdec，扔给Server的ServeSingleRequest方法

```
    //file http.go
    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.
        ctx := context.Background()
        ctx = context.WithValue(ctx, "remote", r.RemoteAddr)
        ctx = context.WithValue(ctx, "scheme", r.Proto)
        ctx = context.WithValue(ctx, "local", r.Host)

body := io.LimitReader(r.Body, maxRequestContentLength)
        codec := NewJSONCodec(&httpReadWriteNopCloser{body, w})
        defer codec.Close()

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

```

* 1 WEBSOCKED 与Http类型生成WebsocketHandler中间件，到消息后解析成ServerCodec对象，扔给Server的ServeCodec方法使用

```
    //websocked.go
    func (srv *Server) WebsocketHandler(allowedOrigins []string) http.Handler {
        return websocket.Server{
            Handshake: wsHandshakeValidator(allowedOrigins),
            Handler: func(conn *websocket.Conn) {
                // Create a custom encode/decode pair to enforce payload size and number encoding
                conn.MaxPayloadBytes = maxRequestContentLength

encoder := func(v interface{}) error {
                    return websocketJSONCodec.Send(conn, v)
                }
                decoder := func(v interface{}) error {
                    return websocketJSONCodec.Receive(conn, v)
                }
                srv.ServeCodec(NewCodec(conn, encoder, decoder), OptionMethodInvocation|OptionSubscriptions)
            },
        }
    }

```

### 2.2.3 rpc响应

上面四种协议再拿到ServerCodec对象后，会把这个对象传递到service的响应请数里面去。最终都是调到handle函数里面，handle里面再根据不同的类型进行响应。

```
    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 { 
            //取消订阅功能
            if len(req.args) >= 1 && req.args[0].Kind() == reflect.String {
                notifier, supported := NotifierFromContext(ctx)  //获取notifier对象
                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对象
                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
    }

```

### 2.2.4 Pub/sub 实现

底层在context绑定一个notifier对象

```
    if options&OptionSubscriptions == OptionSubscriptions {
        ctx = context.WithValue(ctx, notifierKey{}, newNotifier(codec))
    }

```

sub/unsub的时候会通过context.Value中拿notifier对象，调用上面的方法

##### 注册或者取消注册

```
    func NotifierFromContext(ctx context.Context) (*Notifier, bool) {
        n, ok := ctx.Value(notifierKey{}).(*Notifier)
        return n, ok
    }

```

##### 注册

```
    func (n *Notifier) activate(id ID, namespace string) {
        n.subMu.Lock()
        defer n.subMu.Unlock()
        if sub, found := n.inactive[id]; found {
            sub.namespace = namespace
            n.active[id] = sub
            delete(n.inactive, id)
        }
    }

```

##### 注销

```
    func (n *Notifier) unsubscribe(id ID) error {
        n.subMu.Lock()
        defer n.subMu.Unlock()
        if s, found := n.active[id]; found {
            close(s.err)
            delete(n.active, id)
            return nil
        }
        return ErrSubscriptionNotFound
    }

```

##### 消息事件触发

```
    func (api *PrivateAdminAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) {
        // Make sure the server is running, fail otherwise
        server := api.node.Server()
        if server == nil {
            return nil, ErrNodeStopped
        }

// Create the subscription
        //获取notifier对象
        notifier, supported := rpc.NotifierFromContext(ctx)
        if !supported {
            return nil, rpc.ErrNotificationsUnsupported
        }
        //生成标识
        rpcSub := notifier.CreateSubscription()

go func() {
            events := make(chan *p2p.PeerEvent)
            sub := server.SubscribeEvents(events)
            defer sub.Unsubscribe()

for {
                select {
                case event := <-events:
                    //触发事件，发送通知消息
                    notifier.Notify(rpcSub.ID, event)
                case <-sub.Err():
                    return
                case <-rpcSub.Err():
                    return
                case <-notifier.Closed():
                    return
                }
            }
        }()

return rpcSub, nil
    }

```

## 2.2.5 rpc client调用

以太坊提供了RPC服务，可以在geth启动时通过参数设置

### 2.2.5.1 geth启动选项参数

```
--rpc  启动HTTP-RPC服务（基于HTTP的）
--ws  启动WS-RPC服务（基于WebService的）
--rpcapi  value 指定需要调用的HTTP-RPC API接口，默认只有eth,net,web3
--rpcport value  HTTP-RPC服务器监听端口(default: 8545)
--rpcport value  HTTP-RPC服务器监听端口(default: 8545)
例子：geth --rpc --rpcapi "db,eth,net,web3,personal"

```

执行RPC调用的方式有很多，可以使用web3提供的接口、直接发送Json请求（缺点是拼json会很麻烦）、使用go-ethereum/ethclient包提供的函数（缺点是只有eth接口）、也可以自己定义接口来调用。下面代码是使用go-ethereum/ethclient包中的函数的例子。

```
package main

import (
    "fmt"
    "github.com/ethereum/go-ethereum/mobile"
)

func main() {
    // NewEthereumClient函数只是创建一个EthereumClient结构，并设置了HTTP连接的一些参数如的head的一些属性，并没有节点建立连接
    cli, err := geth.NewEthereumClient("http://127.0.0.1:8545")
    if err != nil {
        fmt.Printf("create new ethereum rpc client err:%s\n", err.Error())
    } else {
        fmt.Println("create new ethereum rpc client success")
    }
    eth_ctx := geth.NewContext()
    block, err2 := cli.GetBlockByNumber(eth_ctx, 18)
    fmt.Printf("ethereum mobile Context:%+v\n", eth_ctx)
    if err2 != nil {
        fmt.Printf("get block err:%s\n", err2.Error())
    } else {
        fmt.Printf("block:%+v\n", block)
    }
}

```

连的节点是本地运行的私有链，并且在go-ethereum源码中加了一些日志，执行结果：

```
mylog:DialContext:u:{Scheme:http Opaque: User: Host:127.0.0.1:8545 Path: RawPath: ForceQuery:false RawQuery: Fragment:};
mylog:u.Scheme:http
create new ethereum rpc client success
mylog:JSON-RPC: Client CallContext
mylog:Client.isHTTP:true
ethereum mobile Context:&{context:0xc4200ac008 cancel:<nil>}
block:Block(#18): Size: 650.00 B {
MinerHash: fd55c05ae10a5b0159b3c2d5803c6aa9469c95f5f063b9c400a2c36b49616ab3
Header(84b2cfd65e3197bdfe3f748ecebb040953af5eb73a05d8595757cf42cb40a492):
[
    ParentHash:     7892a0b31d50d67ae20d4a7ec5c24a6fe85f2f264e9f1639aa2388081305a0bd
    UncleHash:      1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347
    Coinbase:       bdc61c81f67983288a6c375a884661edc77286d0
    Root:           0f30637bfc5bd6e123c6a0c38bdc743c94050626a984f9943eaf38367100b3e3
    TxSha           354d185cfa88e50f1a425e5b89500122e4445e9ec737e7a18cdd61b9350ab72b
    ReceiptSha:     a769d28981014fb6095462148a6300cd0b43fa050d75eb6f5b7595cfd13136bb
    Bloom:          00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
    Difficulty:     131072
    Number:         18
    GasLimit:       131877941
    GasUsed:        21000
    Time:           1527044372
    Extra:          ׃��geth�go1.10�darwin
    MixDigest:      70c2bb422b1b834d5173d279e508ffee9dada454650fc3cf63e95deb3073cf32
    Nonce:          58b7495f112ccac2
]
Transactions:
[
    TX(57a3b17f84358098b728fc0f70f0697f175f8ba00d386c88eac0815b3afd6aad)
    Contract: false
    From:     2154bdd7070c99d1a25ff589a08b01dfd6eb65de
    To:       bdc61c81f67983288a6c375a884661edc77286d0
    Nonce:    0
    GasPrice: 0x430e23400
    GasLimit  0x15f90
    Value:    0xde0b6b3a7640000
    Data:     0x
    V:        0x41
    R:        0x45d4952c0190373c56e62ad15e54db54c0246385371b23c70bab4126b51927f8
    S:        0x618e4bb76a36482254352d7e5096c0dff4c1f495218d57c874fc3d8153915ea4
    Hex:      f86d80850430e2340083015f9094bdc61c81f67983288a6c375a884661edc77286d0880de0b6b3a76400008041a045d4952c0190373c56e62ad15e54db54c0246385371b23c70bab4126b51927f8a0618e4bb76a36482254352d7e5096c0dff4c1f495218d57c874fc3d8153915ea4
]
Uncles:
[]
}

```

### 2.2.5.2 分析：

**go-ethereum/mobile包** 是发起RPC请求的客户端直接使用的包。
该包中有 `EthereumClient` 结构提供了Ethereum API的接入。

```
// EthereumClient provides access to the Ethereum APIs.
type EthereumClient struct {
    client *ethclient.Client
}

```

ethclient.Client在ethclient包中，包装了rpc.Client，rpc.Client代表与RPC服务的一个连接。

```
// Client defines typed wrappers for the Ethereum RPC API.
type Client struct {
    c *rpc.Client
}

```

RPC请求客户端在使用时，首先传入想要接入的节点的url作为参数，调用mobile包中的NewEthereumClient函数。创建了EthereumClient实例，并与节点建立连接。建立的RPC连接有三种形式：HTTP、WebSocket、IPC，当传入 `http://127.0.0.1:8545` 时，建立的是HTTP连接。

```
// NewEthereumClient connects a client to the given URL.
func NewEthereumClient(rawurl string) (client *EthereumClient, _ error) {
    rawClient, err := ethclient.Dial(rawurl)
    return &EthereumClient{rawClient}, err
}

```

设置HTTP连接的参数会调用rpc包http.go文件中的DialHTTPWithClient函数。

```
// DialHTTPWithClient creates a new RPC client that connects to an RPC server over HTTP
// using the provided HTTP Client.
func DialHTTPWithClient(endpoint string, client *http.Client) (*Client, error) {
    req, err := http.NewRequest(http.MethodPost, endpoint, nil)
    if err != nil {
        return nil, err
    }
    // Content-Type和Accept是application/json，即发送的数据类型和接收的数据类型都是json
    req.Header.Set("Content-Type", contentType)
    req.Header.Set("Accept", contentType)

initctx := context.Background()
    return newClient(initctx, func(context.Context) (net.Conn, error) {
        return &httpConn{client: client, req: req, closed: make(chan struct{})}, nil
    })
}

```

通过HTTP来做JSON-RPC调用时，需要一个geth.Context实例，通过调用mobile包中的NewContext函数，创建一个空的geth.Context实例。

```
// NewContext returns a non-nil, empty Context. It is never canceled, has no
// values, and has no deadline. It is typically used by the main function,
// initialization, and tests, and as the top-level Context for incoming requests.
func NewContext() *Context {
    return &Context{
        context: context.Background(),
    }
}

```

mobile包中封装了请求区块、区块头、交易等函数，这些函数调用ethclient包中的相关函数，再调用更底层rpc包中封装的函数。
即 **mobile包-->ethclient包-->rpc包** 。如mobile包中根据区块号查找区块的函数最后会调用rpc包中的CallContext函数。

```
// CallContext扮演JSON-RPC调用角色
// CallContext performs a JSON-RPC call with the given arguments. If the context is
// canceled before the call has successfully returned, CallContext returns immediately.
//
// The result must be a pointer so that package json can unmarshal into it. You
// can also pass nil, in which case the result is ignored.
func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
    fmt.Printf("mylog:JSON-RPC: Client CallContext\n")
    msg, err := c.newMessage(method, args...)
    if err != nil {
        return err
    }
    op := &requestOp{ids: []json.RawMessage{msg.ID}, resp: make(chan *jsonrpcMessage, 1)}

fmt.Printf("mylog:Client.isHTTP:%+v\n",c.isHTTP)
    if c.isHTTP {
        err = c.sendHTTP(ctx, op, msg)
    } else {
        err = c.send(ctx, op, msg)
    }
    if err != nil {
        return err
    }

// dispatch has accepted the request and will close the channel it when it quits.
    switch resp, err := op.wait(ctx); {
    case err != nil:
        return err
    case resp.Error != nil:
        return resp.Error
    case len(resp.Result) == 0:
        return ErrNoResult
    default:
        return json.Unmarshal(resp.Result, &result)
    }
}

```

### 2.2.5.3 使用POSTMAN

使用POSTMAN发送请求时，注意设置下Content-type和Accept。
body是 `{"jsonrpc":"2.0","method":"web3_clientVersion","params":[],"id":67}`
这种方式虽然直接，但是自己拼json会很麻烦，所以最方便的还是调用已有的接口。

![](https://img.learnblockchain.cn/2020/02/01_/459602498.png)

如果是做查询区块号为18的区块，则body是
`{"jsonrpc":"2.0","method":"eth_getBlockByNumber","params":["0x12",true],"id":1}`

# 3\. 参考

（1）以太坊源码深入分析（3）-- 以太坊RPC通信实例和原理代码分析（上）
[https://www.jianshu.com/p/92daf6148dc5](https://www.jianshu.com/p/92daf6148dc5)
（2）以太坊RPC
[https://www.jianshu.com/p/8bd3723aa921](https://www.jianshu.com/p/8bd3723aa921)
（3）以太坊RPC原理及实现
[https://my.oschina.net/hunjixin/blog/1803161](https://my.oschina.net/hunjixin/blog/1803161)
（4）从零开始实现RPC框架 - RPC原理及实现
[https://www.jianshu.com/p/dbfac2b876b1](https://www.jianshu.com/p/dbfac2b876b1)
（5）深入浅出RPC原理
[https://ketao1989.github.io/2016/12/10/rpc-theory-in-action/](https://ketao1989.github.io/2016/12/10/rpc-theory-in-action/)
（6）你应该知道的RPC原理
[https://www.cnblogs.com/LBSer/p/4853234.html](https://www.cnblogs.com/LBSer/p/4853234.html)
（7）RPC原理解析
[https://www.cnblogs.com/swordfall/p/8683905.html](https://www.cnblogs.com/swordfall/p/8683905.html)
（8）服务之间的调用之RPC、Restful深入理解
[https://blog.csdn.net/u014590757/article/details/80233901](https://blog.csdn.net/u014590757/article/details/80233901)

1.摘要

本文介绍RPC协议的原理和调用流程，同时介绍以太坊RPC的实现机制。

2. 内容

2.1 RPC协议和调用流程

2.1.1 远程过程调用 (RPC)

Remote Procedure Calls 远程过程调用 (RPC) 是一种协议，就是从一台机器（客户端）上通过参数传递的方式调用另一台机器（服务器）上的一个函数或方法（可以统称为服务）并得到返回的结果。通常的实现有 XML-RPC , JSON-RPC , 通信方式基本相同, 所不同的只是传输数据的格式。

RPC是分布式架构的核心，按响应方式分如下两种： 同步调用： 客户端调用服务方方法，等待直到服务方返回结果或者超时，再继续自己的操作； 异步调用： 客户端把消息发送给中间件，不再等待服务端返回，直接继续自己的操作；

同步调用的实现方式有WebService和RMI。 Web Service提供的服务是基于web容器的，底层使用http协议，因而适合不同语言异构系统间的调用。 RMI（Remote Method Invocation，远程方法调用）实际上是Java语言的RPC实现，允许方法返回 Java 对象以及基本数据类型，适合用于JAVA语言构建的不同系统间的调用。

异步调用的JAVA实现版就是JMS(Java Message Service)，目前开源的的JMS中间件有Apache社区的ActiveMQ、Kafka消息中间件，另外有阿里的RocketMQ。

2.1.2 RPC框架

一个完整的RPC架构里面包含了四个核心的组件，分别是Client，Client Stub，Server以及Server Stub，这个Stub可以理解为存根。

客户端(Client)，服务的调用方。
客户端存根(Client Stub)，存放服务端的地址消息，再将客户端的请求参数打包成网络消息，然后通过网络远程发送给服务方。
服务端(Server)，真正的服务提供者。
服务端存根(Server Stub)，接收客户端发送过来的消息，将消息解包，并调用本地的方法。

2.1.3 RPC调用流程

具体实现步骤： 1、服务调用方（client）(客户端)以本地调用方式调用服务； 2、 client stub接收到调用后负责将方法、参数等组装成能够进行网络传输的消息体；在Java里就是序列化的过程; 3、 client stub找到服务地址，并将消息通过网络发送到服务端； 4、 server stub收到消息后进行解码,在Java里就是反序列化的过程； 5、 server stub根据解码结果调用本地的服务； 6、本地服务执行处理逻辑； 7、本地服务将结果返回给server stub； 8、 server stub将返回结果打包成消息，Java里的序列化； 9、 server stub将打包后的消息通过网络并发送至消费方; 10、 client stub接收到消息，并进行解码, Java里的反序列化； 11、服务调用方（client）得到最终结果。

RPC框架的目标就是把2-10步封装起来，把调用、编码/解码的过程封装起来，让用户像调用本地服务一样的调用远程服务。要做到对客户端（调用方）透明化服务， RPC框架需要考虑解决如下问题： 1、通讯问题 : 主要是通过在客户端和服务器之间建立TCP连接，远程过程调用的所有交换的数据都在这个连接里传输。连接可以是按需连接，调用结束后就断掉，也可以是长连接，多个远程过程调用共享同一个连接。 2、寻址问题 ： A服务器上的应用怎么告诉底层的RPC框架，如何连接到B服务器（如主机或IP地址）以及特定的端口，方法的名称是什么，这样才能完成调用。比如基于Web服务协议栈的RPC，就要提供一个endpoint URI，或者是从UDDI服务上查找。如果是RMI调用的话，还需要一个RMI Registry来注册服务的地址。 3、序列化与反序列化 ：当A服务器上的应用发起远程过程调用时，方法的参数需要通过底层的网络协议如TCP传递到B服务器，由于网络协议是基于二进制的，内存中的参数的值要序列化成二进制的形式，也就是序列化（Serialize）或编组（marshal），通过寻址和传输将序列化的二进制发送给B服务器。同理，B服务器接收参数要将参数反序列化。B服务器应用调用自己的方法处理后返回的结果也要序列化给A服务器，A服务器接收也要经过反序列化的过程。

2.2 以太坊RPC实现

2.2.1 api发布

api接口分布在各个模块，主要分为两种

1：直接code在Node中的几个service(admin,web3j,debug etc)
2: 实现了Service接口的服务结构，已经注册的服务会调用APIs()方法获得其中的api。

//file go-ethereum/node/node.go
func (n *Node) startRPC(services map[reflect.Type]Service) error {
    apis := n.apis()
    for _, service := range services {
        apis = append(apis, service.APIs()...)
    }
}

node中写死的接口

    // node中写死的接口
    func (n *Node) apis() []rpc.API {
        return []rpc.API{
            {
                Namespace: "admin",
                Version:   "1.0",
                Service:   NewPrivateAdminAPI(n),
            }, {
                Namespace: "admin",
                Version:   "1.0",
                Service:   NewPublicAdminAPI(n),
                Public:    true,
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   debug.Handler,
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   NewPublicDebugAPI(n),
                Public:    true,
            }, {
                Namespace: "web3",
                Version:   "1.0",
                Service:   NewPublicWeb3API(n),
                Public:    true,
            },
        }
    }

Ethereum 服务实现的APIs()接口类似的还有其他的服务（dashboard，ethstats）

    //Ethereum 服务实现的APIs()接口
    func (s *Ethereum) APIs() []rpc.API {
        apis := ethapi.GetAPIs(s.ApiBackend)

        // Append any APIs exposed explicitly by the consensus engine
        apis = append(apis, s.engine.APIs(s.BlockChain())...)

        // Append all the local APIs and return
        return append(apis, []rpc.API{
            {
                Namespace: "eth",
                Version:   "1.0",
                Service:   NewPublicEthereumAPI(s),
                Public:    true,
            }, {
                Namespace: "eth",
                Version:   "1.0",
                Service:   NewPublicMinerAPI(s),
                Public:    true,
            }, {
                Namespace: "eth",
                Version:   "1.0",
                Service:   downloader.NewPublicDownloaderAPI(s.protocolManager.downloader, s.eventMux),
                Public:    true,
            }, {
                Namespace: "miner",
                Version:   "1.0",
                Service:   NewPrivateMinerAPI(s),
                Public:    false,
            }, {
                Namespace: "eth",
                Version:   "1.0",
                Service:   filters.NewPublicFilterAPI(s.ApiBackend, false),
                Public:    true,
            }, {
                Namespace: "admin",
                Version:   "1.0",
                Service:   NewPrivateAdminAPI(s),
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   NewPublicDebugAPI(s),
                Public:    true,
            }, {
                Namespace: "debug",
                Version:   "1.0",
                Service:   NewPrivateDebugAPI(s.chainConfig, s),
            }, {
                Namespace: "net",
                Version:   "1.0",
                Service:   s.netRPCService,
                Public:    true,
            },
        }...)
    }

    type service struct {
        name          string        // name for service
        typ           reflect.Type  // receiver type
        callbacks     callbacks     // registered handlers
        subscriptions subscriptions // available subscriptions/notifications
    }

    //反射除Service Api的结构方法
    //file go-ethereum/rpc/utils.go
    func suitableCallbacks(rcvr reflect.Value, typ reflect.Type) (callbacks, subscriptions) {
        callbacks := make(callbacks)
        subscriptions := make(subscriptions)

    METHODS:
        for m := 0; m &lt; typ.NumMethod(); m++ {
            method := typ.Method(m)
            mtype := method.Type
            //转小写
            mname := formatName(method.Name)
            if method.PkgPath != "" { // method must be exported
                continue
            }

            var h callback
            //订阅事件类型判断 主要根据签名的入参第二位和返回参数第一位
            h.isSubscribe = isPubSub(mtype)  
            h.rcvr = rcvr
            h.method = method
            h.errPos = -1

            firstArg := 1
            numIn := mtype.NumIn()
            if numIn >= 2 && mtype.In(1) == contextType {
                h.hasCtx = true
                firstArg = 2
            }

            if h.isSubscribe {
                //订阅类型
                h.argTypes = make([]reflect.Type, numIn-firstArg) // skip rcvr type
                for i := firstArg; i &lt; numIn; i++ {
                    argType := mtype.In(i)
                    if isExportedOrBuiltinType(argType) {
                        h.argTypes[i-firstArg] = argType
                    } else {
                        continue METHODS
                    }
                }

                subscriptions[mname] = &h
                continue METHODS
            }

            // determine method arguments, ignore first arg since it's the receiver type
            // Arguments must be exported or builtin types
            h.argTypes = make([]reflect.Type, numIn-firstArg)
            for i := firstArg; i &lt; numIn; i++ {
                argType := mtype.In(i)
                if !isExportedOrBuiltinType(argType) {
                    continue METHODS
                }
                h.argTypes[i-firstArg] = argType
            }

            // check that all returned values are exported or builtin types
            for i := 0; i &lt; mtype.NumOut(); i++ {
                if !isExportedOrBuiltinType(mtype.Out(i)) {
                    continue METHODS
                }
            }

            // when a method returns an error it must be the last returned value
            h.errPos = -1
            for i := 0; i &lt; mtype.NumOut(); i++ {
                if isErrorType(mtype.Out(i)) {
                    h.errPos = i
                    break
                }
            }

            if h.errPos >= 0 && h.errPos != mtype.NumOut()-1 {
                continue METHODS
            }

            switch mtype.NumOut() {
            case 0, 1, 2:
                if mtype.NumOut() == 2 && h.errPos == -1 { // method must one return value and 1 error
                    continue METHODS
                }
                callbacks[mname] = &h
            }
        }

        return callbacks, subscriptions
    }

2.2.2 底层协议

底层支持了InProc，IPC，HTTP，WEBSOCKED 四种传输协议

1 InProc 直接生成RPCService实例，挂在Node上面可以直接调用。
2 IPC 监听管道，收到消息后解析成ServerCodec对象，扔给Server的ServeCodec方法使用；

    //file ipc.go
    func (srv *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 connection", "addr", conn.RemoteAddr())
            go srv.ServeCodec(NewJSONCodec(conn), OptionMethodInvocation|OptionSubscriptions)
        }
    }

3 HTTP 生成两个中间件，第二个中间件接收消息生成ServerCOdec，扔给Server的ServeSingleRequest方法

    //file http.go
    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.
        ctx := context.Background()
        ctx = context.WithValue(ctx, "remote", r.RemoteAddr)
        ctx = context.WithValue(ctx, "scheme", r.Proto)
        ctx = context.WithValue(ctx, "local", r.Host)

        body := io.LimitReader(r.Body, maxRequestContentLength)
        codec := NewJSONCodec(&httpReadWriteNopCloser{body, w})
        defer codec.Close()

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

1 WEBSOCKED 与Http类型生成WebsocketHandler中间件，到消息后解析成ServerCodec对象，扔给Server的ServeCodec方法使用

    //websocked.go
    func (srv *Server) WebsocketHandler(allowedOrigins []string) http.Handler {
        return websocket.Server{
            Handshake: wsHandshakeValidator(allowedOrigins),
            Handler: func(conn *websocket.Conn) {
                // Create a custom encode/decode pair to enforce payload size and number encoding
                conn.MaxPayloadBytes = maxRequestContentLength

                encoder := func(v interface{}) error {
                    return websocketJSONCodec.Send(conn, v)
                }
                decoder := func(v interface{}) error {
                    return websocketJSONCodec.Receive(conn, v)
                }
                srv.ServeCodec(NewCodec(conn, encoder, decoder), OptionMethodInvocation|OptionSubscriptions)
            },
        }
    }

2.2.3 rpc响应

    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 { 
            //取消订阅功能
            if len(req.args) >= 1 && req.args[0].Kind() == reflect.String {
                notifier, supported := NotifierFromContext(ctx)  //获取notifier对象
                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对象
                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
    }

2.2.4 Pub/sub 实现

底层在context绑定一个notifier对象

    if options&OptionSubscriptions == OptionSubscriptions {
        ctx = context.WithValue(ctx, notifierKey{}, newNotifier(codec))
    }

sub/unsub的时候会通过context.Value中拿notifier对象，调用上面的方法

注册或者取消注册

    func NotifierFromContext(ctx context.Context) (*Notifier, bool) {
        n, ok := ctx.Value(notifierKey{}).(*Notifier)
        return n, ok
    }

注册

    func (n *Notifier) activate(id ID, namespace string) {
        n.subMu.Lock()
        defer n.subMu.Unlock()
        if sub, found := n.inactive[id]; found {
            sub.namespace = namespace
            n.active[id] = sub
            delete(n.inactive, id)
        }
    }

注销

    func (n *Notifier) unsubscribe(id ID) error {
        n.subMu.Lock()
        defer n.subMu.Unlock()
        if s, found := n.active[id]; found {
            close(s.err)
            delete(n.active, id)
            return nil
        }
        return ErrSubscriptionNotFound
    }

消息事件触发

    func (api *PrivateAdminAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) {
        // Make sure the server is running, fail otherwise
        server := api.node.Server()
        if server == nil {
            return nil, ErrNodeStopped
        }

        // Create the subscription
        //获取notifier对象
        notifier, supported := rpc.NotifierFromContext(ctx)
        if !supported {
            return nil, rpc.ErrNotificationsUnsupported
        }
        //生成标识
        rpcSub := notifier.CreateSubscription()

        go func() {
            events := make(chan *p2p.PeerEvent)
            sub := server.SubscribeEvents(events)
            defer sub.Unsubscribe()

            for {
                select {
                case event := &lt;-events:
                    //触发事件，发送通知消息
                    notifier.Notify(rpcSub.ID, event)
                case &lt;-sub.Err():
                    return
                case &lt;-rpcSub.Err():
                    return
                case &lt;-notifier.Closed():
                    return
                }
            }
        }()

        return rpcSub, nil
    }

2.2.5 rpc client调用

以太坊提供了RPC服务，可以在geth启动时通过参数设置

2.2.5.1 geth启动选项参数

--rpc  启动HTTP-RPC服务（基于HTTP的）
--ws  启动WS-RPC服务（基于WebService的）
--rpcapi  value 指定需要调用的HTTP-RPC API接口，默认只有eth,net,web3
--rpcport value  HTTP-RPC服务器监听端口(default: 8545)
--rpcport value  HTTP-RPC服务器监听端口(default: 8545)
例子：geth --rpc --rpcapi "db,eth,net,web3,personal"

package main

import (
    "fmt"
    "github.com/ethereum/go-ethereum/mobile"
)

func main() {
    // NewEthereumClient函数只是创建一个EthereumClient结构，并设置了HTTP连接的一些参数如的head的一些属性，并没有节点建立连接
    cli, err := geth.NewEthereumClient("http://127.0.0.1:8545")
    if err != nil {
        fmt.Printf("create new ethereum rpc client err:%s\n", err.Error())
    } else {
        fmt.Println("create new ethereum rpc client success")
    }
    eth_ctx := geth.NewContext()
    block, err2 := cli.GetBlockByNumber(eth_ctx, 18)
    fmt.Printf("ethereum mobile Context:%+v\n", eth_ctx)
    if err2 != nil {
        fmt.Printf("get block err:%s\n", err2.Error())
    } else {
        fmt.Printf("block:%+v\n", block)
    }
}

连的节点是本地运行的私有链，并且在go-ethereum源码中加了一些日志，执行结果：

mylog:DialContext:u:{Scheme:http Opaque: User: Host:127.0.0.1:8545 Path: RawPath: ForceQuery:false RawQuery: Fragment:};
mylog:u.Scheme:http
create new ethereum rpc client success
mylog:JSON-RPC: Client CallContext
mylog:Client.isHTTP:true
ethereum mobile Context:&{context:0xc4200ac008 cancel:&lt;nil>}
block:Block(#18): Size: 650.00 B {
MinerHash: fd55c05ae10a5b0159b3c2d5803c6aa9469c95f5f063b9c400a2c36b49616ab3
Header(84b2cfd65e3197bdfe3f748ecebb040953af5eb73a05d8595757cf42cb40a492):
[
    ParentHash:     7892a0b31d50d67ae20d4a7ec5c24a6fe85f2f264e9f1639aa2388081305a0bd
    UncleHash:      1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347
    Coinbase:       bdc61c81f67983288a6c375a884661edc77286d0
    Root:           0f30637bfc5bd6e123c6a0c38bdc743c94050626a984f9943eaf38367100b3e3
    TxSha           354d185cfa88e50f1a425e5b89500122e4445e9ec737e7a18cdd61b9350ab72b
    ReceiptSha:     a769d28981014fb6095462148a6300cd0b43fa050d75eb6f5b7595cfd13136bb
    Bloom:          00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
    Difficulty:     131072
    Number:         18
    GasLimit:       131877941
    GasUsed:        21000
    Time:           1527044372
    Extra:          ׃��geth�go1.10�darwin
    MixDigest:      70c2bb422b1b834d5173d279e508ffee9dada454650fc3cf63e95deb3073cf32
    Nonce:          58b7495f112ccac2
]
Transactions:
[
    TX(57a3b17f84358098b728fc0f70f0697f175f8ba00d386c88eac0815b3afd6aad)
    Contract: false
    From:     2154bdd7070c99d1a25ff589a08b01dfd6eb65de
    To:       bdc61c81f67983288a6c375a884661edc77286d0
    Nonce:    0
    GasPrice: 0x430e23400
    GasLimit  0x15f90
    Value:    0xde0b6b3a7640000
    Data:     0x
    V:        0x41
    R:        0x45d4952c0190373c56e62ad15e54db54c0246385371b23c70bab4126b51927f8
    S:        0x618e4bb76a36482254352d7e5096c0dff4c1f495218d57c874fc3d8153915ea4
    Hex:      f86d80850430e2340083015f9094bdc61c81f67983288a6c375a884661edc77286d0880de0b6b3a76400008041a045d4952c0190373c56e62ad15e54db54c0246385371b23c70bab4126b51927f8a0618e4bb76a36482254352d7e5096c0dff4c1f495218d57c874fc3d8153915ea4
]
Uncles:
[]
}

2.2.5.2 分析：

go-ethereum/mobile包 是发起RPC请求的客户端直接使用的包。该包中有 EthereumClient 结构提供了Ethereum API的接入。

// EthereumClient provides access to the Ethereum APIs.
type EthereumClient struct {
    client *ethclient.Client
}

ethclient.Client在ethclient包中，包装了rpc.Client，rpc.Client代表与RPC服务的一个连接。

// Client defines typed wrappers for the Ethereum RPC API.
type Client struct {
    c *rpc.Client
}

RPC请求客户端在使用时，首先传入想要接入的节点的url作为参数，调用mobile包中的NewEthereumClient函数。创建了EthereumClient实例，并与节点建立连接。建立的RPC连接有三种形式：HTTP、WebSocket、IPC，当传入 http://127.0.0.1:8545 时，建立的是HTTP连接。

// NewEthereumClient connects a client to the given URL.
func NewEthereumClient(rawurl string) (client *EthereumClient, _ error) {
    rawClient, err := ethclient.Dial(rawurl)
    return &EthereumClient{rawClient}, err
}

设置HTTP连接的参数会调用rpc包http.go文件中的DialHTTPWithClient函数。

// DialHTTPWithClient creates a new RPC client that connects to an RPC server over HTTP
// using the provided HTTP Client.
func DialHTTPWithClient(endpoint string, client *http.Client) (*Client, error) {
    req, err := http.NewRequest(http.MethodPost, endpoint, nil)
    if err != nil {
        return nil, err
    }
    // Content-Type和Accept是application/json，即发送的数据类型和接收的数据类型都是json
    req.Header.Set("Content-Type", contentType)
    req.Header.Set("Accept", contentType)

    initctx := context.Background()
    return newClient(initctx, func(context.Context) (net.Conn, error) {
        return &httpConn{client: client, req: req, closed: make(chan struct{})}, nil
    })
}

通过HTTP来做JSON-RPC调用时，需要一个geth.Context实例，通过调用mobile包中的NewContext函数，创建一个空的geth.Context实例。

// NewContext returns a non-nil, empty Context. It is never canceled, has no
// values, and has no deadline. It is typically used by the main function,
// initialization, and tests, and as the top-level Context for incoming requests.
func NewContext() *Context {
    return &Context{
        context: context.Background(),
    }
}

mobile包中封装了请求区块、区块头、交易等函数，这些函数调用ethclient包中的相关函数，再调用更底层rpc包中封装的函数。即 mobile包-->ethclient包-->rpc包 。如mobile包中根据区块号查找区块的函数最后会调用rpc包中的CallContext函数。

// CallContext扮演JSON-RPC调用角色
// CallContext performs a JSON-RPC call with the given arguments. If the context is
// canceled before the call has successfully returned, CallContext returns immediately.
//
// The result must be a pointer so that package json can unmarshal into it. You
// can also pass nil, in which case the result is ignored.
func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
    fmt.Printf("mylog:JSON-RPC: Client CallContext\n")
    msg, err := c.newMessage(method, args...)
    if err != nil {
        return err
    }
    op := &requestOp{ids: []json.RawMessage{msg.ID}, resp: make(chan *jsonrpcMessage, 1)}

    fmt.Printf("mylog:Client.isHTTP:%+v\n",c.isHTTP)
    if c.isHTTP {
        err = c.sendHTTP(ctx, op, msg)
    } else {
        err = c.send(ctx, op, msg)
    }
    if err != nil {
        return err
    }

    // dispatch has accepted the request and will close the channel it when it quits.
    switch resp, err := op.wait(ctx); {
    case err != nil:
        return err
    case resp.Error != nil:
        return resp.Error
    case len(resp.Result) == 0:
        return ErrNoResult
    default:
        return json.Unmarshal(resp.Result, &result)
    }
}

2.2.5.3 使用POSTMAN

使用POSTMAN发送请求时，注意设置下Content-type和Accept。 body是 {"jsonrpc":"2.0","method":"web3_clientVersion","params":[],"id":67} 这种方式虽然直接，但是自己拼json会很麻烦，所以最方便的还是调用已有的接口。

如果是做查询区块号为18的区块，则body是 {"jsonrpc":"2.0","method":"eth_getBlockByNumber","params":["0x12",true],"id":1}

3. 参考

（1）以太坊源码深入分析（3）-- 以太坊RPC通信实例和原理代码分析（上） https://www.jianshu.com/p/92daf6148dc5 （2）以太坊RPC https://www.jianshu.com/p/8bd3723aa921 （3）以太坊RPC原理及实现 https://my.oschina.net/hunjixin/blog/1803161 （4）从零开始实现RPC框架 - RPC原理及实现 https://www.jianshu.com/p/dbfac2b876b1 （5）深入浅出RPC原理 https://ketao1989.github.io/2016/12/10/rpc-theory-in-action/ （6）你应该知道的RPC原理 https://www.cnblogs.com/LBSer/p/4853234.html （7）RPC原理解析 https://www.cnblogs.com/swordfall/p/8683905.html （8）服务之间的调用之RPC、Restful深入理解 https://blog.csdn.net/u014590757/article/details/80233901

学分: 7
分类: 入门/科普
标签: RPC

本文参与登链社区写作激励计划，好文好收益，欢迎正在阅读的你也加入。

【深度知识】RPC原理及以太坊RPC的实现

1.摘要

2. 内容

2.1 RPC协议和调用流程

2.1.1 远程过程调用 (RPC)

2.1.2 RPC框架

2.1.3 RPC调用流程

2.2 以太坊RPC实现

2.2.1 api发布

2.2.2 底层协议

2.2.3 rpc响应

2.2.4 Pub/sub 实现

注册或者取消注册

注册

注销

消息事件触发

2.2.5 rpc client调用

2.2.5.1 geth启动选项参数

2.2.5.2 分析：

2.2.5.3 使用POSTMAN

3. 参考

1 条评论

文章目录