fabirc1.0商业正式版本源码解析6——GRPC服务

GRPC简介

GRPC是由自谷歌开发的一项多语言开源的RPC技术，在fabric用于实现客户端与服务器端的远程调用。比如chaincode，客户定义了一项rpc服务并相应生成了客户端代码和服务端代码，在此基础上进行业务逻辑上的开发后，分别运行服务端代码和客户端代码，实现客户端调用服务器端服务的目的。由于gprc相对来说还是很复杂的，所有还请自行学习。在gprc官网非常详细的资料和例子。

极其粗线条的写一下grpc的用法，为后文伏笔：

1. XXX.proto文件中定义一个rpc服务

service Events {
rpc Chat(stream SignedEvent) returns (stream Event) {}
}

1
2
3
1
2
3

2. 命令行使用protoc生成对应的XXX.pb.Go源码

在XXX.pb.go中，Client API for Events service处为供客户端使用的接口定义、接口实例、接口实例的初始化函数。Server
API for Events service处为供服务端使用的接口定义，注册函数。

如果其中某一端或同时两端为流式RPC，在有流的一端，会专门为其流生成接口定义、接口实例。可以直接使用生成的实例，也可以自己实现接口，自定义实例。接口定义的主要方法就是Send和Recv。

protoc --go_out=plugins=grpc:. XXX.proto

3. 编写客户端代码

//注意，由于目前我们关注的是peer node start，而其启动的基本都是后台服务端的服务，
//因此本文中不涉及客户端的代码。

//填充grpc网络链接连接选项
var opts []grpc.DialOption
opts = append(opts, grpc.WithInsecure())
//创建连接服务器端的grpc连接对象
conn, err := grpc.Dial("0.0.0.0:7051", opts...)
defer conn.Close()
//使用连接对象做参数，利用XXX.pb.go中的初始化函数创建grpc客户端对象
client := NewEventsClient(conn)
//调用服务
client.Chat(...)

1
2
3
4
5
6
7
8
9
10
11
12
13
1
2
3
4
5
6
7
8
9
10
11
12
13

4. 编写服务端代码

//定义一个监听对象，即服务器监听的地址
lis,err := net.Listen("tcp",":7051")
//创建grpc服务器选项并填充
var serverOpts []grpc.ServerOption
//创建标准的grpc服务器对象
server = grpc.NewServer(serverOpts...)
//创建服务端对象，根据XXX.pb.go中生成的接口定义，自己实现服务端接口
type eventSever{...}
func (e *eventSever)Chat(...){...}
es := new(eventServer)
//使用XXX.pb.go中的注册函数注册服务，注册到grpc服务器对象
RegisterEventsServer(server,es)
//根据监听对象启动grpc服务器对象
server.Serve(lis)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
1
2
3
4
5
6
7
8
9
10
11
12
13
14

fabric中的grpc服务接口和实例

在/fabric/core/comm/server.go中，定义了安全配置项，GPRCServer的接口、实现和初始化函数。默认情况下fabric中是不使用tls的。

TLS安全配置项

