group

package
v0.12.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 1, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

README

group（群组域）

三层订阅索引 + VIP 分级 + 观察者 + 分组负载——定向广播收敛的数据源。

这个域干什么

group 维护「谁在哪个组」的全部内存索引，是 messaging 域定向广播层层收敛 （appID → namespace → groupID → userID）的数据源。同时提供 VIP 分级投递、 观察者旁路、分组负载三类派生能力。

管：三层订阅索引、群组成员增删与生命周期（含系统组自动进出）、VIP 分级、 观察者（旁路监听）、分组负载（选最闲的目标）。

不管：消息投递本身（messaging 消费本域索引）、群组持久化（GroupStore 可选）。

核心概念

概念 说明
三层索引 appID → namespace → groupID → set(userID)，读多写少，无锁读
Manager 群组域门面：聚合 lifecycle / observer / vip / workload
LifecycleManager 成员进出 + 系统组（全员组）自动加入/退出 + 回调
系统组 连接建立自动加入（如 appID 全员组）；最后一条连接断开才移除身份
ObserverManager 三层观察者：namespace 级 / group 级旁路监听（监管场景）
VIPManager VIPLevel 分级：按 minLevel / exactLevel 过滤投递 + 优先级提升
WorkloadManager 分组负载：GetLeastLoaded / AcquireLeastLoaded 选最闲目标
Host 端口 ports.go 定义 Host，由 hub 实现（子包不持 *Hub）

怎么用

package main

import (
    "context"

    "github.com/kamalyes/go-wsc/group"
)

mgr := group.NewManager(deps)

// 成员管理（连接生命周期内自动驱动）
mgr.AddGroupMembers(ctx, appID, ns, "group-a", []string{"user-1001", "user-1002"})
mgr.RemoveGroupMembers(ctx, appID, ns, "group-a", []string{"user-1001"})

// 收敛查询：messaging 定向广播的底层数据源
// groupID → member userIDs → 逐个 P2P 投递

// VIP 分级投递
mgr.SendToVIPUsers(ctx, msg, group.VIPFilter{MinLevel: group.VIPLevelV3})

// 观察者（旁路）
mgr.GetObserversForMessage(ctx, msg)

// 分组负载：选最闲
target, err := mgr.AcquireLeastLoaded(ctx, appID, ns, "group-work")

端口与依赖

  • Host 端口由 hub 实现；依赖 spi 契约 GroupStore（可选持久化）、WorkloadStore（可选共享负载）。
  • 内存索引为主：广播路径上零 RTT，Redis 只在多节点共享时介入。

性能与陷阱

  • 索引读路径无锁（读多写少），但连接频繁进出时写锁是热点——批量进出合并提交。
  • 系统组移除必须引用计数：同 user 多设备时，最后一条连接断开才移除身份 （LeaveSystemGroupsKeepsIdentityWhenOtherConnectionsExist 语义）。
  • 陷阱：把群组实时落 Redis 再广播——广播路径加 1 次 RTT，收敛索引必须驻内存。

后端绑定

  • Redis（可选）：GroupStore / WorkloadStore 多节点共享。
  • 不启用时：纯内存群组，节点重启后靠重连重建索引。

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Host 

