cluster

package

v0.12.1 Latest Latest Go to latest Published: Nov 1, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/HitGameAI/apex-wsc

Links

Open Source Insights

README ¶

cluster（分布式域）

节点发现 / 跨节点投递 / 自愈 / 防乒乓——多 Pod 对等组网，无选主

这个域干什么

cluster 让多个 go-wsc 节点（K8s 里就是多个 Pod）表现得像一个逻辑服务：谁在线、某用户在哪台节点、消息怎么跨节点送达、节点挂了谁接管、路由索引错了怎么修正

跨节点投递的路由信封经 gRPC metadata 传播（routing.InjectToOutgoingMetadata / routing.RestoreFromIncomingMetadata，由独立 routing 包提供）——本域投递链路只调用，不拥有

核心概念

节点发现（Redis Hash + 心跳租约，REFACTORING_DESIGN §八.1）

概念	说明
`NodeRegistry`	`HSET nodes:grpc <nodeID> <addr>` 注册；`HDEL` 优雅摘除（SIGTERM 联动）
心跳租约	每 5s `HSET nodes:heartbeat` + `EXPIRE`；TTL 过半未续摘流量，耗尽剔除
迟滞恢复	被摘节点需连续 N 周期正常续租才恢复接流，防 P17 乒乓

跨节点投递

主路径：gRPC 直连（GRPCClientPool + per-node 熔断）
兜底 1：Redis PubSub（最后通道，尽力而为）

概念	说明
`GRPCClientPool`	连接池 + 每节点独立熔断（半开探测恢复）
投递语义	at-least-once，幂等由业务 messageID 保证

自愈四件套

机制	触发 → 动作
死索引修正（self_heal）	投递被拒 → 修正 user→node 映射 + 回源重投
幽灵连接回收	路由指向已剔除节点 → rerouteGuard 换节点 + 清理死索引
ACK 超时兜底（node_ack_timeout）	跨节点 ACK 超时 → 标记 + 转离线 + 在线状态核对
重路由防乒乓（reroute_guard）	消息级拒绝集合：拒绝过的节点不再投，超限转离线

ping-pong 死循环场景：节点 A、B 互为对方死索引的兜底 → 无守卫时无限互弹； rerouteGuard 用消息级拒绝集合切断环路

怎么用

package main

import (
    "github.com/kamalyes/go-wsc/cluster"
)

// 节点注册（hub.Run 内部自动执行，这里示意手动用法）
nr := cluster.NewNodeRegistry(nodeID, grpcAddr, redisClient)
nr.Register(ctx)
defer nr.Unregister(ctx)   // 优雅退出时主动摘除

// gRPC 客户端池（跨节点直连投递）
pool := cluster.NewGRPCClientPool(opts...)

端口与依赖

依赖 spi 契约：NodeRegistry（发现）、MessageQueue（NATS 兜底）
gRPC 依赖在适配器仓库（go-wsc-grpc-adapter）：核心仓库 proto/ 仅生成期使用
K8s：Pod IP 经 Downward API 注入，直连走 Pod 网络，不用 Headless Service（§八.4）

性能与陷阱

P11：路由抖动 → 投错 → 重投 → 死循环；rerouteGuard 是唯一防线
P17：租约误判——TTL 不能小于两次续期间隔 ×3；摘除要过半 + 迟滞恢复双保险
熔断 per-node 隔离：单节点故障不能熔断整个池

后端绑定

Redis（一等）：节点发现租约
NATS（一等）：兜底投递通道（JetStream 持久化 + Explicit ack）
不启用时：单节点模式，cluster 域空转（零连接、零租约）

Documentation ¶

Index ¶