type SecureServerConfig struct {
//Whether or not to use TLS for communication
UseTLS bool
//PEM-encoded X509 public key to be used by the server for TLS communication
//在core.yaml中指定，读取的tls目录下server.cert文件数据存储于此
ServerCertificate []byte
//PEM-encoded private key to be used by the server for TLS communication
//在core.yaml中指定，读取的tls目录下server.key文件数据存储于此
ServerKey []byte
//Set of PEM-encoded X509 certificate authorities to optionally send
//as part of the server handshake
//在core.yaml中指定，读取的tls目录下ca.crt文件数据存储于此
ServerRootCAs [][]byte
//Whether or not TLS client must present certificates for authentication
RequireClientCert bool
//Set of PEM-encoded X509 certificate authorities to use when verifying
//client certificates
ClientRootCAs [][]byte
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

GRPCServer接口

type GRPCServer interface {
//返回GRPCServer监听的地址
Address() string
//启动下层的grpc.Server
Start() error
//停止下层的grpc.Server
Stop()
//返回GRPCServer实例对象
Server() *grpc.Server
//返回GRPCServer使用的网络监听实例对象
Listener() net.Listener
//返回grpc.Server使用的Certificate
ServerCertificate() tls.Certificate
//标识GRPCServer实例是否使用TLS
TLSEnabled() bool
//增加PEM-encoded X509 certificate authorities到
//用于验证客户端certificates的authorities列表
AppendClientRootCAs(clientRoots [][]byte) error
//用于验证客户端certificates的authorities列表中
//删除PEM-encoded X509 certificate authorities
RemoveClientRootCAs(clientRoots [][]byte) error
//基于一个PEM-encoded X509 certificate authorities列表
//设置用于验证客户端certificates的authorities列表
SetClientRootCAs(clientRoots [][]byte) error
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

GRPCServer实现实例

type grpcServerImpl struct {
//server指定的监听地址，地址格式：hostname:port
address string
//监听address的监听对象，用于处理网络请求
listener net.Listener
//标准的grpc服务器，通过此对象进行各种grpc服务操作
server *grpc.Server
//Certificate presented by the server for TLS communication
serverCertificate tls.Certificate
//Key used by the server for TLS communication
serverKeyPEM []byte
//List of certificate authorities to optionally pass to the client during
//the TLS handshake
serverRootCAs []tls.Certificate
//lock to protect concurrent access to append / remove
lock *sync.Mutex
//Set of PEM-encoded X509 certificate authorities used to populate
//the tlsConfig.ClientCAs indexed by subject
clientRootCAs map[string]*x509.Certificate
//TLS configuration used by the grpc server
tlsConfig *tls.Config
//Is TLS enabled?
tlsEnabled bool
}
同文件中的NewGRPCServerFromListener函数是grpcServerImpl的初始化函数，其中tls相关代码使用到了crypto下的tls、x509工具库。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

peer node start启动的grpc服务

在start.go中serve函数中，创建的GRPCServer服务对象有两个：
peerServer
globalEventsServer

peer服务器peerServer，在/fabric/core/peer/peer.go中定义。事件服务器globalEventsServer，是一个全局单例，在/fabric/events/producer/producer.go中定义。

peerServer

创建peerServer

追溯serve函数中peerServer对象的创建，代码最终都使用了/fabric/core/comm/server.go中的NewGRPCServerFromListener函数创建了一个grpcServerImpl实例对象，对象中的。

//在serve函数中
peerServer, err := peer.CreatePeerServer(listenAddr, secureConfig)

//在CreatePeerServer函数中
peerServer, err = comm.NewGRPCServer(listenAddress, secureConfig)

//在NewGRPCServer函数中
lis, err := net.Listen("tcp", address)
NewGRPCServerFromListener(lis, secureConfig)

//在NewGRPCServerFromListener函数中，最终建立grpc标准服务器并返回grpcServerImpl
grpcServer.server = grpc.NewServer(serverOpts...)
return grpcServer

1
2
3
4
5
6
7
8
9
10
11
12
13
1
2
3
4
5
6
7
8
9
10
11
12
13

注册服务

注册ChaincodeSupport服务

这是我们第一次遇到ChaincodeSupport这个对象，分为ChaincodeSupport服务原型和对应定义的ChaincodeSupport对象。从名称上就可以看出，是对fabric的chaincode提供一系列。自然，peer的grpc服务中关于chaincode的操作需要这种支持服务。而ChaincodeSupport对象本身比较复杂，在/fabric/core/chaincode/chaincode_support.go中定义，提供了一干配置值成员和chaincode的运行环境，也实现了很多接口，如该处提到的服务所需的Register函数。因在此侧重与grpc服务，因此不展开详述。

服务原型定义：

ChaincodeSupport服务原型在/fabric/protos/peer/chaincode_shim.proto中定义，相应生成chaincode_shim.pb.go源码，在此只展示其中生成的服务端定义。

//服务原型 

service ChaincodeSupport { 

rpc Register(stream ChaincodeMessage) returns (stream ChaincodeMessage) {} 

} 

//生成服务端的接口和注册函数 

type ChaincodeSupportServer interface { 

Register(ChaincodeSupport_RegisterServer) error 

} 

func RegisterChaincodeSupportServer(s *grpc.Server, srv ChaincodeSupportServer){ 

s.RegisterService(&_ChaincodeSupport_serviceDesc, srv) 

} 

//生成的服务端流的接口定义、接口实例 

type ChaincodeSupport_RegisterServer interface { 

Send(*ChaincodeMessage) error 

Recv() (*ChaincodeMessage, error) 

grpc.ServerStream 

}//接口 

type chaincodeSupportRegisterServer struct { 

grpc.ServerStream 

}//接口实口

注册服务：

在serve函数中使用

registerChaincodeSupport(peerServer.Server())

完成注册。在该函数中：

//创建了一个ChaincodeSupport对象，基本都是读取配置值填充成员 

//ChaincodeSupport对象实现了生成的服务端接口ChaincodeSupportServer中的Register方法 

ccSrv := chaincode.NewChaincodeSupport(...) 

//利用生成的注册函数，完成注册 

pb.RegisterChaincodeSupportServer(grpcServer, ccSrv)

在registerChaincodeSupport中还有一句

scc.RegisterSysCCs()

实现了系统链的注册，将在系统链主题文章中进行详述。

Register实现：

将在ChaincodeSupport主题文章中详述。

注册的其他服务

关于peerServer所注册的服务，还有AdminServer、EndorserServer、GossipService，注册的方式和注册ChaincodeSupport服务一样，毕竟都是用的gprc，还是万变不离其宗的。各个服务到底负责什么，会做什么，将会在相应主题文章中详述。

globalEventsServer

//在/fabric/events/producer/producer.go中定义
//全局单例
var globalEventsServer *EventsServer
//定义和Chat实现
type EventsServer struct {
}
func (p *EventsServer) Chat(stream pb.Events_ChatServer) error {...}
//初始化函数
func NewEventsServer(bufferSize uint, timeout int) *EventsServer {...}

1
2
3
4
5
6
7
8
9
1
2
3
4
5
6
7
8
9

事件服务器这个全局单例没有任何成员，只有一个专用初始化函数NewEventsServer，一个Chat实现。这一切看上去非常简单，只是在专用初始化函数中的一句

initializeEvents(bufferSize,
timeout)

又牵扯出了一段文字。原来globalEventsServer自己只是一个事件服务器的代表，实际做事情的是

initializeEvents(bufferSize, timeout)

初始化并运行的eventProcessor对象，下文细说。

在serve函数中，使用

ehubGrpcServer, err := createEventHubServer(secureConfig)

完成了对事件服务器的创建和注册，ehubGrpcServer承接的就是globalEventsServer这个全局单例。

创建globalEventsServer

//在createEventHubServer中
//创建grpcServerImpl对象，其中包含了grpc标准服务器
lis, err = net.Listen("tcp", viper.GetString("peer.events.address"))
grpcServer, err := comm.NewGRPCServerFromListener(lis, secureConfig)

//创建事件服务器，NewEventsServer返回的就是globalEventsServer
ehServer := producer.NewEventsServer(
uint(viper.GetInt("peer.events.buffersize")),
viper.GetInt("peer.events.timeout"))

1
2
3
4
5
6
7
8
9
1
2
3
4
5
6
7
8
9

注册事件服务

事件服务原型定义：

事件服务原型在/fabric/protos/peer/events.proto中定义，相应生成events.pb.go源码，在此只展示其中生成的服务端定义。

//服务原型 

service Events { 

rpc Chat(stream SignedEvent) returns (stream Event) {} 

} 

//生成服务端的接口和注册函数 

type EventsServer interface { 

Chat(Events_ChatServer) error 

} 

func RegisterEventsServer(s *grpc.Server, srv EventsServer) { 

s.RegisterService(&_Events_serviceDesc, srv) 

} 

//生成的服务端流的接口定义、接口实例 

type Events_ChatServer interface { 

Send(*Event) error 

Recv() (*SignedEvent, error) 

grpc.ServerStream 

}//接口 

type eventsChatServer struct { 

grpc.ServerStream 

}//接口实例

注册服务：

//还是在createEventHubServer中 

//ehServer对象实现了生成的服务端接口EventsServer的Chat方法 

//利用生成的注册函数，完成注册 

pb.RegisterEventsServer(grpcServer.Server(), ehServer)

Chat实现：

Chat的操作很清晰，循环的接收数据然后处理数据，即处理客户端的Chat调用，这也自然而然的是grpc服务端所要做的。

handler, err := newEventHandler(stream)

，根据服务端流接口stream创建一个handler，用于处理接收的客户端发送的SignedEvent类型数据。handler于后文详述。

in, err := stream.Recv()

，接收SignedEvent类型的数据，这也是gprc双向流的标准用法。

err = handler.HandleMessage(in)

，使用handler处理数据，HandleMessage是实际的数据处理函数。

在HandleMessage函数中，客户端发送签名过的SignedEvent类型数据，检查有效性后，若是注册或注销事件，则注册或注销，并返回Event类型数据；若是其他类型的事件，则打印一条错误消息后返回。

evt, err := validateEventMessage(msg)

，利用local MSP验证数据的有效性。关于local MSP将在对应主题文章中详述。

switch evt.Event.(type) {...}

，判断事件类型，并对注册事件或注销事件进行注册或注销。

if err := d.ChatStream.Send(evt);err != nil{...}

，若是注册事件或注销事件，执行相应操作之后返回Event数据给客户端，该数据是在验证函数validateEventMessage中获取的。

启动事件服务

在serve函数中靠后的地方，

if ehubGrpcServer != nil {go ehubGrpcServer.Start()}

将该服务的gprc服务端启动起来了。Start内部调用了grpc服务器启动的标准函数

server.Serve(lis)

。

事件实际处理者eventProcessor

事件处理者也是一个全局单例，接收不同类型的事件进行处理。在/fabric/events/events.go中定义，结构如下：



在eventProcessor的成员中：

eventConsumers

按照事件类型分类的事件处理链条，处理链handlerList接口有两种具体实现：一般处理链genericHandlerList和chaincode专用处理链chaincodeHandlerList。在此以一般处理链为例，其实现了对handlers三个操作：add，del，foreach。其中遍历操作foreach则对handlers中的每个handler执行了由参数指定的动作。这里的handlers映射了handler与bool值，bool值应该是起到类似于开关的作用。

handler在/fabric/events/handle.go中定义，其成员ChatStream是一个events.pb.go中生成的Events_ChatServer类型的gprc流接口，用于发送流数据。handler挂载了一系列操作函数，如register，HandleMessage，SendMessage，Stop。

eventChannel

带缓存且专门处理Event类型数据的事件频道，所有的事件都是通过此频道分发出去的。缓存大小由core.yaml定义为100，由initializeEvents的参数带进来并设置。Event类型在由events.proto中定义并对应生成events.pb.go中定义。

timeout

频道eventChannel若满时等待的时间，在core.yaml中设置并有释义。

我们将从其初始化函数，也就是上文提到的initializeEvents入手，分析事件处理者eventProcessor。

事件类型

initializeEvents的前两句很容易理解，

if gEventProcessor != nil{...}

只为保证gEventProcessor的单例性质，

gEventProcessor
= &eventProcessor{...}

则为gEventProcessor创建了对象实例，分配了内存空间。而

addInternalEventTypes()

，实质4次调用了

AddEventType

，则为添加内部事件类型并相应的分配了这些类型各自的处理链handlerList。

添加的已知的事件类型由/fabric/protos/peer/events.proto中定义，对应生成的events.pb.go中的四种：
EventType_BLOCK - 块事件，对应genericHandlerList
EventType_CHAINCODE - chaincode事件，对应chaincodeHandlerList
EventType_REGISTER - addInternalEventTypes中有但是AddEventType未做处理
EventType_REJECTION - 拒绝事件，对应genericHandlerList

start函数

initializeEvents最后一句就是

go gEventProcessor.start()

，就是另起一个goroutine运行全局单例gEventProcessor的start函数。start函数是一个死循环，不断从eventChannel中接收数据Event类型数据并处理。过程如下：

e := <-ep.eventChannel

，获取一个事件e。

eType := getMessageType(e)

，获取事件e的类型eType。

if hl, _ = ep.eventConsumers[eType]; hl == nil {...}

，根据eType获取该事件类型的处理链hl，同时判断该类型是否存在，若不存在则会被忽略本次事件而continue继续处理下一个事件。

hl.foreach(...)

，调用hl的foreach函数，foreach遍历了hl.handlers中的每个handler，并对每个handler执行第二个参数指定的动作。该动作为

func(h
*handler){if e.Event != nil{h.SendMessage(e)}}

，即调用每个handler的SendMessage发送事件e。SendMessage则是使用handler中自有的ChatStream这个生成的grpc流服务接口去发送事件e：

err
:= d.ChatStream.Send(msg)

。

流程图如下：