^-^
【go工程组件】网关grpc-gateway
最近更新：2024-12-12   |   字数总计：3.1k   |   阅读估时：17分钟   |   阅读量：次
  1. 网关grpc-gateway
    1. 网关的默认实现
    2. option定义规则及HttpBody响应
      1. option定义http规则
      2. 返回httpbody作为响应
    3. 配置文件定义http规则及网关文件上传
      1. 定义服务
      2. 使用YAML定义http规则
      3. 生成网关代码
      4. 实现服务
      5. 实现网关
      6. 实现网关对文件上传的处理
      7. 启用网关及服务
    4. http header的透传
      1. 拦截器用于身份验证
      2. 网关服务器添加透传设置
      3. 文件上传网关需要手动透传
    5. grpc健康检查网关配置
      1. 为服务添加健康检查
      2. 网关启用健康检查
      3. 测试结果
      4. 网关中间件跨域支持

网关grpc-gateway

  • 作用
    1. 通过反向代理的方式，将grpc server接口转为http+json api
  • 使用场景
    1. 向后兼容
    2. 支持grpc不支持的语言或客户端
  • 插件安装
    1
    go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@v2.12.0

网关的默认实现

  • 生成默认网关代码

    1
    protoc -I . --grpc-gateway_out . --grpc-gateway_opt paths=source_relative --grpc-gateway_opt generate_unbound_methods=true ./proto/echo.proto

  • 实现网关代码,gateway.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
    package gateway

    import (
    	"context"
    	"flag"
    	"net/http"

    	gw "echo/proto"

    	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
    	"google.golang.org/grpc"
    	"google.golang.org/grpc/credentials/insecure"
    )

    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()
    	mux := runtime.NewServeMux()
    	opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
    	err := gw.RegisterEchoHandlerFromEndpoint(ctx, mux, *grpcServerEndpoint, opts)
    	if err != nil {
    		return err
    	}
    	return http.ListenAndServe(":8888", mux)
    }

  • server.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
    package server

    import (
    	"context"
    	"echo/proto"
    	"fmt"
    )

    type echoServer struct {
    	proto.UnimplementedEchoServer
    }

    func NewServer() proto.EchoServer {
    	return &echoServer{}
    }

    func (s *echoServer) Get(ctx context.Context, in *proto.User) (*proto.User, error) {
    	fmt.Println("#{in}\n")
    	return in, nil
    }

    func (s *echoServer) AddOrUpdate(ctx context.Context, in *proto.User) (*proto.User, error) {
    	fmt.Println("#{in}\n")
    	return in, nil
    }

    func (s *echoServer) Delete(ctx context.Context, in *proto.User) (*proto.User, error) {
    	fmt.Println("#{in}\n")
    	return in, nil
    }

  • 启用网关服务

    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
    package main

    import (
    	"context"
    	"echo/echo-server/gateway"
    	"echo/echo-server/server"
    	"echo/proto"
    	"log"
    	"net"
    	"os"
    	"os/signal"
    	"time"

    	"google.golang.org/grpc"
    )

    func main() {
    	go func() {
    		if err := run(); err != nil {
    			log.Fatal(err)
    		}
    	}()
    	time.Sleep(time.Second) // 确保服务先启动，网关再启动
    	go func() {
    		if err := gateway.Run(); err != nil {
    			log.Fatal(err)
    		}
    	}()
    	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
    	defer stop()
    	<-ctx.Done()
    }

    func run() error {
    	lis, err := net.Listen("tcp", ":50051")
    	if err != nil {
    		log.Fatal(err)
    	}
    	s := grpc.NewServer()
    	userServiceServer := server.NewServer()
    	proto.RegisterEchoServer(s, userServiceServer)
    	return s.Serve(lis)
    }

  • 测试结果

  • 默认网关只支持post方法

option定义规则及HttpBody响应