type Host interface {
	// Context 返回 Hub 生命周期上下文（用于日志关联）
	Context() context.Context
	// GetLogger 日志器
	GetLogger() spi.Logger

	// GetWorkloadStore 客服负载存储；未注入时返回 nil，调用方须报错而非 no-op
	// （负载分配静默失败会导致工单扎堆，必须让调用方感知配置缺失）
	GetWorkloadStore() spi.WorkloadStore

	// GetOnlineUsersByType 按用户类型查询在线用户（负载分配需要在线客服列表）
	// 走本地分片注册表，无 ctx 参数
	GetOnlineUsersByType(userType models.UserType) ([]string, error)

	// GetShardedRegistry 分片客户端注册表（VIP/群组判定都要遍历本地连接）。
	// 注册表是核心运行时结构而非可插拔后端，故直接用具体类型，
	// 与 hub/stats 的端口保持一致。
	GetShardedRegistry() *connection.ShardedRegistry

	// GetGroupStore 群组仓储；未注入时返回 nil，调用方须报错
	// （群组成员关系是业务语义，静默 no-op 会让业务方以为入群成功）
	GetGroupStore() spi.GroupStore

	// PublishGroupInvalidations 批量发布群拓扑失效通知到集群广播频道（拓扑写路径失效的跨节点传播）
	// 单机形态（pubsub 未部署）由实现侧短路返回 nil；事件信封封装与传输细节由编排层收口
	PublishGroupInvalidations(ctx context.Context, appID string, groupIDs []string) error

	// TrySubmitCallback 提交一个带超时保护的异步回调任务，队列满时返回 false。
	// 群组生命周期回调走此处，避免每条连接一个 goroutine。
	TrySubmitCallback(task func()) bool

	// GetGroupDisbandCallback / GetGroupMemberJoinCallback / GetGroupMemberLeaveCallback
	// 群组生命周期回调。
	//
	// 这里用**结构类型**而非 hub 包里的具名下类型（GroupDisbandCallback 等）声明：
	// 那些具名类型定义在 hub 包内，子包引用它们即形成 hub ⇄ hub/group 循环 import。
	// 结构类型与具名类型可赋值兼容，Hub 侧实现 getter 时做一次具名→结构转换。
	// 返回 nil 表示业务方未注册回调，调用方须跳过（不可假定非空）。
	GetGroupDisbandCallback() func(ctx context.Context, namespace, groupID string)
	GetGroupMemberJoinCallback() func(ctx context.Context, namespace, groupID string, userIDs []string)
	GetGroupMemberLeaveCallback() func(ctx context.Context, namespace, groupID string, userIDs []string)

	// SendConditional 条件广播：对满足条件的在线客户端各投递一份，返回投递数。
	// 条件在发送路径内对每个客户端求值，避免先在 Hub 侧物化客户端切片再发。
	SendConditional(ctx context.Context, condition func(*models.Client) bool, msg *models.HubMessage) int

	// SendToUserWithRetry 点对点发送（带重试与路由注入），失败详情在返回值中
	SendToUserWithRetry(ctx context.Context, userID string, msg *models.HubMessage) *models.SendResult
}

Host 群组域所需的 Hub 能力面。

type InvalidationBroadcaster 

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

InvalidationBroadcaster 群拓扑失效广播聚合器（一组件一文件惯例，与 observer/workload 同形态）

func NewInvalidationBroadcaster 

func NewInvalidationBroadcaster(host Host, interval time.Duration) *InvalidationBroadcaster

NewInvalidationBroadcaster 创建群拓扑失效广播聚合器 interval 传零值时使用 constants.DefaultGroupInvalidationFlushInterval

func (*InvalidationBroadcaster) Flush 

func (b *InvalidationBroadcaster) Flush(ctx context.Context)

Flush 排干全部分片：分片锁内仅做 map swap，锁外按 app 分组批量发布

func (*InvalidationBroadcaster) MarkDirty 

func (b *InvalidationBroadcaster) MarkDirty(appID, gid string)

MarkDirty 标记该群组拓扑待失效广播（GroupStore 写路径回调，聚合器唯一入口） 仅收集不发布：窗口内同 (appID, gid) 多次写合并为一条，上线风暴下广播量恒为群组数量级

func (*InvalidationBroadcaster) Run 

func (b *InvalidationBroadcaster) Run(ctx context.Context)

Run 聚合循环：每窗口 Flush 一次，ctx 结束排干残余后退出（panic 保护由 hub 启动侧兜底）

type LifecycleManager 

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

LifecycleManager 群组生命周期管理器

func NewLifecycleManager 

func NewLifecycleManager(host Host) *LifecycleManager

NewLifecycleManager 创建群组生命周期管理器

