Go Kit

发表于 更新于 分类于 阅读次数：

Go Kit是个相当优秀的微服务框架，没有太多使用上的限制，提供了丰富的组件以实现各种强大的功能。

本文简单介绍框架的结构和使用，没有负载均衡，流量限制等实现。

go-kit结构流程

go-kit 框架主要结构分为三层：

  • transport：负责数据的传输方式定义，包括传输协议的数据和业务数据结构的转换（编码，解码）。
  • endpoint：负责连接 transport 和 service 层，起到连接两层的作用。
  • service：业务逻辑定义，通常和 repository 层交互，对数据进行操作。

示例

示例只有一个接口，为传入用户 ID 查询用户信息返回，项目结构如下

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
.
├── cmd
│   ├── grpc
│   │   └── main.go
│   └── http
│       └── main.go
├── endpoint
│   └── user.go
├── generate_pb.sh
├── go.mod
├── go.sum
├── model
│   ├── http_request.go
│   ├── http_response.go
│   ├── user.go
│   └── user_table.go
├── pb
│   ├── user.pb.go
│   └── user_grpc.pb.go
├── repositroy
│   ├── db.go
│   └── user.go
├── service
│   └── user.go
├── transport
│   ├── grpc.go
│   ├── grpc_decode.go
│   ├── grpc_encode.go
│   ├── http.go
│   ├── http_decode.go
│   └── http_encode.go
└── user.proto

请求处理过程

用户—decode—endpoint—service—...数据操作—service 返回—endpoint 返回—encode—用户

service

服务简单定义一下

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

import (
	"gorm.io/gorm"
	"kit-demo/model"
)

type UserSvcInterface interface {
	GetUserByID(user *model.UserTable, tx *gorm.DB) error
}

type UserSvc struct{}

func NewUserSvc() *UserSvc {
	return &UserSvc{}
}

func (u UserSvc) GetUserByID(user *model.UserTable, tx *gorm.DB) error {
	if user.ID == 100 {
		user.Name = "admin"
	}
	return nil
}

endpoint

这里主要是就是转换传入参数的数据，传入调用的服务中然后返回接口执行的结果。

传入参数是在 transport 层中的编码解码部分处理，在这里只需要确保传入参数是业务逻辑需要的数据类型即可

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

import (
	"context"
	"errors"
	"github.com/go-kit/kit/endpoint"
	"kit-demo/model"
	"kit-demo/service"
)

func MakeGetUserByIDEndpoint(svc service.UserSvcInterface) endpoint.Endpoint {
	return func(ctx context.Context, request interface{}) (response interface{}, err error) {
		user, ok := request.(*model.UserTable)
		if !ok {
			return nil, errors.New("request data type error")
		}

		if user.ID == 0 {
			return nil, errors.New("id is required")
		}

		err = svc.GetUserByID(user, nil)
		if err != nil {
			return nil, err
		}
		return user, nil
	}
}

transport

本层分为两部分，grpc 和 http 两种，当然 go-kit 还支持其他传输方式。可以选择仅实现其中一种。

encode、decode、endpoint 执行顺序

go-kit 中的源码可以看到顺序为：decode、endpoint、encode

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
// go-kit http 部分源码
// s.dec  是  decode 函数
// s.enc  是 encode 函数
// s.e 是 endpoint 函数
func (s Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	if len(s.finalizer) > 0 {
		iw := &interceptingWriter{w, http.StatusOK, 0}
		defer func() {
			ctx = context.WithValue(ctx, ContextKeyResponseHeaders, iw.Header())
			ctx = context.WithValue(ctx, ContextKeyResponseSize, iw.written)
			for _, f := range s.finalizer {
				f(ctx, iw.code, r)
			}
		}()
		w = iw
	}

	for _, f := range s.before {
		ctx = f(ctx, r)
	}
  // 执行 decode
	request, err := s.dec(ctx, r)
	if err != nil {
		s.errorHandler.Handle(ctx, err)
		s.errorEncoder(ctx, err, w)
		return
	}
  // 执行 endpoint
	response, err := s.e(ctx, request)
	if err != nil {
		s.errorHandler.Handle(ctx, err)
		s.errorEncoder(ctx, err, w)
		return
	}

	for _, f := range s.after {
		ctx = f(ctx, w)
	}
  // 执行 encode
	if err := s.enc(ctx, w, response); err != nil {
		s.errorHandler.Handle(ctx, err)
		s.errorEncoder(ctx, err, w)
		return
	}
}

grpc

使用 grpc 方式，我们需要先定义 proto 协议文件，本示例协议文件内容如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
syntax = "proto3";

option go_package = "kit-demo/pb";

service UserService {
  rpc GetUserByID(IDRequest) returns (UserInfo);
}

message IDRequest {
  uint64 id = 1;
}

message UserInfo {
  uint64 id = 1;
  string account = 2;
  string name = 3;
  string phone = 4;
  string email = 5;
  int64 create_ts = 6;
}

简单描述了服务需要的参数类型和返回值类型

使用工具生成 go 的 pb 文件

1
2
3
4
5
#!/usr/bin/env bash
protoc --go_out=./pb --go_opt=paths=source_relative \
       --go-grpc_out=./pb --go-grpc_opt=paths=source_relative \
       user.proto

在 transport 层中实现 pb 文件中的接口。

这里在创建服务时需要传入 endpoint 和 decode， encode 两个编码解码方法

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
// transport/grpc.go
package transport

import (
	"context"
	transportGrpc "github.com/go-kit/kit/transport/grpc"
	"kit-demo/endpoint"
	"kit-demo/pb"
	"kit-demo/service"
)

