Grupos de procesos

Los grupos de procesos organizan procesos bajo nombres dinámicos y difunden mensajes a sus miembros en todo el clúster. Un proceso puede unirse a varios grupos y la membresía en todo el clúster es eventualmente consistente.

Esta es una referencia de API. Sus fragmentos presuponen un pg.scope existente, una entrada ejecutable que se ejecuta con contexto de proceso y políticas que autorizan las operaciones documentadas. Los bloques muestran llamadas individuales o flujos parciales de suscripción, no una aplicación independiente.

Para el tipo de entrada de ámbito y su configuración, consulta Grupos de procesos. Para el modelo de clustering general, consulta la Guía del clúster.

Carga

local pg = require("pg")

Añade pg a la lista modules: de la entrada ejecutable antes de requerirlo.

Apertura de un ámbito

Un grupo de procesos pertenece a un ámbito, representado por una entrada de registro pg.scope. Abre el ámbito para obtener una instancia sobre la que realizar operaciones de grupo:

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

Devuelve: pg.Instance, error

Permiso: pg.open sobre el id del ámbito

La instancia se libera automáticamente durante la limpieza del frame de ejecución. Llama a release() para liberarla antes. Las demás operaciones son métodos de la instancia y usan la sintaxis :.

Unión y salida

Las llamadas siguientes son formas independientes; elige la unión individual o por lotes que necesite la aplicación y acompáñala de las operaciones de salida correspondientes.

local ok, err = group:join("workers")           -- single group
if err then return nil, err end

local ok, err = group:join({"workers", "all"})  -- batch
if err then return nil, err end

local ok, err = group:leave("workers")
if err then return nil, err end

Devuelve: boolean, error

Un proceso puede unirse al mismo grupo varias veces y debe salir el mismo número de veces para abandonarlo por completo. En un lote, leave es de mejor esfuerzo y solo devuelve un error cuando el proceso no era miembro de ninguno de los grupos indicados.

Permisos: pg.join / pg.leave sobre cada nombre de grupo

Listado de miembros

local members, err = group:get_members("workers")        -- all nodes
if err then return nil, err end

local local_members, err = group:get_local_members("workers")  -- this node only
if err then return nil, err end

Devuelve: string[], error — un array de strings de PID (vacío para un grupo desconocido)

Permisos: pg.get_members / pg.get_local_members sobre el nombre de grupo

Listado de grupos

local groups, err = group:which_groups()         -- all groups in the cluster
if err then return nil, err end

local local_groups, err = group:which_local_groups()  -- groups with a local member
if err then return nil, err end

Devuelve: string[], error — nombres de grupos que actualmente tienen al menos un miembro

Permisos: pg.which_groups / pg.which_local_groups

Difusión

La difusión envía un mensaje desde el proceso llamador a todos los miembros del grupo bajo topic. Los miembros lo reciben con process.listen(topic).

local ok, err = group:broadcast("workers", "task", {id = 42})   -- all nodes
if err then return nil, err end

ok, err = group:broadcast_local("workers", "task", {id = 42})  -- this node only
if err then return nil, err end

Devuelve: boolean, error

Permisos: pg.broadcast / pg.broadcast_local sobre el nombre de grupo

Monitorización de un grupo

monitor se suscribe a eventos de unión y salida de un grupo y devuelve una instantánea atómica de sus miembros actuales. No puede producirse un cambio de membresía entre la instantánea y la configuración de la suscripción sin que se observe.

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

for _, pid in ipairs(members) do
    -- current members at subscription time
end

local ch = sub:channel()
local event, open = ch:receive()  -- {kind = "member.joined" | "member.left", path = "workers", data = {...}}
if not open then
    return nil, errors.new("Process-group subscription closed")
end

sub:close()  -- unsubscribe; sub:close({flush = true}) drains queued events first

Devuelve: pg.Subscription, string[], error — la suscripción y una instantánea de los miembros actuales

Permiso: pg.monitor sobre el nombre de grupo

Observación de todos los grupos

events se suscribe a cambios de membresía en todos los grupos del ámbito y devuelve una instantánea que asigna cada grupo a sus miembros.

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

local event, open = sub:channel():receive()
if not open then
    return nil, errors.new("Process-group subscription closed")
end
sub:close()

Devuelve: pg.Subscription, table, error

Permiso: pg.events

Campos de los eventos

Los eventos entregados en un canal de suscripción contienen:

Los canales de suscripción tienen búfer (capacidad 64). Si un consumidor lento llena el búfer, los eventos posteriores se conservan en orden en el buzón del proceso y se entregan cuando el consumidor vacía el canal; la suscripción se detiene en lugar de descartar eventos.

Liberar

group:release()

release libera inmediatamente la instancia y es idempotente. Después de liberarla, cualquier otra operación de grupo devuelve un error. La limpieza también se ejecuta automáticamente al finalizar el frame de ejecución.

Devuelve: boolean

Permisos

Errores

Consulta Manejo de errores para trabajar con errores.

Véase también

• Grupos de procesos - Tipo de entrada de ámbito y configuración
• Clúster - Membresía, nombres y modelo de clustering
• Gestión de procesos - Creación y mensajería de procesos individuales