gRPC

transporter/grpc 中基于谷歌的 grpc 框架实现了Transporter，用以注册 grpc 到 kratos.Server() 中。

Server

配置

Network(network string) ServerOption

配置服务端的 network 协议，如 tcp

Address(addr string) ServerOption

配置服务端监听的地址

Timeout(timeout time.Duration) ServerOption

配置服务端的超时设置

Logger(logger log.Logger) ServerOption

配置服务端使用的日志组件

Middleware(m ...middleware.Middleware) ServerOption

配置服务端的 kratos 中间件

TLSConfig(c *tls.Config) ServerOption

配置服务端的 TLS 配置

UnaryInterceptor(in ...grpc.UnaryServerInterceptor) ServerOption

配置服务端使用的 grpc 单元拦截器

StreamInterceptor(in ...grpc.StreamServerInterceptor) ServerOption

配置服务端使用的 grpc 流媒体拦截器

Options(opts ...grpc.ServerOption) ServerOption

配置一些额外的 grpc.ServerOption

主要的实现细节

NewServer()

  1. func NewServer(opts ...ServerOption) *Server {
  2.     // grpc server 默认配置
  3.     srv := &Server{
  4.         network: "tcp",
  5.         address: ":0",
  6.         timeout: 1 * time.Second,
  7.         health:  health.NewServer(),
  8.         log:     log.NewHelper(log.GetLogger()),
  9.     }
  10.     // 递归 opts
  11.     for _, o := range opts {
  12.         o(srv)
  13.     }
  14.     // kratos middleware 转换成 grpc 拦截器，并处理一些细节
  15.     unaryInts := []grpc.UnaryServerInterceptor{
  16.         srv.unaryServerInterceptor(),
  17.     }
  18.     streamInts := []grpc.StreamServerInterceptor{
  19.         srv.streamServerInterceptor(),
  20.     }
  22.     if len(srv.unaryInts) > 0 {
  23.         unaryInts = append(unaryInts, srv.unaryInts...)
  24.     }
  25.     if len(srv.streamInts) > 0 {
  26.         streamInts = append(streamInts, srv.streamInts...)
  27.     }
  29.     // 将 UnaryInterceptor 和 StreamInterceptor 转换成 ServerOption
  30.     var grpcOpts = []grpc.ServerOption{
  31.         grpc.ChainUnaryInterceptor(unaryInts...),
  32.         grpc.ChainStreamInterceptor(streamInts...),
  33.     }
  34.     // 将 TLS 配置转化成 ServerOption
  35.     if srv.tlsConf != nil {
  36.         grpcOpts = append(grpcOpts, grpc.Creds(credentials.NewTLS(srv.tlsConf)))
  37.     }
  38.     // 追加通过 Options(opts ...grpc.ServerOption) 添加的options
  39.     if len(srv.grpcOpts) > 0 {
  40.         grpcOpts = append(grpcOpts, srv.grpcOpts...)
  41.     }
  42.     // 创建 grpc server
  43.     srv.Server = grpc.NewServer(grpcOpts...)
  44.     // 创建 metadata server
  45.     srv.metadata = apimd.NewServer(srv.Server)
  46.     // 配置 lis 和 endpoint
  47.     srv.err = srv.listenAndEndpoint()
  48.     // 内部注册
  49.     grpc_health_v1.RegisterHealthServer(srv.Server, srv.health)
  50.     apimd.RegisterMetadataServer(srv.Server, srv.metadata)
  51.     reflection.Register(srv.Server)
  52.     return srv
  53. }

unaryServerInterceptor()

  1. func (s *Server) unaryServerInterceptor() grpc.UnaryServerInterceptor {
  2.     return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
  3.         // 把两个 ctx 合并成一个
  4.         ctx, cancel := ic.Merge(ctx, s.ctx)
  5.         defer cancel()
  6.         // 从 ctx 中取出 metadata
  7.         md, _ := grpcmd.FromIncomingContext(ctx)
  8.         // 把一些信息绑定到 ctx 上
  9.         replyHeader := grpcmd.MD{}
  10.         ctx = transport.NewServerContext(ctx, &Transport{
  11.             endpoint:    s.endpoint.String(),
  12.             operation:   info.FullMethod,
  13.             reqHeader:   headerCarrier(md),
  14.             replyHeader: headerCarrier(replyHeader),
  15.         })
  16.         // ctx 超时设置
  17.         if s.timeout > 0 {
  18.             ctx, cancel = context.WithTimeout(ctx, s.timeout)
  19.             defer cancel()
  20.         }
  21.         // 中间件处理
  22.         h := func(ctx context.Context, req interface{}) (interface{}, error) {
  23.             return handler(ctx, req)
  24.         }
  25.         if len(s.middleware) > 0 {
  26.             h = middleware.Chain(s.middleware...)(h)
  27.         }
  28.         // 执行中间件 handler
  29.         reply, err := h(ctx, req)
  30.         if len(replyHeader) > 0 {
  31.             _ = grpc.SetHeader(ctx, replyHeader)
  32.         }
  33.         return reply, err
  34.     }
  35. }

使用方式

简单列举了一些 kratos 中 grpc 的用法，其他 grpc 用法可以到 grpc 仓库中查看。

注册 grpc server

  1. gs := grpc.NewServer()
  2. app := kratos.New(
  3.     kratos.Name("kratos"),
  4.     kratos.Version("v1.0.0"),
  5.     kratos.Server(gs),
  6. )

grpc server 中使用 kratos middleware

  1. grpcSrv := grpc.NewServer(
  2.     grpc.Address(":9000"),
  3.     grpc.Middleware(
  4.         logging.Server(),
  5.     ),
  6. )

