From ee61d70419c979459becccb3c4c29aff59b3df5f Mon Sep 17 00:00:00 2001 From: tangmingyou Date: Fri, 26 Jan 2024 22:38:41 +0800 Subject: [PATCH] grpc keepalive --- Dockerfile | 6 +-- cmd/auth/config.toml | 7 ++++ cmd/chat/config.toml | 7 ++++ cmd/gateway_ws/config.toml | 8 ++++ cmd/mahjong/config.toml | 7 ++++ .../gateway_http/logic/postal_balancer.go | 1 + internal/gateway_ws/dao/group.go | 25 ++++++++++++ pkg/config/config.go | 13 ++++--- pkg/config/grpc_options.go | 39 ++++++++++++++----- 9 files changed, 95 insertions(+), 18 deletions(-) create mode 100644 internal/gateway_ws/dao/group.go diff --git a/Dockerfile b/Dockerfile index dc42578..ae3820d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,9 +8,9 @@ WORKDIR /opt COPY target/${APP}/${APP} /opt/app COPY target/${APP}/config/ /opt/config/ +# cp /etc/localtime . +COPY localtime /etc/localtime -RUN chmod +x /opt/app \ - && ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \ - && echo 'Asia/Shanghai' > /etc/timezone +RUN chmod +x /opt/app CMD ["./app", "-conf=config/config.toml"] diff --git a/cmd/auth/config.toml b/cmd/auth/config.toml index 40e9de9..fadb3fa 100644 --- a/cmd/auth/config.toml +++ b/cmd/auth/config.toml @@ -11,6 +11,13 @@ writeBufferSize = "8Ki" [grpc.register.attrs] weight = 10 +[grpc.keepalive] +idleTimeout = "60s" +forceCloseWait = "20s" +keepAliveInterval = "60s" +keepAliveTimeout = "20s" +maxLifeTime = "2h" + [etcd] endpoints = ["124.222.131.236:3279"] username = "root" diff --git a/cmd/chat/config.toml b/cmd/chat/config.toml index 212ca6e..fe07e5d 100644 --- a/cmd/chat/config.toml +++ b/cmd/chat/config.toml @@ -10,6 +10,13 @@ writeBufferSize = "8Ki" [grpc.register.attrs] weight = 10 +[grpc.keepalive] +idleTimeout = "60s" +forceCloseWait = "20s" +keepAliveInterval = "60s" +keepAliveTimeout = "20s" +maxLifeTime = "2h" + [etcd] endpoints = ["124.222.131.236:3279"] username = "root" diff --git a/cmd/gateway_ws/config.toml b/cmd/gateway_ws/config.toml index f8509b2..3e0bed5 100644 --- a/cmd/gateway_ws/config.toml +++ b/cmd/gateway_ws/config.toml @@ -15,6 +15,13 @@ writeBufferSize = "8Ki" [grpc.register.attrs] weight = 10 +[grpc.keepalive] +idleTimeout = "60s" +forceCloseWait = "20s" +keepAliveInterval = "60s" +keepAliveTimeout = "20s" +maxLifeTime = "2h" + [etcd] endpoints = ["124.222.131.236:3279"] username = "root" @@ -28,6 +35,7 @@ MinIdleConns = 3 [nats] Url = "nats://nats.sopod@124.222.131.236:3222" +RetryOnFailedConnect = true [gorm] logMode=true diff --git a/cmd/mahjong/config.toml b/cmd/mahjong/config.toml index f3bd407..f8b41d8 100644 --- a/cmd/mahjong/config.toml +++ b/cmd/mahjong/config.toml @@ -10,6 +10,13 @@ writeBufferSize = "8Ki" [grpc.register.attrs] weight = 10 +[grpc.keepalive] +idleTimeout = "60s" +forceCloseWait = "20s" +keepAliveInterval = "60s" +keepAliveTimeout = "20s" +maxLifeTime = "2h" + [etcd] endpoints = ["124.222.131.236:3279"] username = "root" diff --git a/internal/gateway_http/logic/postal_balancer.go b/internal/gateway_http/logic/postal_balancer.go index 760bc8d..6456e46 100644 --- a/internal/gateway_http/logic/postal_balancer.go +++ b/internal/gateway_http/logic/postal_balancer.go @@ -40,6 +40,7 @@ func (h *PostalBalancer) Init(ctx context.Context, resolver resolver.Builder) { h.postalClient = postal.NewPostalClient(conn) } +// Endpoint 根据consistent hash负载均衡策略调用到 postal 集群的一个节点，拿到节点的客户端连接地址 func (h *PostalBalancer) Endpoint(c *gin.Context) { subject, err := config.GetSubject(c) if err != nil { diff --git a/internal/gateway_ws/dao/group.go b/internal/gateway_ws/dao/group.go new file mode 100644 index 0000000..bf8b86d --- /dev/null +++ b/internal/gateway_ws/dao/group.go @@ -0,0 +1,25 @@ +package dao + +import ( + "errors" + "github.com/redis/go-redis/v9" +) + +var ErrNotExists = errors.New("not exists") + +// PostalDao persistent postal group api +type PostalDao interface { + LoadGroupIdsByUid(uid string) ([]string, error) + GroupCreate(gid string) (string, error) + GroupJoin(gid, uid string) error + GroupLeave(gid, uid string) error + GroupDismiss(gid string) error +} + +type RedisPostalDao struct { + rdb *redis.Client +} + +func (d *RedisPostalDao) LoadGroupIdsByUid(uid string) (gids []string, err error) { + return +} diff --git a/pkg/config/config.go b/pkg/config/config.go index f52bfba..53ae9ca 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -13,7 +13,6 @@ import ( type Configuration struct { App map[string]any Grpc GrpcConfig - GrpcClient GrpcClientConfig Gorm GormConfig Snowflake SnowflakeConfig Etcd clientv3.Config @@ -30,13 +29,15 @@ type GrpcConfig struct { ReadBufferSize string WriteBufferSize string Register discovery.Server + keepalive GrpcKeepalive } -type GrpcKeepaliveConfig struct { - // todo... -} - -type GrpcClientConfig struct { +type GrpcKeepalive struct { + IdleTimeout string + ForceCloseWait string + KeepAliveInterval string + KeepAliveTimeout string + MaxLifeTime string } type GormConfig struct { diff --git a/pkg/config/grpc_options.go b/pkg/config/grpc_options.go index c87991d..3b43c22 100644 --- a/pkg/config/grpc_options.go +++ b/pkg/config/grpc_options.go @@ -2,22 +2,43 @@ package config import ( "google.golang.org/grpc" + "google.golang.org/grpc/keepalive" "sonet/pkg/utils/conver" ) -func GetGrpcOptions(config GrpcConfig, customOpts ...grpc.ServerOption) (opts []grpc.ServerOption) { - if config.MaxSendMsgSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(config.MaxSendMsgSize))) +func GetGrpcOptions(c GrpcConfig, customOpts ...grpc.ServerOption) (opts []grpc.ServerOption) { + if c.MaxSendMsgSize != "" { + opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.MaxSendMsgSize))) } - if config.MaxRecvMsgSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(config.MaxRecvMsgSize))) + if c.MaxRecvMsgSize != "" { + opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.MaxRecvMsgSize))) } - if config.ReadBufferSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(config.ReadBufferSize))) + if c.ReadBufferSize != "" { + opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.ReadBufferSize))) } - if config.WriteBufferSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(config.WriteBufferSize))) + if c.WriteBufferSize != "" { + opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.WriteBufferSize))) } + + // grpc server keepalive + keep := keepalive.ServerParameters{} + if c.keepalive.IdleTimeout != "" { + keep.MaxConnectionIdle = conver.MustParseDuration(c.keepalive.IdleTimeout) + } + if c.keepalive.ForceCloseWait != "" { + keep.MaxConnectionAgeGrace = conver.MustParseDuration(c.keepalive.ForceCloseWait) + } + if c.keepalive.KeepAliveInterval != "" { + keep.Time = conver.MustParseDuration(c.keepalive.KeepAliveInterval) + } + if c.keepalive.KeepAliveTimeout != "" { + keep.Timeout = conver.MustParseDuration(c.keepalive.KeepAliveTimeout) + } + if c.keepalive.MaxLifeTime != "" { + keep.MaxConnectionAge = conver.MustParseDuration(c.keepalive.MaxLifeTime) + } + opts = append(opts, grpc.KeepaliveParams(keep)) + opts = append(opts, customOpts...) return }