// UserGrpcServer 实现 pb 文件接口
type UserGrpcServer struct {
	getUserByIDHandler transportGrpc.Handler
	pb.UnimplementedUserServiceServer
}

func NewUserGrpcServer(svc service.UserSvcInterface, opts ...transportGrpc.ServerOption) pb.UserServiceServer {
	getUserByIDHandler := transportGrpc.NewServer(
		endpoint.MakeGetUserByIDEndpoint(svc),
		decodeGetUserByIDRequest,
		encodeGetUserByIDResponse,
		opts...,
	)

	userGrpcServer := &UserGrpcServer{
		getUserByIDHandler: getUserByIDHandler,
	}
	return userGrpcServer
}

func (u UserGrpcServer) GetUserByID(ctx context.Context, request *pb.IDRequest) (*pb.UserInfo, error) {
  // 调用 handler 方法处理请求
	_, resp, err := u.getUserByIDHandler.ServeGRPC(ctx, request)
	if err != nil {
		return nil, err
	}

	return resp.(*pb.UserInfo), nil
}

编码解码函数如下：

  • 解码方法主要是转换 pb 协议文件中的数据为 业务接口需要的 数据类型。
  • 编码方法主要是转换 业务数据类型 为 pb 协议文件中的数据类型。
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
func decodeGetUserByIDRequest(ctx context.Context, grpcRequest interface{}) (interface{}, error) {
	req, ok := grpcRequest.(*pb.IDRequest)
	if !ok {
		return nil, errors.New("grpc server decode request error")
	}
	request := &model.UserTable{
		ID: req.Id,
	}
	return request, nil
}


func encodeGetUserByIDResponse(ctx context.Context, response interface{}) (interface{}, error) {
	resp, ok := response.(*model.SvcResponseUserInfo)
	if !ok {
		return nil, errors.New("grpc server encode response error")
	}
	r := &pb.UserInfo{
		Id:       resp.ID,
		Account:  resp.Account,
		Name:     resp.Name,
		Phone:    resp.Phone,
		Email:    resp.Email,
		CreateTs: resp.CreateTs,
	}
	return r, 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
package main

import (
	"fmt"
	"google.golang.org/grpc"
	"kit-demo/pb"
	"kit-demo/service"
	"kit-demo/transport"
	"net"
)

func main() {
	var opts []grpc.ServerOption

	grpcServer := grpc.NewServer(opts...)
	pb.RegisterUserServiceServer(grpcServer, transport.NewUserGrpcServer(service.NewUserSvc()))

	fmt.Println("grpc server start")
	listener, err := net.Listen("tcp", ":8099")
	if err != nil {
		fmt.Println(err.Error())
		return
	}
	err = grpcServer.Serve(listener)
	if err != nil {
		fmt.Println(err.Error())
		return
	}
}

http

http 部分本文使用 gin 框架来处理请求和路由，逻辑基本同 grpc 一致

接口部分实现

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
// tansport/http.go
package transport

import (
	"github.com/gin-gonic/gin"
	transportHTTP "github.com/go-kit/kit/transport/http"
	"kit-demo/endpoint"
	"kit-demo/service"
	"net/http"
)

type UserHTTPServerInterface interface {
	GetUserByID(ctx *gin.Context)
}

type UserHTTPServer struct {
	getUserByIDHandler http.Handler
}

func NewUserHTTPServer(svc service.UserSvcInterface, opts ...transportHTTP.ServerOption) UserHTTPServerInterface {

	getUserByIDHandler := transportHTTP.NewServer(
		endpoint.MakeGetUserByIDEndpoint(svc),
		decodeGetUserByIDHttpReq,
		encodeGetUserByIDHttpResp,
		opts...,
	)

	return &UserHTTPServer{
		getUserByIDHandler: getUserByIDHandler,
	}
}

//  gin 接口，保存在 keys 中的数据，可以转换到 header 中向 go-kit 中传递
func (u UserHTTPServer) GetUserByID(ctx *gin.Context) {
	ctx.Request.Header.Set("name", "xxxx")
	u.getUserByIDHandler.ServeHTTP(ctx.Writer, ctx.Request)
}

编码解码函数如下，同 grpc 作用一致，转换数据为服务端需要的类型，然后交给 endpoint 处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func decodeGetUserByIDHttpReq(ctx context.Context, request *http.Request) (interface{}, error) {
	bodyBytes, err := ioutil.ReadAll(request.Body)
	if err != nil {
		return nil, err
	}
	req := &model.RequestGetUserByID{}
	err = json.Unmarshal(bodyBytes, &req)
	if err != nil {
		return nil, err
	}
	user := &model.UserTable{ID: req.ID}
	return user, nil
}


func encodeGetUserByIDHttpResp(ctx context.Context, writer http.ResponseWriter, resp interface{}) error {
	writer.Header().Set("Content-Type", "application/json")
	return json.NewEncoder(writer).Encode(resp)
}

对外提供服务

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

import (
	"fmt"
	"github.com/gin-gonic/gin"
	"kit-demo/service"
	"kit-demo/transport"
)

var (
	engine *gin.Engine
)

func init() {
	gin.SetMode(gin.ReleaseMode)
	engine = gin.Default()

	api := transport.NewUserHTTPServer(service.NewUserSvc())
	engine.POST("/user/get", api.GetUserByID)
}

func main() {
	err := engine.Run(":8100")
	if err != nil {
		fmt.Println(err)
		return
	}
}