进程组

将进程加入命名组并向集群中所有成员广播。以 Erlang/OTP pg 为模型：组是动态的，进程可以属于多个组，成员资格在集群范围内追踪，并具有最终一致性。

有关作用域入口类型及其配置，参见进程组。有关更广泛的集群模型，参见集群指南。

加载

local pg = require("pg")

打开作用域

进程组存在于作用域中——一个 pg.scope 注册表入口。打开它以获取操作所用的实例：

local group, err = pg.open("app:pg")
if err then
    return nil, err
end

返回: pg.Instance, error

权限: 作用域 id 上的 pg.open

进程退出时实例自动释放；调用 release() 可提前释放。所有其他操作均为实例上的方法，使用 : 调用。

加入和离开

local ok, err = group:join("workers")           -- 单个组
local ok, err = group:join({"workers", "all"})  -- 批量
local ok, err = group:leave("workers")

返回: boolean, error

进程可多次加入同一组；必须离开相同次数才能完全退出（多重加入语义）。leave 在批量操作中尽力执行，仅在进程不是任何指定组的成员时才返回错误。

权限: 每个组名上的 pg.join / pg.leave

列出成员

local members, err = group:get_members("workers")        -- 所有节点
local local_members, err = group:get_local_members("workers")  -- 仅此节点

返回: string[], error — PID 字符串数组（未知组返回空数组）

权限: 组名上的 pg.get_members / pg.get_local_members

列出组

local groups, err = group:which_groups()         -- 集群中所有组
local local_groups, err = group:which_local_groups()  -- 有本地成员的组

返回: string[], error — 当前至少有一个成员的组名

权限: pg.which_groups / pg.which_local_groups

广播

向组的所有成员发送消息。每个成员以调用进程为发送方在 topic 下接收——通过 process.listen(topic) 处理。

local ok, err = group:broadcast("workers", "task", {id = 42})   -- 所有节点
local ok, err = group:broadcast_local("workers", "task", {id = 42})  -- 仅此节点

返回: boolean, error

权限: 组名上的 pg.broadcast / pg.broadcast_local

监控组

monitor 订阅单个组的加入/离开事件，并原子性地返回当前成员——快照与订阅之间不会遗漏任何成员变更。

local sub, members, err = group:monitor("workers")
if err then
    return nil, err
end

for _, pid in ipairs(members) do
    -- 订阅时的当前成员
end

local ch = sub:channel()
local event = ch:receive()  -- {kind = "member.joined" | "member.left", path = "workers", data = {...}}

sub:close()  -- 取消订阅；sub:close({flush = true}) 先排空队列中的事件

返回: pg.Subscription, string[], error — 订阅对象和当前成员快照

权限: 组名上的 pg.monitor

监控所有组

events 订阅作用域内所有组的成员变更，并返回所有组到其成员的快照。

local sub, snapshot, err = group:events()
-- snapshot: { ["workers"] = {pid, ...}, ["all"] = {pid, ...} }

local event = sub:channel():receive()
sub:close()

返回: pg.Subscription, table, error

权限: pg.events

事件字段

订阅通道传递的事件包含：

订阅通道有缓冲（容量 64）；若消费者速度过慢填满缓冲区，该订阅的后续事件将被丢弃。

释放

group:release()

立即释放实例。幂等；释放后每个方法都返回错误。进程退出时清理也会自动运行。

返回: boolean

权限

错误

错误处理参见错误处理。

另请参阅

• 进程组 - 作用域入口类型与配置
• 集群 - 成员资格与集群模型
• 进程管理 - 启动和向单个进程发送消息