middleware 中处理 grpc 请求

  1. if info, ok := transport.FromServerContext(ctx); ok {
  2.   kind = info.Kind().String()
  3.   operation = info.Operation()
  4. }

Client

配置

WithEndpoint(endpoint string) ClientOption

配置客户端使用的对端连接地址，如果不使用服务发现则为ip:port,如果使用服务发现则格式为discovery://\<authority>/\<serviceName>

WithTimeout(timeout time.Duration) ClientOption

配置客户端的请求默认超时时间，如果有链路超时优先使用链路超时时间

WithMiddleware(m ...middleware.Middleware) ClientOption

配置客户端使用的 kratos 中间件

WithDiscovery(d registry.Discovery) ClientOption

配置客户端使用的服务发现

WithTLSConfig(c *tls.Config) ClientOption

配置客户端使用的 TLS 配置

WithUnaryInterceptor(in ...grpc.UnaryClientInterceptor) ClientOption

配置客户端使用的 grpc 原生拦截器

WithOptions(opts ...grpc.DialOption) ClientOption

配置一些额外的 grpc.ClientOption

主要的实现细节

dial()

  1. func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.ClientConn, error) {
  2.     // 默认配置
  3.     options := clientOptions{
  4.         timeout:      2000 * time.Millisecond,
  5.         balancerName: wrr.Name,
  6.         logger:       log.GetLogger(),
  7.     }
  8.     // 遍历 opts
  9.     for _, o := range opts {
  10.         o(&options)
  11.     }
  12.     // 将 kratos 中间件转化成 grpc 拦截器
  13.     ints := []grpc.UnaryClientInterceptor{
  14.         unaryClientInterceptor(options.middleware, options.timeout, options.filters),
  15.     }
  16.     if len(options.ints) > 0 {
  17.         ints = append(ints, options.ints...)
  18.     }
  19.     // 负载均衡
  20.     grpcOpts := []grpc.DialOption{
  21.         grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, options.balancerName)),
  22.         grpc.WithChainUnaryInterceptor(ints...),
  23.     }
  24.     if options.discovery != nil {
  25.         // 如果存在服务发现配置，就配置 grpc 的 Resolvers
  26.         grpcOpts = append(grpcOpts,
  27.             grpc.WithResolvers(
  28.                 discovery.NewBuilder(
  29.                     options.discovery,
  30.                     discovery.WithInsecure(insecure),
  31.                     discovery.WithLogger(options.logger),
  32.                 )))
  33.     }
  34.     if insecure {
  35.         // 跳过证书验证
  36.         grpcOpts = append(grpcOpts, grpc.WithTransportCredentials(grpcinsecure.NewCredentials()))
  37.     }
  38.     // TLS 配置
  39.     if options.tlsConf != nil {
  40.         grpcOpts = append(grpcOpts, grpc.WithTransportCredentials(credentials.NewTLS(options.tlsConf)))
  41.     }
  42.     if len(options.grpcOpts) > 0 {
  43.         grpcOpts = append(grpcOpts, options.grpcOpts...)
  44.     }
  45.     return grpc.DialContext(ctx, options.endpoint, grpcOpts...)
  46. }

unaryClientInterceptor()

  1. func unaryClientInterceptor(ms []middleware.Middleware, timeout time.Duration) grpc.UnaryClientInterceptor {
  2.     return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
  3.         // 把一些信息绑定到 ctx 上
  4.         ctx = transport.NewClientContext(ctx, &Transport{
  5.             endpoint:  cc.Target(),
  6.             operation: method,
  7.             reqHeader: headerCarrier{},
  8.             filters:   filters,
  9.         })
  10.         if timeout > 0 {
  11.             // timeout 如果大于 0，就重新设置一下 ctx 的超时时间
  12.             var cancel context.CancelFunc
  13.             ctx, cancel = context.WithTimeout(ctx, timeout)
  14.             defer cancel()
  15.         }
  16.         // 中间件处理
  17.         h := func(ctx context.Context, req interface{}) (interface{}, error) {
  18.             if tr, ok := transport.FromClientContext(ctx); ok {
  19.                 header := tr.RequestHeader()
  20.                 keys := header.Keys()
  21.                 keyvals := make([]string, 0, len(keys))
  22.                 for _, k := range keys {
  23.                     keyvals = append(keyvals, k, header.Get(k))
  24.                 }
  25.                 ctx = grpcmd.AppendToOutgoingContext(ctx, keyvals...)
  26.             }
  27.             return reply, invoker(ctx, method, req, reply, cc, opts...)
  28.         }
  29.         if len(ms) > 0 {
  30.             h = middleware.Chain(ms...)(h)
  31.         }
  32.         _, err := h(ctx, req)
  33.         return err
  34.     }
  35. }

使用方式

创建客户端连接

  1. conn, err := grpc.DialInsecure(
  2.         context.Background(),
  3.         grpc.WithEndpoint("127.0.0.1:9000"),
  4.     )

使用中间件

  1. conn, err := grpc.DialInsecure(
  2.     context.Background(),
  3.     grpc.WithEndpoint("127.0.0.1:9000"),
  4.     grpc.WithTimeout(3600 * time.Second),
  5.     grpc.WithMiddleware(
  6.           recovery.Recovery(),
  7.           validate.Validator(),
  8.     ),
  9. )

使用服务发现

  1. conn, err := grpc.DialInsecure(
  2.     context.Background(),
  3.     grpc.WithEndpoint("discovery:///helloworld"),
  4.     grpc.WithDiscovery(r),
  5. )

References