func (*LifecycleManager) AddGroupMembers 

func (m *LifecycleManager) AddGroupMembers(ctx context.Context, userIDs []string) error

AddGroupMembers 添加成员到群组（多群批量：遍历 ctx groupIDs，best-effort，返回最后错误） 群组不存在时自动创建（register 自动装配场景，无需手动 CreateGroup）；同时更新成员的反向索引 注意：手动 AddGroupMembers 不触发 OnGroupMemberJoin 回调（仅 register 自动装配触发）

func (*LifecycleManager) DisbandGroup 

func (m *LifecycleManager) DisbandGroup(ctx context.Context) error

DisbandGroup 解散群组（多群批量：遍历 ctx groupIDs，best-effort，返回最后错误） 同时清理群组元信息、成员集合、命名空间索引及各成员的反向索引；逐群触发 OnGroupDisband 回调

func (*LifecycleManager) EnsureAndJoinSystemGroup 

func (m *LifecycleManager) EnsureAndJoinSystemGroup(ctx context.Context, groupID, userID string)

EnsureAndJoinSystemGroup 确保系统组存在并加入成员 appID/namespace 从 ctx 读取（调用方 JoinSystemGroupsOnConnect/JoinMemberGroupOnConnect 已注入路由） 加入成功后触发 OnGroupMemberJoin 回调（observer/agent 自动入群也通知业务层）

func (*LifecycleManager) GetGroup 

func (m *LifecycleManager) GetGroup(ctx context.Context) (*models.Group, error)

GetGroup 获取群组元信息（appID/namespace/groupID 全从 ctx 取，groupID 取首个）

func (*LifecycleManager) GetGroupMemberCount 

func (m *LifecycleManager) GetGroupMemberCount(ctx context.Context) (int64, error)

GetGroupMemberCount 获取群组成员数量（groupID 取 ctx 首个 groupID）

func (*LifecycleManager) GetGroupMembers 

func (m *LifecycleManager) GetGroupMembers(ctx context.Context) ([]string, error)

GetGroupMembers 获取群组所有成员ID（groupID 取 ctx 首个 groupID）

func (*LifecycleManager) GetNamespaceGroups 

func (m *LifecycleManager) GetNamespaceGroups(ctx context.Context) ([]string, error)

GetNamespaceGroups 获取 ctx 的 (appID, namespace) 下所有群组ID 调用方需要「向命名空间所有群组广播」时，先调本方法取得 groupIDs， 再通过 routing.NewRoute().WithAppID(...).WithNamespace(...).Inject(ctx) 注入 ctx 后调一次 hub.Deliver。

func (*LifecycleManager) GetUserGroups 

func (m *LifecycleManager) GetUserGroups(ctx context.Context, userID string) ([]string, error)

GetUserGroups 获取用户在 ctx 的 (appID, namespace) 下加入的所有群组ID

func (*LifecycleManager) IsGroupMember 

func (m *LifecycleManager) IsGroupMember(ctx context.Context, userID string) (bool, error)

IsGroupMember 判断用户是否为群组成员（groupID 取 ctx 首个 groupID）

func (*LifecycleManager) JoinMemberGroupOnConnect 

func (m *LifecycleManager) JoinMemberGroupOnConnect(ctx context.Context, client *models.Client)

JoinMemberGroupOnConnect 客户端连接时自动加入成员组（多群组）

普通用户（非观察者）连接时遍历 client.GetGroupIDs() 逐群加入：

  • 系统保留组名（__ 前缀，如默认组 __default_gp__）走 ensureAndJoinSystemGroup（CreateGroup 拒绝 __ 前缀）
  • 业务组名走 addGroupMembersSingle（自动创建 + MaxMembers 校验）

观察者不作为成员加入群组（仅观察，不接收成员消息） GetGroupIDs 回退链保证非空（空→[DefaultGroupID]，保留单值时代加入默认组行为） 每群加入成功后触发 OnGroupMemberJoin 回调