func FromProtoDispatchOperation(op wscpb.DispatchOperation) models.OperationType
func ToProtoDispatchOperation(op models.OperationType) wscpb.DispatchOperation
type ClusterDispatchOptions
type ClusterOperation
type DispatchItem
type DispatchOutcome
type GRPCBatchDispatcher
- func NewGRPCBatchDispatcher(pool *GRPCClientPool, window time.Duration, maxBatch int) *GRPCBatchDispatcher
type GRPCClientPool
- func NewGRPCClientPool() *GRPCClientPool
type NodeRegistry
- func NewNodeRegistry(redisClient redis.UniversalClient, ...) *NodeRegistry

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func FromProtoDispatchOperation ¶

func FromProtoDispatchOperation(op wscpb.DispatchOperation) models.OperationType

FromProtoDispatchOperation 将 proto DispatchOperation 枚举映射回领域操作类型与 ToProtoDispatchOperation 互逆

func ToProtoDispatchOperation ¶

func ToProtoDispatchOperation(op models.OperationType) wscpb.DispatchOperation

ToProtoDispatchOperation 将领域操作类型映射为 proto DispatchOperation 枚举双向转换的唯一实现（服务端 FromProtoDispatchOperation 反向映射），避免两处散落 switch

Types ¶

type ClusterDispatchOptions ¶

type ClusterDispatchOptions struct {
	Operation     ClusterOperation // 操作类型（SendMessage/GroupsBroadcast/Broadcast/ObserverNotify/KickUser）
	AppID         string           // 应用ID（最上层隔离维度，空=全局共享）
	Namespace     string           // 命名空间ID（空="default"；Broadcast 时空表示全命名空间）
	TargetNodeID  string           // 目标节点ID（精确路由，空=所有已知节点广播）
	TargetNodeIDs []string         // 已知目标节点列表（P2P 跨节点路由用，gRPC 未启用时优先定向 PubSub 而非广播频道）
	TargetUserID  string           // 目标用户ID（Operation=SendMessage 时使用）
	GroupIDs      []string         // 群组ID列表（len==1 单群组广播，len>1 批量广播，len==0 不广播）
	ExcludeSender bool             // 是否排除发送者（群组广播时使用）
	SenderID      string           // 发送者ID（排除发送者时使用）
	Reason        string           // 辅助信息（踢人原因等）
}

ClusterDispatchOptions 跨节点分发选项封装所有跨节点通信的路由参数，由统一的跨节点路由入口消费

type ClusterOperation ¶

type ClusterOperation = models.OperationType

ClusterOperation 集群操作类型（统一别名，消除 DistributedMessage 历史命名歧义）

type DispatchItem ¶

type DispatchItem struct {
	Operation     models.OperationType
	AppID         string
	Namespace     string
	GroupIDs      []string
	TargetUserID  string
	Reason        string
	MessageData   []byte
	ExcludeSender bool
	SenderID      string
	TraceID       string
}

DispatchItem 单条跨节点投递指令（微批合帧最小单元）路由信封内嵌，不依赖 gRPC metadata

type DispatchOutcome ¶

type DispatchOutcome int

DispatchOutcome 单条投递结果分类（与 hub 内 grpcDispatchOutcome 语义对齐）

const (
	// OutcomeDelivered 投递成功
	OutcomeDelivered DispatchOutcome = iota
	// OutcomeFallback 调用失败或结果未知 → 交由 PubSub 兜底
	OutcomeFallback
	// OutcomeUserMiss 目标节点明确用户不在 → 跳过 PubSub 兜底
	OutcomeUserMiss
	// OutcomeRejected 用户在线但目标节点拒绝 → 不降级 PubSub 兜底、保留在线索引
	OutcomeRejected
)

func NewGRPCBatchDispatcher ¶

func NewGRPCBatchDispatcher(pool *GRPCClientPool, window time.Duration, maxBatch int) *GRPCBatchDispatcher

NewGRPCBatchDispatcher 创建跨节点 gRPC 微批分派器 window 为批量窗口（<=0 表示禁用微批）；maxBatch<=0 时默认不设批上限（仅靠窗口）

func (*GRPCBatchDispatcher) Close ¶