option定义http规则

  • 导入google api的proto定义: https://github.com/googleapis/googleapis

  • 使用option定义http规则

    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
    syntax = "proto3";
    package echo;
    option go_package = "echo/proto";
    import "google/api/annotations.proto";

    message User{
        int64 id = 1;
        string name = 2;
        int32 age= 3;
        string phone = 4;
        Addr addr = 5;
    }

    message Addr{
        string province =1 ;
        string city = 2;
        string county = 3;
    }

    service Echo{
        rpc Get(User) returns (User){
            option (google.api.http) = {
                get : "/echo/user/{id}"
            };
        }
        rpc AddOrUpdate(User) returns (User){
            option (google.api.http) = {
                post : "/echo/user"
                body:"*"
                additional_bindings {
                    put:"/echo/user"
                    body:"*"
                }
                additional_bindings {
                    patch:"/echo/user"
                    body:"addr"
                }
            };
        }
        rpc Delete(User) returns (User){
            option (google.api.http) = {
                delete : "/echo/user/{id}"
            };
        }
    }

  • 重新生成网关代码

    1
    2
    3
    4
    5
    6
    7
    8
    protoc -I ./proto/ --grpc-gateway_out ./proto/ --grpc-gateway_opt paths=source_relative ./proto/echo.proto # 注意路径
    # 重新生成所有内容
    protoc -I ./proto/ \
     --go-grpc_out ./proto/ \
     --go-grpc_opt paths=source_relative \
     --go_out ./proto/ --go_opt paths=source_relative \
     --grpc-gateway_out ./proto/ --grpc-gateway_opt paths=source_relative \
      ./proto/echo.proto

  • 再次测试结果

  • patch方法因为只接收body的内容是addr，所以只能单独写一个addr结构的body作为请求

返回httpbody作为响应

  • 新加一个rpc接口，返回httpbody
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    syntax = "proto3";
    package echo;
    option go_package = "echo/proto";
    import "google/api/annotations.proto";
    import "google/api/httpbody.proto";
    import "google/protobuf/empty.proto";

    message User{
     //...
     string userName = 2[json_name="user_name"]; // 干预一下接收的格式，可以接收userName和user_name，不能是username
    }

    message Addr{
     //...
    }

    service Echo{
        // ...
        rpc List(google.protobuf.Empty) returns (stream google.api.HttpBody){
            option (google.api.http) = {
                get:"/echo/user/list"
            };
        }
    }
  • 实现接口,server.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
    func (s *echoServer) List(emp *emptypb.Empty, stream proto.Echo_ListServer) error {
    	userList := []*proto.User{
    		{
    			Id:       1,
    			UserName: "nick",
    			Addr: &proto.Addr{
    				Province: "hunan",
    			},
    		},
    		{
    			Id:       2,
    			UserName: "mm",
    			Addr: &proto.Addr{
    				Province: "shenzhen",
    			},
    		},
    		{
    			Id:       3,
    			UserName: "ll",
    			Addr: &proto.Addr{
    				Province: "beij",
    			},
    		},
    	}
    	for _, u := range userList {
    		m := jsonpb.Marshaler{}
    		data, _ := m.MarshalToString(u)
    		// data, _ := json.Marshal(u) // 用平时用的json包会忽略掉我们再proto中对json格式的干预
    		msg := &httpbody.HttpBody{
    			ContentType: "application/json",
    			Data:        []byte(data),
    		}
    		stream.Send(msg)
    	}
    	return nil
    }
  • 测试结果

配置文件定义http规则及网关文件上传

定义服务

  • user.proto
    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
    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 country=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){}
    }

使用YAML定义http规则

  • user.yaml
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    type: google.api.Service
    config_version: 3

    http:
      rules:
        - selector: user.User.Get
          get: "/user/{id}"
        - selector: user.User.AddOrUpdate
          post: "/user"
          body: "*"
          additional_bindings:
            - put: "/user"
              body: "*"
            - patch: "/user"
              body: "addr"
        - selector: user.User.Delete
          delete: "/user/{id}"

生成网关代码

1
2
3
4
5
6
7
protoc -I ./proto/ \
 --go-grpc_out ./proto/ \
 --go-grpc_opt paths=source_relative \
 --go_out ./proto/ --go_opt paths=source_relative \
 --grpc-gateway_out ./proto/ --grpc-gateway_opt grpc_api_configuration=./proto/user.yaml \ 
 --grpc-gateway_opt paths=source_relative \
 ./proto/user.proto

实现服务

  • server.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
    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
    package server

    import (
    	"context"
    	"errors"
    	"fmt"
    	"io"
    	"log"
    	"os"
    	"user/proto"

    	"google.golang.org/grpc/metadata"
    )

    type userServer struct {
    	proto.UnimplementedUserServer
    }

    func NewServer() proto.UserServer {
    	return &userServer{}
    }

    func (s *userServer) Get(ctx context.Context, in *proto.Member) (*proto.Member, error) {
    	fmt.Println("#{in}\n")
    	return in, nil
    }

    func (s *userServer) AddOrUpdate(ctx context.Context, in *proto.Member) (*proto.Member, error) {
    	fmt.Println("#{in}\n")
    	return in, nil
    }

    func (s *userServer) Delete(ctx context.Context, in *proto.Member) (*proto.Member, error) {
    	fmt.Println("#{in}\n")
    	return in, nil
    }

    func (s *userServer) Upload(stream proto.User_UploadServer) error {
    	md, ok := metadata.FromIncomingContext(stream.Context())
    	if !ok || len(md["file_name"]) <= 0 {
    		log.Println("metadata get failed")
    		return errors.New("文件名获取失败")
    	}
    	filename := md["file_name"][0]
    	filePath := "./user-server/upload-file/" + filename
    	dst, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
    	if err != nil {
    		log.Fatal(err)
    	}
    	defer dst.Close()
    	for {
    		req, err := stream.Recv()
    		if err == io.EOF {
    			break
    		}
    		if err != nil {
    			log.Println(err)
    			break
    		}
    		dst.Write(req.Content[:req.Size])
    		// fmt.Printf("server recv : %v\n", req.Message)
    	}
    	err = stream.SendAndClose(&proto.UploadResponse{
    		FilePath: filePath,
    	})
    	return nil
    }

实现网关

  • gateway.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
    package gateway

    import (
    	"context"
    	"flag"
    	"net/http"

    	gw "user/proto"

    	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
    	"google.golang.org/grpc"
    	"google.golang.org/grpc/credentials/insecure"
    )

    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()
    	mux := runtime.NewServeMux()

    	// upload 不能通过配置或者proto option的方式配置，只能手动添加handle自定义处理函数
    	mux.HandlePath("POST", "/upload", uploadHandle)

    	opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
    	err := gw.RegisterUserHandlerFromEndpoint(ctx, mux, *grpcServerEndpoint, opts)
    	if err != nil {
    		return err
    	}
    	return http.ListenAndServe(":8888", mux)
    }

实现网关对文件上传的处理

  • upload.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
    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
    72
    73
    74
    75
    package gateway

    import (
    	"context"
    	"fmt"
    	"io"
    	"net/http"
    	"user/proto"

    	"github.com/golang/protobuf/jsonpb"
    	"google.golang.org/grpc"
    	"google.golang.org/grpc/credentials/insecure"
    	"google.golang.org/grpc/metadata"
    )

    func uploadHandle(w http.ResponseWriter, r *http.Request, pathParams map[string]string) {
    	err := r.ParseForm()
    	if err != nil {
    		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
    		return
    	}
    	file, header, err := r.FormFile("attachment")
    	if err != nil {
    		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
    		return
    	}
    	defer file.Close()
    	// 代理不能直接做文件上传，这里我们需要gateway作为客户端去访问grpc的server
    	// 实际生产中，请使用连接池，不要每次都创建连接
    	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,
    	}))
    	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 := file.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 && err != io.EOF {
    		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
    		return
    	}
    	m := jsonpb.Marshaler{}
    	str, err := m.MarshalToString(res)
    	if err != nil && err != io.EOF {
    		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
    		return
    	}
    	w.Header().Add("Content-Type", "application/json")
    	fmt.Fprint(w, str)
    }

启用网关及服务

  • main.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
    42
    43
    44
    45
    package main

    import (
    	"context"
    	"user/proto"
    	"user/user-server/gateway"
    	"user/user-server/server"

    	"log"
    	"net"
    	"os"
    	"os/signal"
    	"time"

    	"google.golang.org/grpc"
    )

    func main() {
    	go func() {
    		if err := run(); err != nil {
    			log.Fatal(err)
    		}
    	}()
    	time.Sleep(time.Second) // 确保服务先启动，网关再启动
    	go func() {
    		if err := gateway.Run(); err != nil {
    			log.Fatal(err)
    		}
    	}()
    	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
    	defer stop()
    	<-ctx.Done()
    }

    func run() error {
    	lis, err := net.Listen("tcp", ":50051")
    	if err != nil {
    		log.Fatal(err)
    	}
    	s := grpc.NewServer()
    	userServiceServer := server.NewServer()
    	proto.RegisterUserServer(s, userServiceServer)
    	return s.Serve(lis)
    }

