mirror of
https://github.com/chai2010/advanced-go-programming-book.git
synced 2025-05-24 20:47:11 +00:00
ch4-04: 细节完善
This commit is contained in:
parent
bd6780e26c
commit
dbd7962044
@ -4,7 +4,7 @@ GRPC是Google公司基于Protobuf开发的跨语言的开源RPC框架。GRPC基
|
||||
|
||||
## GRPC入门
|
||||
|
||||
如果从Protobuf的角度看,GRPC只不过是一个针对service接口生成代码的生成器。我们在本章的第二节中手工实现了一个简单的Protobuf代码生成器插件,只不过当时生成的代码是适配标准库的RPC框架的。
|
||||
如果从Protobuf的角度看,GRPC只不过是一个针对service接口生成代码的生成器。我们在本章的第二节中手工实现了一个简单的Protobuf代码生成器插件,只不过当时生成的代码是适配标准库的RPC框架的。现在我们将学习GRPC的用法。
|
||||
|
||||
创建hello.proto文件,定义HelloService接口:
|
||||
|
||||
@ -36,7 +36,7 @@ type HelloServiceServer interface {
|
||||
}
|
||||
|
||||
type HelloServiceClient interface {
|
||||
Hello(ctx context.Context, in *String, opts ...grpc.CallOption) (*String, error)
|
||||
Hello(context.Context, *String, ...grpc.CallOption) (*String, error)
|
||||
}
|
||||
```
|
||||
|
||||
@ -47,7 +47,9 @@ GRPC通过context.Context参数,为每个方法调用提供了上下文支持
|
||||
```go
|
||||
type HelloServiceImpl struct{}
|
||||
|
||||
func (p *HelloServiceImpl) Hello(ctx context.Context, args *String) (*String, error) {
|
||||
func (p *HelloServiceImpl) Hello(
|
||||
ctx context.Context, args *String,
|
||||
) (*String, error) {
|
||||
reply := &String{Value: "hello:" + args.GetValue()}
|
||||
return reply, nil
|
||||
}
|
||||
@ -58,7 +60,7 @@ GRPC服务的启动流程和标准库的RPC服务启动流程类似:
|
||||
```go
|
||||
func main() {
|
||||
grpcServer := grpc.NewServer()
|
||||
RegisterHelloServiceServer(grpcServer, &HelloServiceImpl{})
|
||||
RegisterHelloServiceServer(grpcServer, new(HelloServiceImpl))
|
||||
|
||||
lis, err := net.Listen("tcp", ":1234")
|
||||
if err != nil {
|
||||
@ -91,11 +93,11 @@ func main() {
|
||||
|
||||
其中grpc.Dial负责和GRPC服务建立链接,然后NewHelloServiceClient函数基于已经建立的链接构造HelloServiceClient对象。返回的client其实是一个HelloServiceClient接口对象,通过接口定义的方法就可以调用服务端对应的GRPC服务提供的方法。
|
||||
|
||||
GRPC和标准库的RPC框架还有一个区别,GRPC生成的接口并不支持异步调用。
|
||||
GRPC和标准库的RPC框架有一个区别,GRPC生成的接口并不支持异步调用。不过我们可以在多个Goroutine之间安全地共享GRPC底层的HTTP/2链接,因此可以通过在另一个Goroutine阻塞调用的方式模拟异步调用。
|
||||
|
||||
## GRPC流
|
||||
|
||||
RPC是远程函数调用,因此每次调用的函数参数和返回值不能太大,否则将严重影响每次调用的性能。因此传统的RPC方法调用对于上传和下载较大数据量场景并不适合。同时传统RPC模式也不适用于对时间不确定的订阅和发布模式。为此,GRPC框架分别提供了服务器端和客户端的流特性。
|
||||
RPC是远程函数调用,因此每次调用的函数参数和返回值不能太大,否则将严重影响每次调用的响应时间。因此传统的RPC方法调用对于上传和下载较大数据量场景并不适合。同时传统RPC模式也不适用于对时间不确定的订阅和发布模式。为此,GRPC框架针对服务器端和客户端分别提供了流特性。
|
||||
|
||||
服务端或客户端的单向流是双向流的特例,我们在HelloService增加一个支持双向流的Channel方法:
|
||||
|
||||
@ -165,7 +167,7 @@ func (p *HelloServiceImpl) Channel(stream HelloService_ChannelServer) error {
|
||||
}
|
||||
```
|
||||
|
||||
服务端在循环中接收客户端发来的数据,如果遇到io.EOF表示客户端流被关闭,如果函数退出表示服务端流关闭。然后生成返回的数据通过流发送给客户端。需要注意的是,发送和接收的操作并不需要一一对应,用户可以根据真实场景进行组织代码。
|
||||
服务端在循环中接收客户端发来的数据,如果遇到io.EOF表示客户端流被关闭,如果函数退出表示服务端流关闭。生成返回的数据通过流发送给客户端,双向流数据的发送和接收都是完全独立的行为。需要注意的是,发送和接收的操作并不需要一一对应,用户可以根据真实场景进行组织代码。
|
||||
|
||||
客户端需要先调用Channel方法获取返回的流对象:
|
||||
|
||||
@ -272,8 +274,10 @@ type PubsubServiceServer interface {
|
||||
Subscribe(*String, PubsubService_SubscribeServer) error
|
||||
}
|
||||
type PubsubServiceClient interface {
|
||||
Publish(ctx context.Context, in *String, opts ...grpc.CallOption) (*String, error)
|
||||
Subscribe(ctx context.Context, in *String, opts ...grpc.CallOption) (PubsubService_SubscribeClient, error)
|
||||
Publish(context.Context, *String, ...grpc.CallOption) (*String, error)
|
||||
Subscribe(context.Context, *String, ...grpc.CallOption) (
|
||||
PubsubService_SubscribeClient, error,
|
||||
)
|
||||
}
|
||||
|
||||
type HelloService_SubscribeServer interface {
|
||||
@ -282,6 +286,8 @@ type HelloService_SubscribeServer interface {
|
||||
}
|
||||
```
|
||||
|
||||
因为SubscribeTopic是服务端的单向流,因此生成的HelloService_SubscribeServer接口中只有Send方法。
|
||||
|
||||
然后就可以实现发布和订阅服务了:
|
||||
|
||||
```go
|
||||
@ -299,12 +305,16 @@ func NewPubsubService() *PubsubService {
|
||||
然后是实现发布方法和订阅方法:
|
||||
|
||||
```go
|
||||
func (p *PubsubService) Publish(ctx context.Context, arg *String) (*String, error) {
|
||||
func (p *PubsubService) Publish(
|
||||
ctx context.Context, arg *String,
|
||||
) (*String, error) {
|
||||
p.pub.Publish(arg.GetValue())
|
||||
return &String{}, nil
|
||||
}
|
||||
|
||||
func (p *PubsubService) Subscribe(arg *String, stream PubsubService_SubscribeServer) error {
|
||||
func (p *PubsubService) Subscribe(
|
||||
arg *String, stream PubsubService_SubscribeServer,
|
||||
) error {
|
||||
ch := p.SubscribeTopic(func(v interface{}) bool {
|
||||
if key, ok := v.(string); ok {
|
||||
if strings.Hasprefix(arg.GetValue()) {
|
||||
@ -347,7 +357,7 @@ func main() {
|
||||
}
|
||||
```
|
||||
|
||||
然后就可以在新的客户端进行订阅信息了:
|
||||
然后就可以在另一个客户端进行订阅信息了:
|
||||
|
||||
```go
|
||||
func main() {
|
||||
|
Loading…
x
Reference in New Issue
Block a user