func (d *GRPCBatchDispatcher) Close()

Close 关闭分派器：排空所有缓冲区并拒绝后续提交（优雅停机）

func (*GRPCBatchDispatcher) Enabled ¶

func (d *GRPCBatchDispatcher) Enabled() bool

Enabled 是否启用微批（窗口 > 0 且已注入连接池）

func (*GRPCBatchDispatcher) Submit ¶

func (d *GRPCBatchDispatcher) Submit(ctx context.Context, addr string, item DispatchItem) DispatchOutcome

Submit 提交单条投递指令并同步等待结果返回该指令的投递分类；调用方（dispatchViaGRPC）据此决定 PubSub 兜底

type GRPCClientPool ¶

type GRPCClientPool struct {
	// contains filtered or unexported fields
}

GRPCClientPool gRPC 客户端连接池，管理到其他节点的连接使用 sync.Map 按 addr 缓存连接，LoadOrStore 避免并发重复创建

func (*GRPCClientPool) BroadcastGroup ¶

func (p *GRPCClientPool) BroadcastGroup(ctx context.Context, addr string, msgData []byte, excludeSender bool, senderID string) (int32, error)

BroadcastGroup 向指定节点的群组成员广播消息 namespace/groupID 从 ctx 提取并注入 gRPC metadata 跨节点传播带熔断保护：广播失败（节点不可达）不应阻塞调用方

func (*GRPCClientPool) CheckUsersOnline ¶

func (p *GRPCClientPool) CheckUsersOnline(ctx context.Context, addr string, userIDs []string) (map[string]bool, error)

CheckUsersOnline 批量检查用户是否在指定节点在线带熔断保护：读操作也需保护，避免故障节点拖慢批量查询

func (*GRPCClientPool) Close ¶

func (p *GRPCClientPool) Close()

Close 关闭所有连接并清空连接池

func (*GRPCClientPool) ConnectionCount ¶

func (p *GRPCClientPool) ConnectionCount() int

ConnectionCount 返回连接池中缓存的连接数（观测/健康检查用）

func (*GRPCClientPool) GetBreakerStats ¶

func (p *GRPCClientPool) GetBreakerStats(addr string) breaker.CircuitStats

GetBreakerStats 获取指定节点的熔断器统计信息（用于监控/运维）

func (*GRPCClientPool) GetClient ¶

func (p *GRPCClientPool) GetClient(addr string) (wscpb.NodeServiceClient, error)

GetClient 获取或创建到指定地址的 gRPC 连接同一地址复用同一连接，避免重复建连

func (*GRPCClientPool) IsCircuitOpen ¶

func (p *GRPCClientPool) IsCircuitOpen(addr string) bool

IsCircuitOpen 判断指定节点的熔断器是否处于开启状态（快速短路）

func (*GRPCClientPool) KickUser ¶

func (p *GRPCClientPool) KickUser(ctx context.Context, addr, userID, reason string) (*wscpb.KickUserResponse, error)

KickUser 向指定节点发送踢人指令路由信封（appID+namespace）经 gRPC metadata 传播，远端按信封隔离踢出同名用户连接带熔断保护：故障节点快速失败，由上层 PubSub 兜底补踢

func (*GRPCClientPool) NotifyObservers ¶

func (p *GRPCClientPool) NotifyObservers(ctx context.Context, addr string, msgData []byte) (int32, error)

NotifyObservers 通知指定节点的观察者 namespace/groupID 从 ctx 提取并注入 gRPC metadata 跨节点传播带熔断保护：observer 通知失败不应阻塞主流程

func (*GRPCClientPool) Ping ¶

func (p *GRPCClientPool) Ping(ctx context.Context, addr string) (*wscpb.PingResponse, error)

Ping 对指定节点进行健康检查不经过熔断器——Ping 是探测节点是否恢复的手段，即使在半开状态也需直接调用

func (*GRPCClientPool) Remove ¶