http header的透传

拦截器用于身份验证

  • interceptor.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
    42
    43
    44
    45
    package server

    import (
    	"context"
    	"errors"
    	"strings"

    	"google.golang.org/grpc"
    	"google.golang.org/grpc/metadata"
    )

    func UnaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
    	err = auth(ctx)
    	if err != nil {
    		return nil, err
    	}
    	return handler(ctx, req)
    }

    func StreamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    	err := auth(ss.Context())
    	if err != nil {
    		return err
    	}
    	return handler(srv, ss)
    }

    func auth(ctx context.Context) error {
    	md, ok := metadata.FromIncomingContext(ctx)
    	if !ok {
    		return errors.New("获取元数据失败，身份校验失败")
    	}
    	authorization := md["service-authorization"]
    	if len(authorization) < 1 {
    		return errors.New("获取身份令牌失败")
    	}
    	token := strings.TrimPrefix(authorization[0], "Bearer ")
    	if token != bearerToken {
    		return errors.New("身份令牌对比失败")
    	}
    	return nil
    }

    var bearerToken = "ioadnklghoiulknlknlijlkjlkwerqwerfdiufioshofndaklsnfd"

网关服务器添加透传设置

  • gateway.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
    42
    43
    44
    45
    46
    package gateway

    import (
    	"context"
    	"flag"
    	"net/http"

    	gw "user/proto"

    	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
    	"google.golang.org/grpc"
    	"google.golang.org/grpc/credentials/insecure"
    )

    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()
    	// 将http header的中的内容转发到grpc metadata上下文中
    	inComintOpt := runtime.WithIncomingHeaderMatcher(func(s string) (string, bool) {
    		switch s {
    		case "Service-Authorization":
    			return "service-authorization", true
    		default:
    			return "", false
    		}
    	})
    	// 将grpc metadata上下文中的内容写入到http响应给客户端的header中
    	// outComintOpt := runtime.WithOutgoingHeaderMatcher(func(s string) (string, bool) {})
    	mux := runtime.NewServeMux(inComintOpt)

    	// upload 不能通过配置或者proto option的方式配置，只能手动添加handle自定义处理函数
    	mux.HandlePath("POST", "/upload", uploadHandle)

    	opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
    	err := gw.RegisterUserHandlerFromEndpoint(ctx, mux, *grpcServerEndpoint, opts)
    	if err != nil {
    		return err
    	}
    	return http.ListenAndServe(":8888", mux)
    }

文件上传网关需要手动透传

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
72
73
74
75
76
77
78
79
package gateway

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"user/proto"

	"github.com/golang/protobuf/jsonpb"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
)

func uploadHandle(w http.ResponseWriter, r *http.Request, pathParams map[string]string) {
	// upload的转发需要手动添加透传的header
	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
	}
	file, header, err := r.FormFile("attachment")
	if err != nil {
		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
		return
	}
	defer file.Close()
	// 代理不能直接做文件上传，这里我们需要gateway作为客户端去访问grpc的server
	// 实际生产中，请使用连接池，不要每次都创建连接
	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, // upload的转发需要手动添加透传的header
	}))
	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 := file.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 && err != io.EOF {
		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
		return
	}
	m := jsonpb.Marshaler{}
	str, err := m.MarshalToString(res)
	if err != nil && err != io.EOF {
		http.Error(w, fmt.Sprintf("上传失败：%s", err.Error()), http.StatusInternalServerError)
		return
	}
	w.Header().Add("Content-Type", "application/json")
	fmt.Fprint(w, str)
}

grpc健康检查网关配置

为服务添加健康检查

  • main.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    func run() error {
    	lis, err := net.Listen("tcp", ":50051")
    	if err != nil {
    		log.Fatal(err)
    	}
    	s := grpc.NewServer(grpc.UnaryInterceptor(server.UnaryInterceptor), grpc.StreamInterceptor(server.StreamInterceptor))
    	userServiceServer := server.NewServer()
    	proto.RegisterUserServer(s, userServiceServer)
    	// 添加健康检查
    	grpc_health_v1.RegisterHealthServer(s, health.NewServer())

    	return s.Serve(lis)
    }
  • 身份验证拦截器，忽略健康检查
  • interceptor.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    func UnaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
    	if info.FullMethod != "/grpc.health.v1.Health/Check" {
    		err = auth(ctx)
    		if err != nil {
    			return nil, err
    		}
    	}
    	return handler(ctx, req)
    }

    func StreamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    	if info.FullMethod != "/grpc.health.v1.Health/Check" {
    		err := auth(ss.Context())
    		if err != nil {
    			return err
    		}
    	}
    	return handler(srv, ss)
    }

网关启用健康检查

  • gateway.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
    func Run() error {
    	ctx := context.Background()
    	ctx, cancel := context.WithCancel(ctx)
    	defer cancel()
    	// 将http header的中的内容转发到grpc metadata上下文中
    	inComintOpt := runtime.WithIncomingHeaderMatcher(func(s string) (string, bool) {
    		switch s {
    		case "Service-Authorization":
    			return "service-authorization", true
    		default:
    			return "", false
    		}
    	})

    	// 创建健康检查客户端连接，用于健康检查
    	conn, err := grpc.Dial(*grpcServerEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
    	if err != nil {
    		return err
    	}
    	// 将grpc metadata上下文中的内容写入到http响应给客户端的header中
    	// outComintOpt := runtime.WithOutgoingHeaderMatcher(func(s string) (string, bool) {})
    	mux := runtime.NewServeMux(inComintOpt, runtime.WithHealthzEndpoint(grpc_health_v1.NewHealthClient(conn)))

    	// upload 不能通过配置或者proto option的方式配置，只能手动添加handle自定义处理函数
    	mux.HandlePath("POST", "/upload", uploadHandle)

    	opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
    	err = gw.RegisterUserHandlerFromEndpoint(ctx, mux, *grpcServerEndpoint, opts)
    	if err != nil {
    		return err
    	}
    	return http.ListenAndServe(":8888", mux)
    }

测试结果

网关中间件跨域支持

  • middleware.go
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    package middleware

    import "net/http"

    func Cors(handler http.Handler) http.Handler {
    	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    		method := r.Method
    		origin := r.Header.Get("Origin")
    		if origin != "" {
    			w.Header().Set("Access-Control-Allow-Origin", "*")
    			w.Header().Set("Access-Control-Allow-Methods", "POST,GET,OPTIONS,PUT,DELETE,UPDATE, PATCH")
    			w.Header().Set("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept, Authorization")
    			w.Header().Set("Access-Control-Allow-Credentials", "true")
    			w.Header().Set("Access-Control-Expose-Headers", "Content-Length, Access-Control-Allow-Origin, Access-Control-Allow-Headers, Cache-Control, Content-Language, Content-Type")
    		}
    		if method == "OPTIONS" {
    			w.WriteHeader(http.StatusNoContent)
    			return
    		}
    		handler.ServeHTTP(w, r)
    	})
    }

  • gateway.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
    func Run() error {
    	ctx := context.Background()
    	ctx, cancel := context.WithCancel(ctx)
    	defer cancel()
    	// 将http header的中的内容转发到grpc metadata上下文中
    	inComintOpt := runtime.WithIncomingHeaderMatcher(func(s string) (string, bool) {
    		switch s {
    		case "Service-Authorization":
    			return "service-authorization", true
    		default:
    			return "", false
    		}
    	})

    	// 创建健康检查客户端连接，用于健康检查
    	conn, err := grpc.Dial(*grpcServerEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
    	if err != nil {
    		return err
    	}
    	// 将grpc metadata上下文中的内容写入到http响应给客户端的header中
    	// outComintOpt := runtime.WithOutgoingHeaderMatcher(func(s string) (string, bool) {})
    	mux := runtime.NewServeMux(inComintOpt, runtime.WithHealthzEndpoint(grpc_health_v1.NewHealthClient(conn)))

    	// upload 不能通过配置或者proto option的方式配置，只能手动添加handle自定义处理函数
    	mux.HandlePath("POST", "/upload", uploadHandle)
    	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(":8888", handler)
    }

  • 响应结果
2024-12-12 该篇文章被 Cleofwine 归为分类: Go云原生

上一篇：【go工程组件】配置解决方案viper 下一篇：【go工程组件】结构化日志记录logrus
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题