grpcpool

README

grpcpool

特性

• 池化对象为逻辑连接，本质是逻辑连接池。
• 支持应用层自定义参数。
  • dial func(address string) (*grpc.ClientConn, error) 创建连接的函数，同时支持配置 grpc 连接自定义参数。
  • maxIdle int 连接池内最大空闲（物理）连接数。默认初始化数量与之相同。
  • maxActive int 连接池内最大活跃（物理）连接数。0 表示无限制。
  • maxConcurrentStreams int 每个物理连接内支持的最大并发流数。
  • reuse bool 如果 maxActive 已达上限，继续获取连接时，是否继续使用池内连接。否：会创建一个一次性连接（用完即销毁）返回。
• 根据参数自动扩、缩容。
• 根据参数执行池满后获取连接的策略。

基准测试

1. 每轮并发请求共用一个连接池，每次请求从池获取一个连接：

go test -run=none -parallel=2 -bench="^BenchmarkPoolRPC" -benchtime=5000x -count=3 -benchmem
goos: linux
goarch: amd64
pkg: github.com/chengyayu/grpcpool
cpu: AMD Ryzen 7 3700U with Radeon Vega Mobile Gfx  
BenchmarkPoolRPC-8          5000           4657010 ns/op         8404197 B/op        109 allocs/op
BenchmarkPoolRPC-8          5000           4654517 ns/op         8404151 B/op        109 allocs/op
BenchmarkPoolRPC-8          5000           4642664 ns/op         8404177 B/op        109 allocs/op
PASS
ok      github.com/chengyayu/grpcpool   69.855s

1. 每个请求创建一个新连接：

go test -run=none -parallel=2 -bench="^BenchmarkSingleRPC" -benchtime=5000x -count=3 -benchmem
goos: linux
goarch: amd64
pkg: github.com/chengyayu/grpcpool
cpu: AMD Ryzen 7 3700U with Radeon Vega Mobile Gfx  
BenchmarkSingleRPC-8        5000           5077726 ns/op         8523635 B/op        691 allocs/op
BenchmarkSingleRPC-8        5000           5094132 ns/op         8523640 B/op        691 allocs/op
BenchmarkSingleRPC-8        5000           5307944 ns/op         8523638 B/op        691 allocs/op
PASS
ok      github.com/chengyayu/grpcpool   77.499s

1. 每轮并发请求共用一个连接：

go test -run=none -parallel=2 -bench="^BenchmarkOnlyOneRPC" -benchtime=5000x -count=3 -benchmem
goos: linux
goarch: amd64
pkg: github.com/chengyayu/grpcpool
cpu: AMD Ryzen 7 3700U with Radeon Vega Mobile Gfx  
BenchmarkOnlyOneRPC-8               5000           6050398 ns/op         8403255 B/op        106 allocs/op
BenchmarkOnlyOneRPC-8               5000           5979344 ns/op         8403247 B/op        106 allocs/op
BenchmarkOnlyOneRPC-8               5000           6037154 ns/op         8403248 B/op        106 allocs/op
PASS
ok      github.com/chengyayu/grpcpool   90.410s

负载均衡

虽然 GRPC 负载均衡不在本库解决范围之内，但是由于 K8S+GRPC 组合的广泛应用，且由于众所周知的原因，K8S service 无法对 GRPC 请求进行负载。 example/k8slb 也给出了一个基于 kuberesolver 的客户端负载方案 Demo，仅供参考。

特别感谢

本库基于 github.com/shimingyah/pool 修改而成。

参考资料

• https://github.com/grpc/grpc-go
• 需要每次检查连接状态吗？stackoverflow.com
• GRPC doc 连接状态变化
• GRPC doc 负载均衡机制
• GRPC 在 K8S 中的负载均衡

Documentation

Index

• Constants
• Variables
• func DftDial(address string) (*grpc.ClientConn, error)
• type Conn
• type Option
  • func Dial(factoryFn func(address string) (*grpc.ClientConn, error)) Option
  • func MaxActive(maxActive int) Option
  • func MaxConcurrentStreams(maxConcurrentStreams int) Option
  • func MaxIdle(maxIdle int) Option
  • func Reuse(reuse bool) Option
• type Pool
  • func New(address string, opts ...Option) (Pool, error)

Constants

const (
	// DialTimeout the timeout of create connection
	DialTimeout = 5 * time.Second

	// BackoffMaxDelay provided maximum delay when backing off after failed connection attempts.
	BackoffMaxDelay = 3 * time.Second

	// KeepAliveTime 如此时长后客户端未收到响应则会 ping 服务端。
	KeepAliveTime = time.Duration(10) * time.Second

	// KeepAliveTimeout 客户端 ping 服务端后，如此时长没响应会关闭连接。
	KeepAliveTimeout = time.Duration(3) * time.Second

	// InitialWindowSize we set it 1GB is to provide system's throughput.
	InitialWindowSize = 1 << 30

	// InitialConnWindowSize we set it 1GB is to provide system's throughput.
	InitialConnWindowSize = 1 << 30

	// MaxSendMsgSize set max gRPC request message size sent to server.
	// If any request message size is larger than current value, an error will be reported from gRPC.
	MaxSendMsgSize = 4 << 30

	// MaxRecvMsgSize set max gRPC receive message size received from server.
	// If any message size is larger than current value, an error will be reported from gRPC.
	MaxRecvMsgSize = 4 << 30
)

const (
	// DftMaxIdle see options.MaxIdle
	DftMaxIdle = int(8)
	// DftMaxActive see options.MaxActive
	DftMaxActive = int(64)
	// DftMaxConcurrentStreams see options.MaxConcurrentStreams
	DftMaxConcurrentStreams = int(64)
)

Variables

var ErrClosed = errors.New("pool is closed")

ErrClosed is the error resulting if the pool is closed via pool.Close().

Functions

func DftDial

func DftDial(address string) (*grpc.ClientConn, error)

DftDial return a grpc connection with defined configurations.

Types

type Conn

type Conn interface {
	// Value return the actual grpc connection type *grpc.ClientConn.
	Value() *grpc.ClientConn

	// Close decrease the reference of grpc connection, instead of close it.
	// if the pool is full, just close it.
	Close() error
}

Conn single grpc connection interface

type Option

type Option func(o *options)

Option is an options setting function.

func Dial

func Dial(factoryFn func(address string) (*grpc.ClientConn, error)) Option

Dial with factory function for *grpc.ClientConn

func MaxActive

func MaxActive(maxActive int) Option

MaxActive with pool maxActive

func MaxConcurrentStreams

func MaxConcurrentStreams(maxConcurrentStreams int) Option

MaxConcurrentStreams with pool maxConcurrentStreams

func MaxIdle

func MaxIdle(maxIdle int) Option

MaxIdle with pool maxIdle

func Reuse

func Reuse(reuse bool) Option

Reuse with pool reuse

type Pool

type Pool interface {
	// Get returns a new connection from the pool. Closing the connections puts
	// it back to the Pool. Closing it when the pool is destroyed or full will
	// be counted as an error. we guarantee the conn.Value() isn't nil when conn isn't nil.
	Get() (Conn, error)

	// Close closes the pool and all its connections. After Close() the pool is
	// no longer usable. You can't make concurrent calls Close and Get method.
	// It will be cause panic.
	Close()

	// Status returns the current status of the pool.
	Status() string
}

Pool interface describes a pool implementation. An ideal pool is thread-safe and easy to use.

func New

func New(address string, opts ...Option) (Pool, error)