func (p *GRPCClientPool) Remove(addr string)

Remove 关闭并移除指定地址的连接与熔断器（节点下线/滚动发布时回收死连接）不回收则历史 addr 的 ClientConn 随发布次数无界累积——每条含后台重连 backoff 协程与传输缓冲，addr 失效后永不再被路由命中与 GetClient 的 LoadOrStore 竞态由「删除后重建」自愈：节点实际仍存活时下次 GetClient 重新建连，仅付出一次重连开销；在途 RPC 不受影响（grpc ClientConn.Close 优雅排空，仅拒绝新调用）

func (*GRPCClientPool) SendToUser ¶

func (p *GRPCClientPool) SendToUser(ctx context.Context, addr, userID string, msgData []byte) (*wscpb.SendToUserResponse, error)

SendToUser 向指定节点的用户发送消息带熔断保护：连续失败后自动熔断，避免向故障节点持续发请求

func (*GRPCClientPool) SetBreakerConfig ¶

func (p *GRPCClientPool) SetBreakerConfig(config breaker.Config)

SetBreakerConfig 设置自定义熔断器配置（用于测试或生产调优）必须在首次调用任何 gRPC 方法前设置，否则已创建的熔断器不受影响；原子替换：与 getOrCreateBreaker 并发安全（运行期调优不构成数据竞争）

type NodeRegistry ¶

type NodeRegistry struct {
	// contains filtered or unexported fields
}

NodeRegistry 管理节点发现与 gRPC 地址映射

func NewNodeRegistry ¶

func NewNodeRegistry(redisClient redis.UniversalClient, nodeID, grpcAddr, grpcKey, heartbeatKey string, logger spi.Logger) *NodeRegistry

NewNodeRegistry 创建节点注册中心 grpcKey/heartbeatKey 来自 go-config Cluster 配置，默认 "wsc:nodes:grpc" / "wsc:nodes:heartbeat"

func (*NodeRegistry) GetAllNodes ¶

func (r *NodeRegistry) GetAllNodes() map[string]string

GetAllNodes 获取所有已知节点（不含本节点）

func (*NodeRegistry) GetNodeAddr ¶

func (r *NodeRegistry) GetNodeAddr(nodeID string) (string, bool)

GetNodeAddr 获取指定节点的 gRPC 地址

func (*NodeRegistry) OnNodeEvicted ¶

func (r *NodeRegistry) OnNodeEvicted(fn func(nodeID, addr string))

OnNodeEvicted 注册节点淘汰回调（nil 清除回调）回调在 refreshNodes 剔除死节点时同步触发（refreshLoop goroutine 内），不得执行阻塞操作；addr 为该节点被剔除时的最后已知地址持 mu 写入：与 refreshLoop 的 notifyEvicted 读并发安全

func (*NodeRegistry) Register ¶

func (r *NodeRegistry) Register(ctx context.Context) error

Register 注册本节点到 Redis 并启动定期刷新

func (*NodeRegistry) SetGRPCAddr ¶

func (r *NodeRegistry) SetGRPCAddr(addr string)

SetGRPCAddr 回填本节点对外的 gRPC 监听地址配置端口为 0（随机端口）时，监听器绑定后才能得知实际端口，必须在 Register 之前回填，否则注册进发现表的是无效地址

func (*NodeRegistry) SetNodeAddr ¶

func (r *NodeRegistry) SetNodeAddr(nodeID, addr string)

SetNodeAddr 手动登记某节点的 gRPC 地址

与 Redis 自动发现互补：适用于静态节点列表部署，或在测试中绕过 Redis 发现刷新延迟直接建立节点映射

func (*NodeRegistry) Stop ¶

func (r *NodeRegistry) Stop()

Stop 停止节点注册中心

func (*NodeRegistry) Unregister ¶

func (r *NodeRegistry) Unregister(ctx context.Context) error

Unregister 注销本节点

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL