temp

Server/common/discover/common/define.go

@@ -0,0 +1,25 @@
+package common
+
+type ListenerType int32
+
+const (
+	ListenerTypeNewServer   = 1 // 服务启动
+	ListenerTypeCloseServer = 2 // 服务关闭
+)
+
+var (
+	KeyDiscover        = "discover"
+	KeyDiscoverService = KeyDiscover + "/service"
+)
+
+var (
+	KeyDiscoverGateway  = KeyDiscoverService + "/gateway"
+	KeyDiscoverDatabase = KeyDiscoverService + "/database"
+)
+
+// ServiceProvider 服务提供者
+type ServiceProvider struct {
+	Target string
+	SID    string
+	Addr   string
+}

Server/common/discover/common/tool.go

@@ -0,0 +1,28 @@
+package common
+
+import (
+	"common/db/etcd"
+	"common/log"
+	"context"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+func NewLeaseAndKeepAlive(ttl int64) (clientv3.LeaseID, error) {
+	lease, err := etcd.Client().Grant(context.Background(), ttl)
+	if err != nil {
+		return 0, err
+	}
+	chKeepAlive, err := etcd.Client().KeepAlive(context.Background(), lease.ID)
+	if err != nil {
+		return 0, err
+	}
+	go func() {
+		for r := range chKeepAlive {
+			if r == nil {
+				log.Errorf("lease timeout!")
+				return
+			}
+		}
+	}()
+	return lease.ID, nil
+}

Server/common/discover/grpc_client/conn.go

@@ -0,0 +1,41 @@
+package grpc_client
+
+import (
+	"common/log"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
+	"time"
+)
+
+type GrpcConnection struct {
+	sid  string
+	conn *grpc.ClientConn
+}
+
+func NewGrpcConnection(sid, address string) (*GrpcConnection, error) {
+	p := &GrpcConnection{
+		sid: sid,
+	}
+	conn, err := grpc.Dial(
+		address,
+		grpc.WithInsecure(),
+		grpc.WithKeepaliveParams(
+			keepalive.ClientParameters{
+				Time:                30 * time.Second, // 保活探测包发送的时间间隔
+				Timeout:             10 * time.Second, // 保活探测包的超时时间
+				PermitWithoutStream: true,
+			},
+		),
+		//grpc.WithStatsHandler(&StatsHandler{}),
+	)
+	if err != nil {
+		log.Errorf("create grpc err: %v, sid: %v, addr: %v", err, sid, address)
+		return nil, err
+	}
+	p.conn = conn
+	return p, nil
+}
+
+func (g *GrpcConnection) GetConnection() *grpc.ClientConn {
+	return g.conn
+}

Server/common/discover/grpc_client/conn_mgr.go

@@ -0,0 +1,62 @@
+package grpc_client
+
+import (
+	"common/log"
+	"fmt"
+	"google.golang.org/grpc"
+	"math/rand"
+)
+
+type GrpcConnectionMgr struct {
+	poolM map[string]*GrpcConnection
+	poolS []*GrpcConnection
+}
+
+func NewGrpcConnectionMgr() *GrpcConnectionMgr {
+	return &GrpcConnectionMgr{
+		poolM: make(map[string]*GrpcConnection),
+		poolS: make([]*GrpcConnection, 0),
+	}
+}
+
+func (p *GrpcConnectionMgr) Store(sid, addr string) {
+	pool, err := NewGrpcConnection(sid, addr)
+	if err != nil {
+		log.Errorf("create grpc err: %v, sid: %v, addr: %v", err, sid, addr)
+		return
+	}
+	p.poolM[sid] = pool
+	p.poolS = append(p.poolS, pool)
+}
+
+func (p *GrpcConnectionMgr) Delete(sid string) int {
+	delete(p.poolM, sid)
+	for i, pool := range p.poolS {
+		if pool.sid == sid {
+			p.poolS = append(p.poolS[:i], p.poolS[i+1:]...)
+			break
+		}
+	}
+	return len(p.poolS)
+}
+
+func (p *GrpcConnectionMgr) Load(sid ...string) (*grpc.ClientConn, error) {
+	var pool *GrpcConnection
+	if len(sid) > 0 && len(sid[0]) > 0 {
+		pool = p.poolM[sid[0]]
+	} else {
+		pool = p.poolS[rand.Intn(len(p.poolS))]
+	}
+	if pool == nil {
+		return nil, fmt.Errorf("cannot find connection")
+	}
+	return pool.GetConnection(), nil
+}
+
+func (p *GrpcConnectionMgr) LoadAll() map[string]*grpc.ClientConn {
+	sidM := make(map[string]*grpc.ClientConn)
+	for sid, pool := range p.poolM {
+		sidM[sid] = pool.GetConnection()
+	}
+	return sidM
+}

Server/common/discover/grpc_client/stats.go

@@ -0,0 +1,45 @@
+package grpc_client
+
+import (
+	"context"
+	"fmt"
+	"google.golang.org/grpc/stats"
+)
+
+// 1. 实现 stats.Handler 接口
+type StatsHandler struct{}
+
+func (h *StatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
+	// 给 RPC 调用打标签（例如记录方法名）
+	return context.WithValue(ctx, "rpc_method", info.FullMethodName)
+}
+
+func (h *StatsHandler) HandleRPC(ctx context.Context, s stats.RPCStats) {
+	// 处理 RPC 统计信息
+	switch t := s.(type) {
+	case *stats.Begin:
+		fmt.Printf("RPC started: %s\n", ctx.Value("rpc_method"))
+	case *stats.End:
+		fmt.Printf("RPC finished: %s (duration: %v)\n",
+			ctx.Value("rpc_method"), t.EndTime.Sub(t.BeginTime))
+	case *stats.InPayload:
+		fmt.Printf("Received %d bytes\n", t.WireLength)
+	case *stats.OutPayload:
+		fmt.Printf("Sent %d bytes\n", t.WireLength)
+	}
+}
+
+func (h *StatsHandler) TagConn(ctx context.Context, info *stats.ConnTagInfo) context.Context {
+	// 给连接打标签
+	return ctx
+}
+
+func (h *StatsHandler) HandleConn(ctx context.Context, s stats.ConnStats) {
+	// 处理连接事件
+	switch s.(type) {
+	case *stats.ConnBegin:
+		fmt.Println("Connection established")
+	case *stats.ConnEnd:
+		fmt.Println("Connection closed")
+	}
+}

Server/common/discover/listener.go

@@ -0,0 +1,92 @@
+package discover
+
+import (
+	"common/db/etcd"
+	"common/discover/common"
+	"context"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"strings"
+	"sync"
+)
+
+var (
+	wg         = &sync.WaitGroup{}
+	listenerMU = &sync.RWMutex{}
+	listener   = make(map[common.ListenerType][]func(data any))
+	stopFunc   context.CancelFunc
+)
+
+// RegisterListener 注册服务变动监听
+func RegisterListener(t common.ListenerType, cb func(data any)) {
+	listenerMU.Lock()
+	defer listenerMU.Unlock()
+	arr := listener[t]
+	if arr == nil {
+		arr = make([]func(data any), 0)
+	}
+	arr = append(arr, cb)
+	listener[t] = arr
+}
+
+// 根据类型触发回调
+func onCBByType(t common.ListenerType, data any) {
+	listenerMU.RLock()
+	defer listenerMU.RUnlock()
+	for _, f := range listener[t] {
+		f(data)
+	}
+}
+
+func Listen() {
+	var stopCtx context.Context
+	stopCtx, stopFunc = context.WithCancel(context.Background())
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// 服务
+		serviceAll, _ := etcd.Client().Get(stopCtx, common.KeyDiscoverService, clientv3.WithPrefix())
+		for _, kv := range serviceAll.Kvs {
+			onServerChange(clientv3.EventTypePut, string(kv.Key), string(kv.Value))
+		}
+		chService := etcd.Client().Watch(stopCtx, common.KeyDiscoverService, clientv3.WithPrefix(), clientv3.WithRev(serviceAll.Header.Revision+1))
+		for {
+			select {
+			case msg := <-chService:
+				for _, event := range msg.Events {
+					onServerChange(event.Type, string(event.Kv.Key), string(event.Kv.Value))
+				}
+			case <-stopCtx.Done():
+				return
+			}
+		}
+	}()
+}
+
+// 服务发生变化
+func onServerChange(t mvccpb.Event_EventType, key, value string) {
+	split := strings.Split(key, "/")
+	if len(split) != 5 {
+		return
+	}
+	switch t {
+	case clientv3.EventTypePut:
+		onCBByType(common.ListenerTypeNewServer, &common.ServiceProvider{
+			Target: common.KeyDiscoverService + "/" + split[2],
+			SID:    split[3],
+			Addr:   value,
+		})
+	case clientv3.EventTypeDelete:
+		onCBByType(common.ListenerTypeCloseServer, &common.ServiceProvider{
+			Target: common.KeyDiscoverService + "/" + split[2],
+			SID:    split[3],
+		})
+	}
+}
+
+func Close() {
+	if stopFunc != nil {
+		stopFunc()
+		wg.Wait()
+	}
+}