逐群注入单群 ctx（routing.NewRoute().WithAppID(...).WithGroup(gid).Inject(ctx)），避免与 AddGroupMembers 多群 best-effort 语义叠加

func (*LifecycleManager) JoinSystemGroupsOnConnect 

func (m *LifecycleManager) JoinSystemGroupsOnConnect(ctx context.Context, client *models.Client)

JoinSystemGroupsOnConnect 客户端连接时自动加入系统保留组 注入路由（appID + namespace + 系统组 groupID）供 EnsureAndJoinSystemGroup 从 ctx 读取 namespace 用 client.Namespace 原值（全局观察者保持 ""，EnsureAndJoinSystemGroup 不归一化 namespace）

func (*LifecycleManager) LeaveSystemGroupsOnDisconnect 

func (m *LifecycleManager) LeaveSystemGroupsOnDisconnect(ctx context.Context, client *models.Client)

LeaveSystemGroupsOnDisconnect 客户端断开时自动离开系统保留组

多端登录保护：仅当该 userID 已无任何同信封在线连接时才离开系统组 调用时当前 client 已由 removeClientUnsafe（handleUnregister Phase 1）从注册表移除， 因此在线检查查询的是"移除当前连接后是否还存在其他在线连接" 竞态可接受：RemoveMembers 对不存在成员幂等，重连时 JoinSystemGroupsOnConnect 会重新加入

系统组单值，appID 归一化、namespace 保持原值（与 JoinSystemGroupsOnConnect 写入的分区一致）： 全局观察者/agent 的 Namespace 为 ""，归一化成 DefaultNamespace 会使 leave 落错分区， join 写入的成员关系永不清理（后端系统组随历史断开的观察者持续膨胀）

func (*LifecycleManager) RemoveGroupMembers 

func (m *LifecycleManager) RemoveGroupMembers(ctx context.Context, userIDs []string) error

RemoveGroupMembers 从群组移除成员（多群批量：遍历 ctx groupIDs，best-effort，返回最后错误） 同时清理成员的反向索引；逐群触发 OnGroupMemberLeave 回调

func (*LifecycleManager) TriggerGroupDisband 

func (m *LifecycleManager) TriggerGroupDisband(ctx context.Context, namespace, groupID string)

TriggerGroupDisband 异步触发群组解散回调 在 DisbandGroup 成功后逐群调用；cbCtx 注入路由供回调内 routing.AppIDFromContext 取 appID

func (*LifecycleManager) TriggerGroupMemberJoin 

func (m *LifecycleManager) TriggerGroupMemberJoin(ctx context.Context, namespace, groupID string, userIDs []string)

TriggerGroupMemberJoin 异步触发群组成员加入回调 在客户端连接时自动加群成功后调用（register 自动装配 + 系统组自动加入），手动 AddGroupMembers 不触发 复制切片避免调用方后续修改影响异步回调；cbCtx 注入路由供回调内 routing.AppIDFromContext 取 appID

func (*LifecycleManager) TriggerGroupMemberLeave 

func (m *LifecycleManager) TriggerGroupMemberLeave(ctx context.Context, namespace, groupID string, userIDs []string)

TriggerGroupMemberLeave 异步触发群组成员离开回调 在 RemoveGroupMembers 成功后逐群调用；cbCtx 注入路由供回调内 routing.AppIDFromContext 取 appID 复制切片避免调用方后续修改影响异步回调

type Manager 

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

Manager 群组域管理器

func NewManager 

func NewManager(host Host) *Manager

NewManager 创建群组域管理器

func (*Manager) Lifecycle 

func (m *Manager) Lifecycle() *LifecycleManager

Lifecycle 群组生命周期子管理器（CRUD、成员管理、系统组自动装配）

func (*Manager) Observer 

func (m *Manager) Observer() *ObserverManager

Observer 观察者查询与统计子管理器

func (*Manager) VIP 

func (m *Manager) VIP() *VIPManager

VIP 会员分级发送子管理器

func (*Manager) Workload 

func (m *Manager) Workload() *WorkloadManager

Workload 客服负载子管理器

type ObserverManager 

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

ObserverManager 观察者查询与统计管理器

func NewObserverManager 

func NewObserverManager(host Host) *ObserverManager

NewObserverManager 创建观察者管理器

func (*ObserverManager) GetObserverClients 

func (m *ObserverManager) GetObserverClients() []*models.Client

GetObserverClients 获取所有观察者客户端（所有设备）- O(n) n=观察者设备数 仅用于统计场景，消息通知走 GetObserversForMessage

func (*ObserverManager) GetObserverClientsByNamespace 

func (m *ObserverManager) GetObserverClientsByNamespace(namespace string) []*models.Client

GetObserverClientsByNamespace 获取指定命名空间的观察者客户端（兼容接口） 使用三级索引查找：全局观察者 + 命名空间级观察者（不含群组级） 群组级观察者请使用 GetObserversForMessage(namespace, groupID)

func (*ObserverManager) GetObserverCount 

func (m *ObserverManager) GetObserverCount() int

GetObserverCount 获取观察者数量（用户数，非设备数）- O(1)

func (*ObserverManager) GetObserverDeviceCount 

func (m *ObserverManager) GetObserverDeviceCount() int

GetObserverDeviceCount 获取观察者设备总数 - O(n) n=观察者数量

func (*ObserverManager) GetObserverManagerStats 

func (m *ObserverManager) GetObserverManagerStats() *models.ObserverManagerStats

GetObserverManagerStats 获取观察者管理器统计信息（按 namespace/groupID 分组）- O(n) n=观察者设备数

func (*ObserverManager) GetObserverStats 

func (m *ObserverManager) GetObserverStats() []*models.ObserverStats

GetObserverStats 获取所有观察者的统计信息 - O(n) n=观察者设备数

func (*ObserverManager) GetObserversForMessage 

func (m *ObserverManager) GetObserversForMessage(namespace string, groupIDs ...string) []*models.Client

GetObserversForMessage 获取应接收指定命名空间+群组消息的观察者 合并三级：全局 + 命名空间 + 各群组，按 clientID 去重，O(k) k=匹配的观察者数

func (*ObserverManager) IsObserver 

func (m *ObserverManager) IsObserver(userID string) bool

IsObserver 检查用户是否为观察者 - O(1)

type VIPManager 

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

VIPManager VIP 分级发送与统计管理器

func NewVIPManager 

func NewVIPManager(host Host) *VIPManager

NewVIPManager 创建 VIP 管理器

func (*VIPManager) FilterVIPClients 

func (m *VIPManager) FilterVIPClients(minLevel models.VIPLevel) []*models.Client

FilterVIPClients 筛选VIP用户客户端

func (*VIPManager) GetVIPStatistics 

func (m *VIPManager) GetVIPStatistics() map[string]int

GetVIPStatistics 获取VIP用户统计

func (*VIPManager) SendToExactVIPLevel 

func (m *VIPManager) SendToExactVIPLevel(ctx context.Context, vipLevel models.VIPLevel, msg *models.HubMessage) int

SendToExactVIPLevel 发送消息给指定VIP等级用户

func (*VIPManager) SendToUserWithClassification 

func (m *VIPManager) SendToUserWithClassification(ctx context.Context, userID string, msg *models.HubMessage, classification *models.MessageClassification)

SendToUserWithClassification 使用完整分类系统发送消息 发送结果通过 OnMessageSend 回调通知

func (*VIPManager) SendToVIPUsers 

func (m *VIPManager) SendToVIPUsers(ctx context.Context, minVIPLevel models.VIPLevel, msg *models.HubMessage) int

SendToVIPUsers 发送消息给指定VIP等级及以上的用户

func (*VIPManager) SendToVIPWithPriority 

func (m *VIPManager) SendToVIPWithPriority(ctx context.Context, vipLevel models.VIPLevel, msg *models.HubMessage) int

