gRPC学习笔记记录以及整合gin开发

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 mainimport ("context""fmt""hello_server/pb""net""google.golang.org/grpc"
)// hello servertype 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 mainimport ("context""flag""log""time""hello_client/pb""google.golang.org/grpc""google.golang.org/grpc/credentials/insecure"
)// hello_clientconst (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()// 客户端流式RPCstream, 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函数调用

	runBidiHello(c)

执行

这里的流式数据传输是双向的

调用步骤

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，服务端循环接受客户端的流式响应数据

in, err := stream.Recv()

4，服务端对于客户端的做加工处理和返回

reply := in.GetName() + "收到了你的问候，祝你生活愉快！" // 对收到的数据做些处理// 返回流式响应if err := stream.Send(&pb.HelloResponse{Reply: reply}); err != nil {return err}

gRPC结合gin开发用户注册接口

需求：使用gin+grpc+gorm实现用户注册接口，分模块微服务设计

项目总目录

douyin
├─ 📁gateway //网关层
│  ├─ 📁cmd
│  │  └─ 📄main.go   //网关主入口
│  ├─ 📁internal
│  │  ├─ 📁controller //网关控制器
│  │  │  └─ 📄user.go
│  │  ├─ 📁grpc_client //grpc客户端
│  │  │  └─ 📄grpc_client.go
│  │  └─ 📁router
│  │     └─ 📄router.go //路由
│  ├─ 📄go.mod
│  └─ 📄go.sum
├─ 📁user_service   //user_service微服务模块
│  ├─ 📁internal
│  │  ├─ 📁config    //配置类如sql redis等
│  │  │  ├─ 📄application.yaml
│  │  │  └─ 📄config.go
│  │  ├─ 📁handler   //grpc处理
│  │  │  └─ 📄user.go
│  │  ├─ 📁repository  //db repository
│  │  │  └─ 📄user.go
│  │  ├─ 📁service  //service层
│  │  │  └─ 📄user.go
│  │  └─ 📁store    // db层
│  │     └─ 📁model
│  │        ├─ 📄model.go
│  │        └─ 📄user.go
│  ├─ 📁proto  //grpc自动生成proto
│  │  ├─ 📁pb
│  │  │  └─ 📄user.proto
│  │  ├─ 📄user.pb.go
│  │  └─ 📄user_grpc.pb.go
│  ├─ 📄go.mod
│  ├─ 📄go.sum
│  └─ 📄main.go //grpc启动主入口
├─ 📄go.mod
└─ 📄go.sum

首先执行go mod init 项目名称

导入go.mod依赖

网关层

在gateway项目下执行

go get github.com/gin-gonic/gin

定义网关控制器函数internal/controller/user.go

package controllerimport ("gateway/internal/grpc_client""net/http""github.com/gin-gonic/gin"
)type UserController struct {userClient *grpc_client.UserClient
}//提供给grpc调用客户端
func NewUserController(client *grpc_client.UserClient) *UserController {return &UserController{userClient: client}
}func (c *UserController) Register(ctx *gin.Context) {type Request struct {Username string `json:"username" binding:"required,min=4,max=20"`Password string `json:"password" binding:"required,min=6,max=20"`Phone    string `json:"phone" binding:"required"`}var req Request//使用json格式传输数据if err := ctx.ShouldBindJSON(&req); err != nil {ctx.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})return}// 调用gRPC服务resp, err := c.userClient.Register(ctx, req.Username, req.Password, req.Phone)if err != nil {ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})return}ctx.JSON(http.StatusOK, gin.H{"user_id": resp,})
}

定义grpc客户端

package grpc_clientimport ("context""user_service/proto" // 确保proto生成的代码路径正确"google.golang.org/grpc""google.golang.org/grpc/credentials/insecure"
)type UserClient struct {conn   *grpc.ClientConnclient proto.UserServiceClient // 假设proto生成的接口为UserServiceClient
}func NewUserClient(serverAddr string) (*UserClient, error) {// 建立gRPC连接conn, err := grpc.Dial(serverAddr,grpc.WithTransportCredentials(insecure.NewCredentials()), // 禁用TLSgrpc.WithBlock(), // 阻塞直到连接成功)if err != nil {return nil, err}return &UserClient{conn:   conn,client: proto.NewUserServiceClient(conn),}, nil
}// Register 调用gRPC服务注册方法
func (c *UserClient) Register(ctx context.Context,username, password, phone string,
) (uint32, error) {resp, err := c.client.Register(ctx, &proto.UserRequest{Username: username,Password: password,Phone:    phone,})if err != nil {return 0, err}return resp.UserId, nil
}// Close 关闭连接
func (c *UserClient) Close() error {return c.conn.Close()
}