Server/common/discover/server.go

@@ -0,0 +1,101 @@
+package discover
+
+import (
+	"common/db/etcd"
+	"common/discover/common"
+	"common/discover/grpc_client"
+	"common/log"
+	"context"
+	"fmt"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"google.golang.org/grpc"
+	"sync"
+)
+
+// 大量读少量写的情况下，读写锁比同步Map更高效
+var (
+	serverMU     = sync.RWMutex{}
+	conn         = make(map[string]*grpc_client.GrpcConnectionMgr)
+	serverLeaseM = make(map[string]clientv3.LeaseID)
+)
+
+func init() {
+	RegisterListener(common.ListenerTypeNewServer, onServerStart)
+	RegisterListener(common.ListenerTypeCloseServer, onServerStop)
+}
+
+// FindServer 根据SID或随机查找服务
+func FindServer(target string, sid ...string) (*grpc.ClientConn, error) {
+	serverMU.RLock()
+	defer serverMU.RUnlock()
+	if v, ok := conn[target]; ok {
+		return v.Load(sid...)
+	}
+	return nil, fmt.Errorf("cannot find server")
+}
+
+func FindServerAll(target string) map[string]*grpc.ClientConn {
+	serverMU.RLock()
+	defer serverMU.RUnlock()
+	if v, ok := conn[target]; ok {
+		return v.LoadAll()
+	}
+	return make(map[string]*grpc.ClientConn)
+}
+
+// RegisterGrpcServer 注册服务提供者
+func RegisterGrpcServer(target, sid, addr string, ttl int64) error {
+	serverMU.Lock()
+	defer serverMU.Unlock()
+	leaseID, err := common.NewLeaseAndKeepAlive(ttl)
+	if err != nil {
+		return err
+	}
+	_, err = etcd.Client().Put(context.Background(), target+"/"+sid, addr, clientv3.WithLease(leaseID))
+	if err != nil {
+		return err
+	}
+	serverLeaseM[sid] = leaseID
+	return nil
+}
+
+// UnRegisterGrpcServer 解注册服务提供者
+func UnRegisterGrpcServer(sid string) {
+	serverMU.Lock()
+	defer serverMU.Unlock()
+	if leaseID, ok := serverLeaseM[sid]; ok {
+		_, err := etcd.Client().Revoke(context.Background(), leaseID)
+		if err != nil {
+			log.Errorf("server.go UnRegisterGrpcServer err: %v", err)
+		}
+		delete(serverLeaseM, sid)
+	}
+}
+
+// 某个服务启动了
+func onServerStart(data any) {
+	if provider, ok := data.(*common.ServiceProvider); ok {
+		serverMU.Lock()
+		defer serverMU.Unlock()
+		if v, ok := conn[provider.Target]; ok {
+			v.Store(provider.SID, provider.Addr)
+		} else {
+			mgr := grpc_client.NewGrpcConnectionMgr()
+			mgr.Store(provider.SID, provider.Addr)
+			conn[provider.Target] = mgr
+		}
+	}
+}
+
+// 某个服务关闭了
+func onServerStop(data any) {
+	if provider, ok := data.(*common.ServiceProvider); ok {
+		serverMU.Lock()
+		defer serverMU.Unlock()
+		if v, ok := conn[provider.Target]; ok {
+			if v.Delete(provider.SID) == 0 {
+				delete(conn, provider.Target)
+			}
+		}
+	}
+}

