golang 工程组件 grpc

组件 grpc"/>

golang 工程组件 grpc

yaml定义http规则，和自定义实现网关路由

proto定义http规则总归是麻烦的，因为proto文件还是定义消息，grpc接口好一些。配置http规则有更好的方式。我们可以使用yaml文件定义接口的http规则。 同时有些接口不是只是让网关转发这么简单 有时需要自己定网关接口handler

yaml定义http规则

type: google.api.Service
config_version: 3http:rules:# {package}.{message}.{method}- selector: user.User.Getget: "/user/{id}"- selector: user.User.AddOrUpdatepost: "/user"body: "*"additional_bindings:- put: "/user"body: "*"- patch: "/user"body: "addr"- selector: user.User.Deletedelete: "/user/{id}"

proto文件

syntax = "proto3";
package  user;
option go_package = "user/proto";message Member{int64 id = 1;string userName = 2[json_name = "user_name"];int32 age = 3;string phone = 4;Addr addr = 5;
}
message Addr {string province = 1;string city = 2;string county = 3;
}message UploadRequest {int64  size = 1;bytes content = 2;
}
message UploadResponse {string filePath= 1[json_name = "file_path"];
}
service User{rpc Get(Member) returns (Member) {}rpc AddOrUpdate(Member) returns (Member) { }rpc Delete(Member) returns (Member) {}rpc Upload(stream UploadRequest) returns (UploadResponse){}
}

生成消息,grpc,网关

# 生成message 
protoc --proto_path=proto --go_out=proto --go_opt=paths=source_relative proto/user.proto 
# 生成grpc service 
protoc --proto_path=proto --go-grpc_out=proto --go-grpc_opt=paths=source_relative proto/user.proto
#生成gateway
protoc --proto_path=proto  --grpc-gateway_out=proto  --grpc-gateway_opt logtostderr=true  --grpc-gateway_opt paths=source_relative  --grpc-gateway_opt grpc_api_configuration=proto/user.yaml  proto/user.proto

参考grpc-gateway入门中的启动代码就能调用对应接口啦

自定义实现网关路由

在生成gateway后，前面proto文件中我们预留了一个文件上传grpc接口。然后在yaml中我们是没有定义对应http规则。 所以需要自定义实现对应的网关路由，来应对复杂的业务情况。

gateway.go

通过mux.HandlePath添加自定义处理的路由和对应handler函数

package gatewayimport ("context""flag""google.golang/grpc/health/grpc_health_v1""net/http""user/user-server/gateway/middleware""github/grpc-ecosystem/grpc-gateway/v2/runtime""google.golang/grpc""google.golang/grpc/credentials/insecure"gw "user/proto"
)var (grpcServerEndpoint = flag.String("grpc-server-endpoint", "localhost:50051", "gRPC server endpoint")
)func Run() error {ctx := context.Background()ctx, cancel := context.WithCancel(ctx)defer cancel()inComingOpt := runtime.WithIncomingHeaderMatcher(func(s string) (string, bool) {switch s {case "Service-Authorization":return "service-authorization", truedefault:return "", false}return "", false})//创建连接,用于健康检查conn, err := grpc.Dial(*grpcServerEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))if err != nil {return err}mux := runtime.NewServeMux(inComingOpt, runtime.WithHealthzEndpoint(grpc_health_v1.NewHealthClient(conn)))//添加自定义处理函数mux.HandlePath("POST", "/upload", uploadHandler)handler := middleware.Cors(mux)opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}err = gw.RegisterUserHandlerFromEndpoint(ctx, mux, *grpcServerEndpoint, opts)if err != nil {return err}return http.ListenAndServe(":8081", handler)
}

upload.go

写对应网关需要注册的handler

package gatewayimport ("context""fmt""github/golang/protobuf/jsonpb""google.golang/grpc""google.golang/grpc/credentials/insecure""google.golang/grpc/metadata""io""net/http""user/proto"
)func uploadHandler(w http.ResponseWriter, r *http.Request, pathParams map[string]string) {serviceAuthorization := r.Header.Get("Service-Authorization")fmt.Println(serviceAuthorization)err := r.ParseForm()if err != nil {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}f, header, err := r.FormFile("attachment")if err != nil {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}defer f.Close()conn, err := grpc.Dial(*grpcServerEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))if err != nil {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}defer conn.Close()c := proto.NewUserClient(conn)ctx := context.Background()ctx = metadata.NewOutgoingContext(ctx, metadata.New(map[string]string{"file_name": header.Filename, "service-authorization": serviceAuthorization}))stream, err := c.Upload(ctx)if err != nil {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}buf := make([]byte, 100)for {n, err := f.Read(buf)if err != nil && err != io.EOF {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}if n == 0 {break}stream.Send(&proto.UploadRequest{Content: buf[:n],Size:    int64(n),})}res, err := stream.CloseAndRecv()if err != nil {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}m := jsonpb.Marshaler{}str, err := m.MarshalToString(res)if err != nil {http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)return}w.Header().Add("Content-Type", "application/json")fmt.Fprint(w, str)
}

重新启动即可