网关层执行主函数cmd/main.go

package mainimport (// 控制器所在路径"gateway/internal/controller""gateway/internal/grpc_client""log""os""os/signal""syscall""github.com/gin-gonic/gin"
)func main() {// 1. 初始化gRPC客户端userClient, err := grpc_client.NewUserClient("localhost:50051") // 替换为实际地址if err != nil {log.Fatalf("Failed to create gRPC client: %v", err)} else {log.Println("gRPC client created successfully")}defer userClient.Close()// 2. 创建控制器userController := controller.NewUserController(userClient)// 3. 配置Gin路由router := gin.Default()api := router.Group("/api/v1"){api.POST("/register", userController.Register)}// 4. 启动HTTP服务器router.Run("127.0.0.1:8089") // listen and serve on 0.0.0.0:8080// go func() {// 	if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {// 		log.Fatalf("Server error: %v", err)// 	}// }()// 5. 优雅关闭quit := make(chan os.Signal, 1)signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)<-quitlog.Println("Shutting down server...")
}

注意gateway网关和user_service为两个独立的服务模块，而且gateway依赖user_service，需在网关层的go.mod文件显示声明两者的依赖关系

require user_service v0.0.0 // 声明依赖replace user_service => ../user_service // 替换为本地路径（假设 user_service 和 gateway 是同级目录）

下载文件所需依赖

go mod tidy

grpc微服务用户模块user_service

config配置类

用于定义服务所需的各个中间件服务和依赖的配置等，可以类比成Java的springboot的application.yml文件

定义application.yml

#项目相关配置
app:name: douyindebug: true
#数据库相关配置
database:driver: mysqlhost: 127.0.0.1port: 3306username: rootdbname:库名password:密码
#redis相关配置
redis:host: 127.0.0.1port: 6379

定义config配置加载函数 config.go

这里采用go最佳配置工具 github.com/spf13/viper

package configimport ("fmt""github.com/fsnotify/fsnotify""github.com/spf13/viper"
)// InitConfig 初始化配置文件
func InitConfig() {viper.AddConfigPath("./internal/config")viper.AddConfigPath(".") //多路径查找viper.SetConfigName("application")viper.SetConfigType("yaml")if err := viper.ReadInConfig(); err != nil {fmt.Printf("Failed to read config file: %v\nSearch paths: %v\n",err, viper.ConfigFileUsed())panic(err)}//监控并重新读取配置文件viper.WatchConfig()viper.OnConfigChange(func(e fsnotify.Event) {// 配置文件发生变更之后会调用的回调函数fmt.Println("Config file changed:", e.Name)})
}

核心处理函数internal/handler/user.go

这里相当于mvc模式三层架构的控制器层，整体grpc的三层调用顺序为handler->service->repositry

package handlerimport ("context""user_service/internal/service""user_service/proto""google.golang.org/grpc/codes""google.golang.org/grpc/status"
)type UserHandler struct {service *service.UserServiceproto.UnimplementedUserServiceServer
}func NewUserHandler(service *service.UserService) *UserHandler {return &UserHandler{service: service}
}func (h *UserHandler) Register(ctx context.Context, req *proto.UserRequest) (*proto.Response, error) {res, err := h.service.Register(ctx, req)if err != nil {if isDuplicateError(err) {return nil, status.Error(codes.AlreadyExists, "用户已存在")}return nil, status.Error(codes.Internal, err.Error())}return &proto.Response{Code:    200,UserId:  uint32(res.UserId),Message: "注册成功",}, nil
}func isDuplicateError(err error) bool {// 根据具体数据库错误判断return true
}

service服务 service/user.go

package serviceimport ("context"repository "user_service/internal/repostry""user_service/internal/store/model""user_service/proto"
)type UserService struct {proto.UnimplementedUserServiceServer // 关键！必须嵌入repo *repository.UserRepository
}func NewUserService(repo *repository.UserRepository) *UserService {return &UserService{repo: repo}
}func (s *UserService) Register(ctx context.Context,req *proto.UserRequest,
) (*proto.Response, error) {// 实现具体注册逻辑// 例如：user, err := s.repo.Create(&model.User{Username: req.Username,Password: req.Password,Phone:    req.Phone,})if err != nil {return nil, err}return &proto.Response{UserId: user.Id}, nil// 示例返回（替换为实际逻辑）// return &proto.Response{UserId: 1}, nil
}

repositry层repositry/user.go