Server/common/discover/service/client_gateway.go

@@ -0,0 +1,24 @@
+package service
+
+import (
+	"common/discover"
+	"common/discover/common"
+	"common/discover/service/game/game_pb"
+)
+
+func GatewayNewClient(sid ...string) (game_pb.GameClient, error) {
+	c, err := discover.FindServer(common.KeyDiscoverGateway, sid...)
+	if err != nil {
+		return nil, err
+	}
+	return game_pb.NewGameClient(c), nil
+}
+
+func GatewayNewBroadcastClient() map[string]game_pb.GameClient {
+	clientM := make(map[string]game_pb.GameClient)
+	connM := discover.FindServerAll(common.KeyDiscoverGateway)
+	for sid, conn := range connM {
+		clientM[sid] = game_pb.NewGameClient(conn)
+	}
+	return clientM
+}

Server/common/discover/service/service.go

@@ -0,0 +1,94 @@
+package service
+
+import (
+	"common/discover"
+	"common/discover/common"
+	"common/log"
+	"common/utils"
+	"context"
+	"fmt"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+)
+
+type IService interface {
+	Init(addr string)
+	Close()
+}
+
+type ServiceBase struct {
+	Target  string
+	SID     string
+	Serve   *grpc.Server
+	EtcdTTL int64
+	OnInit  func(serve *grpc.Server)
+	OnClose func()
+
+	wg *sync.WaitGroup
+}
+
+func (s *ServiceBase) Init(addr string) {
+	s.wg = &sync.WaitGroup{}
+	s.wg.Add(1)
+	s.SID = utils.SnowflakeInstance().Generate().String()
+	go func() {
+		defer s.wg.Done()
+		defer s.OnClose()
+		defer discover.UnRegisterGrpcServer(s.SID)
+
+		pMin, _ := strconv.ParseInt(os.Getenv("P_MIN"), 10, 64)
+		pMax, _ := strconv.ParseInt(os.Getenv("P_MAX"), 10, 64)
+		if pMin == 0 || pMax == 0 || pMin > pMax {
+			log.Errorf(" %v init err: pMin or pMax is 0 or pMin > pMax", s.Target)
+			return
+		}
+
+		// 服务注册
+		lis, port, err := common.ListenRandomPort(pMin, pMax)
+		if lis == nil {
+			log.Errorf(" %v ListenRandomPort err: %v", s.Target, err)
+			return
+		}
+
+		s.Serve = grpc.NewServer(
+			grpc.UnaryInterceptor(
+				func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+					defer func() {
+						if r := recover(); r != nil {
+							log.Errorf("server Panic: %v", r)
+						}
+					}()
+					resp, err = handler(ctx, req)
+					return
+				},
+			),
+			grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
+				MinTime:             20 * time.Second,
+				PermitWithoutStream: true,
+			}),
+		)
+		s.OnInit(s.Serve)
+
+		// 服务注册
+		if err = discover.RegisterGrpcServer(s.Target, s.SID, fmt.Sprintf("%v:%d", addr, port), s.EtcdTTL); err != nil {
+			log.Errorf("%v RegisterGrpcServer err: %v", s.Target, err)
+			return
+		}
+		if err = s.Serve.Serve(lis); err != nil {
+			log.Errorf(" %v Serve err: %v", s.Target, err)
+			return
+		}
+		log.Infof("%v server stop.", s.Target)
+	}()
+}
+
+func (s *ServiceBase) Close() {
+	if s.Serve != nil {
+		s.Serve.Stop()
+		s.wg.Wait()
+	}
+}