gprc底子
前置环境准备
grpc下载
项目目录下执行
- go get google.golang.org/grpc@latest
复制代码 Protocol Buffers v3
https://github.com/protocolbuffers/protobuf/releases/download/v3.20.1/protoc-3.20.1-linux-x86_64.zip
go语言插件:
- go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.28
复制代码 rpc插件
- go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.2
复制代码
gRPC Hello World快速上手
基本rpc调用
服务端protoc编写
定义一个hello.proto文件
- syntax = "proto3"; // 版本声明,使用Protocol Buffers v3版本
- option go_package = "hello_server/pb"; // 指定生成的Go代码在你项目中的导入路径
- package pb; // 包名
- // 定义服务
- service Greeter {
- // SayHello 方法
- rpc SayHello (HelloRequest) returns (HelloResponse) {}
- }
- // 请求消息
- message HelloRequest {
- string name = 1;
- }
- // 响应消息
- message HelloResponse {
- string reply = 1;
- }
复制代码 服务端编写
- package main
- import (
- "context"
- "fmt"
- "hello_server/pb"
- "net"
- "google.golang.org/grpc"
- )
- // hello server
- type server struct {
- pb.UnimplementedGreeterServer
- }
- func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloResponse, error) {
- return &pb.HelloResponse{Reply: "Hello " + in.Name}, nil
- }
- func main() {
- // 监听本地的8972端口
- lis, err := net.Listen("tcp", ":8972")
- if err != nil {
- fmt.Printf("failed to listen: %v", err)
- return
- }
- s := grpc.NewServer() // 创建gRPC服务器
- pb.RegisterGreeterServer(s, &server{}) // 在gRPC服务端注册服务
- // 启动服务
- err = s.Serve(lis)
- if err != nil {
- fmt.Printf("failed to serve: %v", err)
- return
- }
- }
复制代码 客户端protoc编写
- syntax = "proto3"; // 版本声明,使用Protocol Buffers v3版本
- option go_package = "hello_client/pb"; // 指定生成的Go代码在你项目中的导入路径
- package pb; // 包名
- // 定义服务
- service Greeter {
- // SayHello 方法
- rpc SayHello (HelloRequest) returns (HelloResponse) {}
- }
- // 请求消息
- message HelloRequest {
- string name = 1;
- }
- // 响应消息
- message HelloResponse {
- string reply = 1;
- }
复制代码 客户端编写
- package main
- import (
- "context"
- "flag"
- "log"
- "time"
- "hello_client/pb"
- "google.golang.org/grpc"
- "google.golang.org/grpc/credentials/insecure"
- )
- // hello_client
- const (
- defaultName = "world"
- )
- var (
- addr = flag.String("addr", "127.0.0.1:8972", "the address to connect to")
- name = flag.String("name", defaultName, "Name to greet")
- )
- func main() {
- flag.Parse()
- // 连接到server端,此处禁用安全传输
- conn, err := grpc.NewClient(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
- if err != nil {
- log.Fatalf("did not connect: %v", err)
- }
- defer conn.Close()
- c := pb.NewGreeterClient(conn)
- // 执行RPC调用并打印收到的响应数据
- ctx, cancel := context.WithTimeout(context.Background(), time.Second)
- defer cancel()
- r, err := c.SayHello(ctx, &pb.HelloRequest{Name: *name})
- if err != nil {
- log.Fatalf("could not greet: %v", err)
- }
- log.Printf("Greeting: %s", r.GetReply())
- }
复制代码
分别在客户端和服务端执行如下步伐生成代码
- protoc --go_out=. --go_opt=paths=source_relative \
- --go-grpc_out=. --go-grpc_opt=paths=source_relative \
- pb/hello.proto
复制代码
生成服务端目录布局
生成客户端目录布局
服务端编译执行
go build编译生成hello_server
执行
客户段编译执行
如今来看看为啥客户端会打印”Hello 哈哈哈“
起首看下客户端main函数
其中有一行关键代码
- r, err := c.SayHello(ctx, &pb.HelloRequest{Name: *name})
复制代码 这里调用了SayHello函数而客户段的Sayhello函数又调用了服务端的SayHello函数
服务端SayHello函数
- func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloResponse, error) {
- return &pb.HelloResponse{Reply: "Hello " + in.Name}, nil
- }
复制代码
流式rpc调用
服务端流式调用
在原有底子上proto文件上增加如下函数
- rpc LotsOfReplies(HelloRequest) returns (stream HelloResponse);
复制代码 重新生成代码
重新go build
服务端增加LotsOfReplies实现
- // LotsOfReplies 返回使用多种语言打招呼
- func (s *server) LotsOfReplies(in *pb.HelloRequest, stream pb.Greeter_LotsOfRepliesServer) error {
- words := []string{
- "你好",
- "hello",
- "こんにちは",
- "안녕하세요",
- }
- for _, word := range words {
- data := &pb.HelloResponse{
- //循环拼接打招呼信息与客户端传过来的用户
- Reply: word + in.GetName(),
- }
- // 拼接打招呼信息使用流式的Send方法返回多个数据
- if err := stream.Send(data); err != nil {
- return err
- }
- }
- return nil
- }
复制代码 客户端增加LotsOfReplies实现
- func runLotsOfReplies(c pb.GreeterClient) {
- // server端流式RPC
- // 延长超时时间避免中断
- ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
- defer cancel()
- stream, err := c.LotsOfReplies(ctx, &pb.HelloRequest{Name: *name})
- if err != nil {
- log.Fatalf("c.LotsOfReplies failed, err: %v", err)
- }
- for {
- // 接收服务端返回的流式数据,服务端通过send发送,客户端通过Recv接受,当收到io.EOF或错误时退出
- res, err := stream.Recv()
- if err == io.EOF {
- break
- }
- if err != nil {
- log.Fatalf("c.LotsOfReplies failed, err: %v", err)
- }
- log.Printf("got reply: %q\n", res.GetReply())
- }
- }
复制代码 留意这里服务端是通过stream.Send(data)发送数据的 客户端是通过stream.Recv()接受数据的
执行
留意此时的流的流向为服务端流向客户端 所以称之为服务端流式调用
服务端实时发送数据到流中,客户端实时监听流中有无数据,当监听到没有数据了流关闭,客户端关闭
场景举例:
- 股票行情推送:客户端哀求某股票代码后,服务端连续推送实时价格颠簸数据
- 物联网
设备监控 :服务端连续推送温度传感器、GPS定位等实时收罗数据流
- 在线游戏状态同步:服务端向玩家客户端连续推送其他玩家的位置和动作数据
- 视频传播输:客户端哀求视频文件后,服务端分块传输视频流数据
- 日记文件传输:服务端将大型日记文件拆分为多个数据包流式传输
- 数据库查询结果集传输:当查询结果包含百万级记录时,服务端分批次流式返回数据
客户端流式调用
在客户端和服务端的proto文件依次增加如下步伐
- rpc LotsOfGreetings(stream HelloRequest) returns (HelloResponse);
复制代码 重新生成代码
重新go build
服务端增加LotsOfGreetings实现
- func (s *server) LotsOfGreetings(stream pb.Greeter_LotsOfGreetingsServer) error {
- reply := "你好:"
- for {
- // 接收客户端发来的流式数据
- res, err := stream.Recv()
- if err == io.EOF {
- // 最终统一回复
- return stream.SendAndClose(&pb.HelloResponse{
- Reply: reply,
- })
- }
- if err != nil {
- return err
- }
- reply += res.GetName()
- }
- }
复制代码
客户端增加LotsOfGreetings实现
- func runLotsOfGreeting(c pb.GreeterClient) {
- ctx, cancel := context.WithTimeout(context.Background(), time.Second)
- defer cancel()
- // 客户端流式RPC
- stream, err := c.LotsOfGreetings(ctx)
- if err != nil {
- log.Fatalf("c.LotsOfGreetings failed, err: %v", err)
- }
- names := []string{"风清扬,", "扫地僧,", "无嗔大师"}
- for _, name := range names {
- // 发送流式数据
- err := stream.Send(&pb.HelloRequest{Name: name})
- if err != nil {
- log.Fatalf("c.LotsOfGreetings stream.Send(%v) failed, err: %v", name, err)
- }
- }
- res, err := stream.CloseAndRecv()
- if err != nil {
- log.Fatalf("c.LotsOfGreetings failed: %v", err)
- }
- log.Printf("got reply: %v", res.GetReply())
- }
复制代码 这里的调用和服务端流式调用反过来了
流式数据由客户端进行发送多次数据stream.Send,客户端统一做接受stream.Recv()
执行
场景举例:
- 日记聚合系统:多个客户端步伐连续发送日记片段,服务端进行归并存储并返回写入状态
- 图片分块上传:移动端将大图拆分为多个数据包流式传输,服务端完成重组后返回MD5校验
- 直播推流场景:客户端分片上传视频流,服务端转码后返回转码乐成相应
双向流式调用
在客户端和服务端的proto文件中加上如下步伐
- // 双向流式数据
- rpc BidiHello(stream HelloRequest) returns (stream HelloResponse);
复制代码 重新生成代码
重新go build
服务端增加BidiHello实现
- func (s *server) BidiHello(stream pb.Greeter_BidiHelloServer) error {
- for {
- // 接收流式请求
- in, err := stream.Recv()
- if err == io.EOF {
- return nil
- }
- if err != nil {
- return err
- }
- reply := in.GetName() + "收到了你的问候,祝你生活愉快!" // 对收到的数据做些处理
- // 返回流式响应
- if err := stream.Send(&pb.HelloResponse{Reply: reply}); err != nil {
- return err
- }
- }
- }
复制代码
客户端增加BidiHello实现
- func runBidiHello(c pb.GreeterClient) {
- ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
- defer cancel()
- // 双向流模式
- stream, err := c.BidiHello(ctx)
- if err != nil {
- log.Fatalf("c.BidiHello failed, err: %v", err)
- }
- waitc := make(chan struct{})
- go func() {
- for {
- // 接收服务端返回的响应
- in, err := stream.Recv()
- if err == io.EOF {
- // read done.
- close(waitc)
- return
- }
- if err != nil {
- log.Fatalf("c.BidiHello stream.Recv() failed, err: %v", err)
- }
- fmt.Printf("回答:%s\n", in.GetReply())
- }
- }()
- // 从标准输入获取用户输入
- reader := bufio.NewReader(os.Stdin) // 从标准输入生成读对象
- for {
- cmd, _ := reader.ReadString('\n') // 读到换行
- cmd = strings.TrimSpace(cmd)
- if len(cmd) == 0 {
- continue
- }
- if strings.ToUpper(cmd) == "QUIT" {
- break
- }
- // 将获取到的数据发送至服务端
- if err := stream.Send(&pb.HelloRequest{Name: cmd}); err != nil {
- log.Fatalf("c.BidiHello stream.Send(%v) failed: %v", cmd, err)
- }
- }
- stream.CloseSend()
- <-waitc
- }
复制代码
main函数调用
执行
这里的流式数据传输是双向的
调用步骤
1,客户端创建流式调用
- stream, err := c.BidiHello(ctx)
复制代码 2,客户端发送终端输入的指令进行send发送流式数据给服务端
- if err := stream.Send(&pb.HelloRequest{Name: cmd}); err != nil {
- log.Fatalf("c.BidiHello stream.Send(%v) failed: %v", cmd, err)
- }
复制代码 3,服务端循环接受客户端的流式相应数据
4,服务端对于客户端的做加工处理和返回
- reply := in.GetName() + "收到了你的问候,祝你生活愉快!" // 对收到的数据做些处理
- // 返回流式响应
- if err := stream.Send(&pb.HelloResponse{Reply: reply}); err != nil {
- return err
- }
复制代码
gRPC联合gin开辟用户注册接口
需求:利用gin+grpc+gorm实现用户注册接口,分模块微服务设计
项目总目录
[code]douyin
├─
|