1. Unary RPC
proto文件如下
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
syntax = "proto3"; option go_package=".;service";
message HelloRequest { // Name of the person to greet string name = 1; }
message HelloResponse { // Greeting message string greeting = 1; }
service HelloService { // RPC method to say hello rpc SayHello (HelloRequest) returns (HelloResponse){} } |
使用命令(注意命令路径和自己的对应):
|
1 |
protoc -I . --go-grpc_out=require_unimplemented_servers=false:. --go_out=. *.proto |
对应目录上有xx.pb.go和xx_grpc.pb.go然后对应目录实现服务端接口
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
package main
import ( "context" "fmt" "google.golang.org/grpc" "net" "test_grpc/service" )
type HelloService struct { }
func (hs *HelloService) SayHello(ctx context.Context, req *service.HelloRequest) (*service.HelloResponse, error) { resp := &service.HelloResponse{ Greeting: fmt.Sprintf("hello %s --from Golang Server", req.Name), } return resp, nil }
func main() { // listen on 127.0.0.1:50051 listen, err := net.Listen("tcp", "127.0.0.1:50051") if err != nil { fmt.Println("Error happened when listen on 127.0.0.1:50051:", err) return }
// grpc server s := grpc.NewServer()
// register HelloService in grpc server service.RegisterHelloServiceServer(s, &HelloService{})
// start rpc server fmt.Println("Golang rpc server is waiting messages......") if err = s.Serve(listen); err != nil { fmt.Println("Error happened when start rpc server:", err) return } } |
客户端接口如下：
|
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 26 27 28 29 30 31 32 33 34 35 36 |
package main
import ( "context" "fmt" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "test_grpc/service" "time" )
func main() { // connect to server conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { fmt.Println("Connect to rpc server err:", err) return } defer conn.Close()
// init service client c := service.NewHelloServiceClient(conn)
// init context with timeout ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel()
// send message req := &service.HelloRequest{Name: "Golang"} r, err := c.SayHello(ctx, req) if err != nil { fmt.Println("Send message err:", err) return } fmt.Println("Client:", r.Greeting) } |
实际上为了更好得感受gRPC这种跨语言调用的感觉，可以尝试使用python编写client端代码，直接复制proto文件，在python中使用以下命令生成对应的proto文件（注意命令和自己的对应）：
|
1 |
python -m grpc_tools.protoc -I . --python_out=. --pyi_out=. --grpc_python_out=. *.proto |
使用python实现的客户端代码如下：
|
1 2 3 4 5 6 7 8 9 10 |
# client template # grpc server address channel = grpc.insecure_channel("127.0.0.1:50051") stub = hello_pb2_grpc.HelloServiceStub(channel)
# send request response = stub.SayHello(hello_pb2.HelloRequest(name="Python"))
print(response.greeting) # hello Python --from Golang Server |
2. Server-side streaming RPC
重新给出这个的proto文件，服务端将以流式数据的形式发送给客户端数据
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
syntax = "proto3"; option go_package=".;service";
message HelloRequest { // Name of the person to greet string name = 1; }
message HelloResponse { // Greeting message string greeting = 1; }
service HelloService { // RPC method to say hello rpc SayHello (HelloRequest) returns (stream HelloResponse){} } |
同理，生成对应的proto文件后，在对应的文件先生成server端的代码：
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 |
package main
import ( "fmt" "google.golang.org/grpc" "net" "test_grpc/service" )
type HelloService struct { }
func (hs *HelloService) SayHello(req *service.HelloRequest, stream service.HelloService_SayHelloServer) error { resp := &service.HelloResponse{ Greeting: fmt.Sprintf("hello %s --from Golang Server", req.Name), } // 连续发送5次 for i := 0; i < 5; i++ { if err := stream.Send(resp); err != nil { return err } } return nil }
func main() { // listen on 127.0.0.1:50051 listen, err := net.Listen("tcp", "127.0.0.1:50051") if err != nil { fmt.Println("Error happened when listen on 127.0.0.1:50051:", err) return }
// grpc server s := grpc.NewServer()
// register HelloService in grpc server service.RegisterHelloServiceServer(s, &HelloService{})
// start rpc server fmt.Println("Golang rpc server is waiting messages......") if err = s.Serve(listen); err != nil { fmt.Println("Error happened when start rpc server:", err) return } } |
同理给出客户端的代码：
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
package main
import ( "context" "fmt" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "io" "log" "test_grpc/service" "time" )
func main() { // connect to server conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { fmt.Println("Connect to rpc server err:", err) return } defer conn.Close()
// init service client c := service.NewHelloServiceClient(conn)
// init context with timeout ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel()
// send message req := &service.HelloRequest{Name: "Golang"} stream, err := c.SayHello(ctx, req) if err != nil { fmt.Println("Send message err:", err) return }
// 加载消息 for { resp, err := stream.Recv() // 读到结束标志 if err == io.EOF { log.Fatalf("end.....") break }
if err != nil { log.Fatalf("failed to receive response: %v", err) }
log.Printf("Greeting: %s", resp.Greeting) } } |
3. Client-side streaming RPC
对应的proto文件如下
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
syntax = "proto3"; option go_package=".;service";
message HelloRequest { // Name of the person to greet string name = 1; }
message HelloResponse { // Greeting message string greeting = 1; }
service HelloService { // RPC method to say hello rpc SayHello (stream HelloRequest) returns (HelloResponse){} } |
同理使用protoc命令生成对应的proto文件，后先编写client端的代码，如下：
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
package main
import ( "context" "fmt" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "log" "test_grpc/service" "time" )
func main() { // connect to server conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { fmt.Println("Connect to rpc server err:", err) return } defer conn.Close()
// init service client c := service.NewHelloServiceClient(conn)
// init context with timeout ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel()
// create stream
stream, err := c.SayHello(ctx) if err != nil { log.Fatalf("could not greet: %v", err) }
names := []string{"World", "Gophers", "Anthropic"}
for _, name := range names { // request body req := &service.HelloRequest{Name: name} if err := stream.Send(req); err != nil { log.Fatalf("faild to send request: %v", err) } }
resp, err := stream.CloseAndRecv() if err != nil { log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil) } log.Printf("Greeting: %s", resp.Greeting) } |
对应得完成服务端的代码：
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
package main
import ( "fmt" "google.golang.org/grpc" "io" "net" "strings" "test_grpc/service" )
type HelloService struct { }
func (hs *HelloService) SayHello(stream service.HelloService_SayHelloServer) error { var strs []string for { msg, err := stream.Recv() if err == io.EOF { break } if err != nil { return err }
strs = append(strs, msg.Name) }
resp := &service.HelloResponse{Greeting: strings.Join(strs, " ")}
err := stream.SendAndClose(resp) if err != nil { return err } return nil }
func main() { // listen on 127.0.0.1:50051 listen, err := net.Listen("tcp", "127.0.0.1:50051") if err != nil { fmt.Println("Error happened when listen on 127.0.0.1:50051:", err) return }
// grpc server s := grpc.NewServer()
// register HelloService in grpc server service.RegisterHelloServiceServer(s, &HelloService{})
// start rpc server fmt.Println("Golang rpc server is waiting messages......") if err = s.Serve(listen); err != nil { fmt.Println("Error happened when start rpc server:", err) return } } |
4. Bidirectional streaming RPC
新的proto文件被如下给出：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
syntax = "proto3"; option go_package=".;service";
message HelloRequest { // Name of the person to greet string name = 1; }
message HelloResponse { // Greeting message string greeting = 1; }
service HelloService { // RPC method to say hello rpc SayHello (stream HelloRequest) returns (stream HelloResponse){} } |
和上文中的操作一致，同时给出server端的代码：
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 |
package main
import ( "fmt" "google.golang.org/grpc" "io" "log" "net" "test_grpc/service" )
type HelloService struct { }
func (hs *HelloService) SayHello(stream service.HelloService_SayHelloServer) error { for { msg, err := stream.Recv() if err == io.EOF { break } if err != nil { return err }
name := msg.Name resp := &service.HelloResponse{Greeting: name}
if err = stream.Send(resp); err != nil { log.Fatalf("Failed to send a resp:%s", err) } }
return nil }
func main() { // listen on 127.0.0.1:50051 listen, err := net.Listen("tcp", "127.0.0.1:50051") if err != nil { fmt.Println("Error happened when listen on 127.0.0.1:50051:", err) return }
// grpc server s := grpc.NewServer()
// register HelloService in grpc server service.RegisterHelloServiceServer(s, &HelloService{})
// start rpc server fmt.Println("Golang rpc server is waiting messages......") if err = s.Serve(listen); err != nil { fmt.Println("Error happened when start rpc server:", err) return } } |
同时给出下面的client端的代码：
|
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 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 |
package main
import ( "context" "fmt" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "io" "log" "test_grpc/service" "time" )
func main() { // connect to server conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { fmt.Println("Connect to rpc server err:", err) return } defer conn.Close()
// init service client c := service.NewHelloServiceClient(conn)
// init context with timeout ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel()
// create stream stream, err := c.SayHello(ctx) if err != nil { log.Fatalf("could not greet: %v", err) }
names := []string{"World", "Gophers", "Anthropic"}
waitc := make(chan struct{}) go func() { for { resp, err := stream.Recv() if err == io.EOF { close(waitc) return } if err != nil { log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil) } log.Printf("Greeting: %s", resp.Greeting) } }()
go func() { for _, name := range names { // request body req := &service.HelloRequest{Name: name} if err := stream.Send(req); err != nil { log.Fatalf("faild to send request: %v", err) }
// send delay time.Sleep(1) } // 发送结束的消息 if err := stream.CloseSend(); err != nil { log.Fatalf("failed to close stream: %v", err) } }()
<-waitc } |
一定要注意关闭发送或者避免针对一个已经关闭stream进行发送消息，读取消息是被允许的，这里有一点类似chan
4. ALTS
4.1 ALTS的介绍
应用层传输安全（ALTS）是谷歌开发的一种相互验证和传输加密系统。它用于确保谷歌基础设施内 RPC 通信的安全。ALTS 类似于相互 TLS，但经过设计和优化，可满足 Google 生产环境的需要。ALTS在gRPC中有以下的特征：
- 使用ALTS作为传输协议创建gRPC的服务端和客户端；
- ALSTS是一个端到端的保护，具有隐私性和完成性；
- 应用可以访问对等信息比如对等服务账户；
- 支持客户端和服务端的认知；
- 最小的代码更改就能使用ALTS；
值得注意的是ALTS被全部发挥作用如果应用程序运行在CE或者GKE中
4.2 gRPC客户端使用ALTS传输安全协议
gRPC客户端使用ALTS认证去连接服务端，正如下面代码中所描述的：
|
1 2 3 4 5 6 7 8 |
import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/alts" )
altsTC := alts.NewClientCreds(alts.DefaultClientOptions()) // connect to server conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithTransportCredentials(altsTC)) |
gRPC服务端能够使用ALTS认证来运行客户端连接到它，正如下面的描述：
|
1 2 3 4 5 6 7 |
import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/alts" )
altsTC := alts.NewServerCreds(alts.DefaultServerOptions()) server := grpc.NewServer(grpc.Creds(altsTC)) |
4.3 Server Authorization
gRPC 内置了使用 ALTS 的服务器授权支持。使用 ALTS 的 gRPC 客户端可以在建立连接前设置预期的服务器服务账户。然后，在握手结束时，服务器授权会保证服务器身份与客户端指定的服务账户之一相匹配。否则，连接将失败。
|
1 2 3 4 5 6 7 8 9 |
import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/alts" )
clientOpts := alts.DefaultClientOptions() clientOpts.TargetServiceAccounts = []string{expectedServerSA} altsTC := alts.NewClientCreds(clientOpts) conn, err := grpc.Dial(serverAddr, grpc.WithTransportCredentials(altsTC)) |
