gRPC
transporter/grpc 中基于谷歌的 grpc 框架实现了 Transporter，用以注册 grpc 到 eagle.Server() 中。
Server#
配置#
Network(network string) ServerOption#
配置服务端的 network 协议，如 tcp
Address(addr string) ServerOption#
配置服务端监听的地址
Timeout(timeout time.Duration) ServerOption#
配置服务端的超时设置
EnableTracing() ServerOption#
启用服务端的链路追踪
EnableLog() ServerOption#
启用服务端的日志
UnaryInterceptor(in ...grpc.UnaryServerInterceptor) ServerOption#
配置服务端使用的 grpc 单元拦截器
Options(opts ...grpc.ServerOption) ServerOption#
配置一些额外的 grpc.ServerOption
主要的实现细节#
NewServer()#
func NewServer(opts ...ServerOption) *Server { srv := &Server{ network: "tcp", address: ":0", timeout: 1 * time.Second, health: health.NewServer(), } for _, o := range opts { o(srv) } // Unary chainUnaryInterceptors := []grpc.UnaryServerInterceptor{ unaryServerInterceptor(srv), grpcPrometheus.UnaryServerInterceptor, grpcRecovery.UnaryServerInterceptor(), } if len(srv.inters) > 0 { chainUnaryInterceptors = append(chainUnaryInterceptors, srv.inters...) }
// stream chainStreamInterceptors := []grpc.StreamServerInterceptor{ grpcPrometheus.StreamServerInterceptor, grpcRecovery.StreamServerInterceptor(), }
// enable tracing if srv.enableTracing { chainUnaryInterceptors = append(chainUnaryInterceptors, otelgrpc.UnaryServerInterceptor(srv.TracerOptions...)) chainStreamInterceptors = append(chainStreamInterceptors, otelgrpc.StreamServerInterceptor(srv.TracerOptions...)) }
// enable log if srv.enableLog { chainUnaryInterceptors = append(chainUnaryInterceptors, grpcZap.UnaryServerInterceptor(logger.GetZapLogger())) chainStreamInterceptors = append(chainStreamInterceptors, grpcZap.StreamServerInterceptor(logger.GetZapLogger())) }
grpcOpts := []grpc.ServerOption{ grpc.ChainUnaryInterceptor(chainUnaryInterceptors...), grpc.ChainStreamInterceptor(chainStreamInterceptors...), } if len(srv.grpcOpts) > 0 { grpcOpts = append(grpcOpts, srv.grpcOpts...) }
grpcServer := grpc.NewServer(grpcOpts...)
// health check healthPb.RegisterHealthServer(grpcServer, srv.health)
// register reflection and the interface can be debugged through the grpcurl tool reflection.Register(grpcServer)
// set zero values for metrics registered for this grpc server grpcPrometheus.Register(grpcServer)
srv.Server = grpcServer
return srv}}使用方式#
简单列举了一些 eagle 中 grpc 的用法，其他 grpc 用法可以到 grpc 仓库中查看。
注册 grpc server#
gs := grpc.NewServer()app := eagle.New( eagle.Name("eagle"), eagle.Version("v1.0.0"), eagle.Server(gs),)Client#
配置#
WithEndpoint(endpoint string) ClientOption#
配置客户端使用的对端连接地址，如果不使用服务发现则为ip:port,如果使用服务发现则格式为discovery://\<authority>/\<serviceName>
WithTimeout(timeout time.Duration) ClientOption#
配置客户端的请求默认超时时间，如果有链路超时优先使用链路超时时间
WithMetric() ClientOption#
启用监控指标
WithLog() ClientOption#
启用日志
WithTracing() ClientOption#
启用链路追踪
WithKeepalive() ClientOption#
启用长连接
WithGzip() ClientOption#
启用日志压缩
WithoutRetry() ClientOption#
启用重试功能
WithDiscovery(d registry.Discovery) ClientOption#
配置客户端服务发现
WithUnaryInterceptor(in ...grpc.UnaryClientInterceptor) ClientOption#
配置客户端使用的 grpc 原生拦截器
WithOptions(opts ...grpc.DialOption) ClientOption#
配置一些额外的 grpc.ClientOption
主要的实现细节#
dial()#
func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.ClientConn, error) { // default client options options := clientOptions{ timeout: 2000 * time.Millisecond, balancerName: roundrobin.Name, enableGzip: true, enableMetric: true, disableRetry: false, NumRetries: 2, enableKeepalive: true, kp: keepalive.ClientParameters{ Time: 10 * time.Second, Timeout: time.Second, PermitWithoutStream: false, }, } for _, opt := range opts { opt(&options) }
// merge inters inters := []grpc.UnaryClientInterceptor{ unaryClientInterceptor(), } if len(options.inters) > 0 { inters = append(inters, options.inters...) }
// default dial option dialOpts := []grpc.DialOption{ grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"loadBalancingPolicy": "%s"}`, options.balancerName)), grpc.WithChainUnaryInterceptor(inters...), } if len(options.dialOpts) > 0 { dialOpts = append(dialOpts, options.dialOpts...) } // service discovery if options.discovery != nil { dialOpts = append(dialOpts, grpc.WithResolvers(discovery.NewBuilder( options.discovery, discovery.WithInsecure(insecure)))) } if insecure { dialOpts = append(dialOpts, grpc.WithTransportCredentials(grpcInsecure.NewCredentials())) } else { tlsConfig := &tls.Config{ InsecureSkipVerify: true, } cred := credentials.NewTLS(tlsConfig) dialOpts = append(dialOpts, grpc.WithTransportCredentials(cred)) } if options.enableKeepalive { kp := keepalive.ClientParameters{ Time: options.kp.Time, Timeout: options.kp.Timeout, PermitWithoutStream: options.kp.PermitWithoutStream, } dialOpts = append(dialOpts, grpc.WithKeepaliveParams(kp)) } if options.enableGzip { dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(gzip.Name))) } if options.enableMetric { dialOpts = append(dialOpts, grpc.WithChainUnaryInterceptor(grpcPrometheus.UnaryClientInterceptor), grpc.WithChainStreamInterceptor(grpcPrometheus.StreamClientInterceptor), ) } // enable tracing if options.enableTracing { dialOpts = append(dialOpts, grpc.WithChainUnaryInterceptor(otelgrpc.UnaryClientInterceptor()), grpc.WithChainStreamInterceptor(otelgrpc.StreamClientInterceptor()), ) } if options.enableLog { dialOpts = append(dialOpts, grpc.WithChainUnaryInterceptor(grpcZap.UnaryClientInterceptor(logger.GetZapLogger())), grpc.WithChainStreamInterceptor(grpcZap.StreamClientInterceptor(logger.GetZapLogger())), ) } if !options.disableRetry { dialOpts = append(dialOpts, grpc.WithDefaultServiceConfig(getRetryPolicy(options.balancerName, options.NumRetries)), ) }
return grpc.DialContext(ctx, options.endpoint, dialOpts...)}
使用方式#
创建客户端连接#
conn, err := grpc.DialInsecure( context.Background(), grpc.WithEndpoint("127.0.0.1:9000"), )使用服务发现#
conn, err := grpc.DialInsecure( context.Background(), grpc.WithEndpoint("discovery:///helloworld"), grpc.WithDiscovery(r),)