SendToVIPWithPriority 根据VIP等级优先发送

func (*VIPManager) SendWithVIPPriority 

func (m *VIPManager) SendWithVIPPriority(ctx context.Context, userID string, msg *models.HubMessage)

SendWithVIPPriority 根据用户VIP等级自动设置消息优先级 使用 ForEachUserClientFiltered 零拷贝遍历（按 ctx 信封过滤）+ 提前终止，替代 GetUserClients + 手动迭代

func (*VIPManager) UpgradeVIPLevel 

func (m *VIPManager) UpgradeVIPLevel(ctx context.Context, userID string, newLevel models.VIPLevel) bool

UpgradeVIPLevel 升级用户VIP等级（按 ctx 路由信封 appID+namespace 隔离） 使用 ForEachUserClientFiltered 零拷贝遍历（仅升级匹配信封的客户端）+ SetVIPLevel 原子更新，消除数据竞争

type WorkloadManager 

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

WorkloadManager 客服负载管理器

func NewWorkloadManager 

func NewWorkloadManager(host Host) *WorkloadManager

NewWorkloadManager 创建客服负载管理器

func (*WorkloadManager) AcquireLeastLoadedAgent 

func (m *WorkloadManager) AcquireLeastLoadedAgent(ctx context.Context, onlineAgents []string, dimension models.WorkloadDimension) (string, int64, error)

AcquireLeastLoadedAgent 原子地选择负载最小的在线客服并将其负载 +1（分布式安全）

与 GetLeastLoadedAgent 的区别：在同一个 Redis Lua 中完成"选中 + 多维度原子 +1"， 多副本/多 goroutine 并发场景下不会因"读-改-写"窗口读到同一份旧快照而选中同一客服

参数 onlineAgents 由调用方传入（可以是过滤后的"可接单"列表）；若传入空切片， 函数会自动回退到调用 GetOnlineUsersByType(UserTypeAgent) 获取全部在线客服

业务失败回滚：调用方应在后续业务失败时显式调用 DecrementAgentWorkload 回滚此次预扣减

func (*WorkloadManager) DecrementAgentWorkload 

func (m *WorkloadManager) DecrementAgentWorkload(ctx context.Context, agentID string) error

DecrementAgentWorkload 减少客服工作负载

func (*WorkloadManager) ForceSetAgentWorkload 

func (m *WorkloadManager) ForceSetAgentWorkload(ctx context.Context, agentID string, workload int64) error

ForceSetAgentWorkload 强制设置客服工作负载（慎用）

func (*WorkloadManager) GetAgentWorkload 

func (m *WorkloadManager) GetAgentWorkload(ctx context.Context, agentID string) (int64, error)

GetAgentWorkload 获取客服工作负载

func (*WorkloadManager) GetAllAgentWorkloads 

func (m *WorkloadManager) GetAllAgentWorkloads(ctx context.Context, limit int64) ([]models.WorkloadInfo, error)

GetAllAgentWorkloads 获取所有客服的负载信息

func (*WorkloadManager) GetLeastLoadedAgent 

func (m *WorkloadManager) GetLeastLoadedAgent(ctx context.Context, dimension models.WorkloadDimension) (string, int64, error)

GetLeastLoadedAgent 获取负载最小的在线客服

func (*WorkloadManager) IncrementAgentWorkload 

func (m *WorkloadManager) IncrementAgentWorkload(ctx context.Context, agentID string) error

IncrementAgentWorkload 增加客服工作负载

func (*WorkloadManager) ReloadAgentWorkload 

func (m *WorkloadManager) ReloadAgentWorkload(ctx context.Context, agentID string) (int64, error)

ReloadAgentWorkload 重新加载客服工作负载（客服上线时调用）

func (*WorkloadManager) RemoveAgentWorkload 

func (m *WorkloadManager) RemoveAgentWorkload(ctx context.Context, agentID string) error

RemoveAgentWorkload 移除客服工作负载

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.