package repositoryimport ("log""user_service/internal/store/model""github.com/jinzhu/gorm"
)type UserRepository struct {db *gorm.DB
}func NewUserRepository(db *gorm.DB) *UserRepository {return &UserRepository{db: db}
}func (r *UserRepository) Create(user *model.User) (*model.User, error) {if err := r.db.Create(user).Error; err != nil {return nil, err}log.Printf("Created user ID: %d", user.Id) // 插入后打印 IDreturn user, nil
}

数据库驱动加载 store/model/model.go

package modelimport ("fmt""log"_ "github.com/go-sql-driver/mysql""github.com/jinzhu/gorm""github.com/spf13/viper"
)// Db 用来承接db变量
var Db *gorm.DB// InitDb 初始化数据库连接
func InitDb() *gorm.DB {var (Username = viper.GetString("database.username")Password = viper.GetString("database.password")Host     = viper.GetString("database.host")Port     = viper.GetInt("database.port")DbName   = viper.GetString("database.dbname"))dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8&parseTime=True&loc=Local", Username, Password, Host, Port, DbName)fmt.Println(Username)db, err := gorm.Open("mysql", dsn)if err != nil {log.Fatal("数据库连接失败,报错信息" + err.Error())}// 设置连接池，空闲连接db.DB().SetMaxIdleConns(50)// 打开链接db.DB().SetMaxOpenConns(100)// 表明禁用后缀加sdb.SingularTable(true)// 启用Logger，显示详细日志db.LogMode(viper.GetBool("app.debug"))return db
}

数据库映射实体类store/model/user.go

package modelimport ("time""gorm.io/gorm"
)type User struct {Id        uint32 `gorm:"primaryKey;autoIncrement"`Username  string `gorm:"uniqueIndex;size:64;not null"`Password  string `gorm:"size:128;not null"`Phone     string `gorm:"size:128;not null"`CreatedAt time.TimeUpdatedAt time.Time
}func (User) TableName() string {return "user_db"
}// Message模型（shared/model/message.go）
type Message struct {gorm.ModelContent    string `gorm:"type:text;not null"`SenderID   uint   `gorm:"index:idx_sender"`ReceiverID uint   `gorm:"index:idx_receiver"`
}

gprc proto文件自动生成代码定义

syntax = "proto3";
package user;
option go_package = "internal/proto";  // 完整 Go 导入路径:ml-citation{ref="1,2" data="citationList"}service UserService {rpc Register(UserRequest) returns (Response);
}message UserRequest {string username = 1;string password = 2;string email = 3;
}message Response {int32 code = 1;string message = 2;uint32 user_id = 3;
}

执行生成grpc代码

protoc --go_out=. --go-grpc_out=. internal/proto/pb/user.proto

生成的grpc的代码如下

定义grpc入口主函数

package mainimport ("log""net""user_service/internal/config"repository "user_service/internal/repostry""user_service/internal/service""user_service/internal/store/model""user_service/proto""google.golang.org/grpc"
)// 初始化配置文件
func init() {config.InitConfig()
}
func main() {// 初始化数据库db := model.InitDb()defer model.Db.Close()// 创建服务实例userRepo := repository.NewUserRepository(db)userService := service.NewUserService(userRepo)grpcServer := grpc.NewServer()初始化数据库连接后执行迁移db.AutoMigrate(&model.User{}, &model.Message{})// 注册gRPC服务proto.RegisterUserServiceServer(grpcServer, userService)// 启动服务lis, err := net.Listen("tcp", ":50051")if err != nil {log.Fatal(err)}log.Println("gRPC服务启动在 :50051")if err := grpcServer.Serve(lis); err != nil {log.Fatal(err)}
}

执行go mod tidy下载相关依赖

db表为 user_db

表结构如下

create table kanyuServer.user_db
(id          bigint unsigned auto_increment comment '主键'primary key,phone       varchar(11)                                     not null comment '手机号码',password    varchar(128) default ''                         null comment '密码，加密存储',user_name   varchar(32)  default ''                         null comment '昵称，默认是用户id',create_time timestamp    default CURRENT_TIMESTAMP          not null comment '创建时间',update_time timestamp    default CURRENT_TIMESTAMP          not null on update CURRENT_TIMESTAMP comment '更新时间',avatar      varchar(255) default 'https://picsum.photos/60' null,constraint uniqe_key_phoneunique (phone)
);

执行

grpc服务启动

go run main.go

执行成功可以看到如下日志

网关服务启动

执行成功可以看到如下日志

请求注册接口/api/v1/register

返回

执行到这里则grpc整合gin的例子成功完成了

参考

李文周go教程