Grupos de Processos

Agrupe processos em grupos nomeados e faça broadcast para todos os membros em todo o cluster. Modelado no pg do Erlang/OTP: grupos são dinâmicos, um processo pode pertencer a muitos grupos e a associação é rastreada em todo o cluster via gossip.

Para o tipo de entrada de escopo e sua configuração, veja Grupos de Processos. Para o modelo de clustering mais amplo, veja o Guia de Cluster.

Carregamento

local pg = require("pg")

Abrindo um Escopo

Um grupo de processos vive dentro de um escopo — uma entrada de registro pg.scope. Abra-o para obter uma instância sobre a qual você opera:

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

Retorna: pg.Instance, error

Permissão: pg.open no id do escopo

A instância é liberada automaticamente quando o processo sai; chame release() para liberá-la antes. Todas as outras operações são métodos na instância, chamados com :.

Entrando e Saindo

local ok, err = group:join("workers")           -- grupo único
local ok, err = group:join({"workers", "all"})  -- lote
local ok, err = group:leave("workers")

Retorna: boolean, error

Um processo pode entrar no mesmo grupo mais de uma vez; deve sair o mesmo número de vezes para partir completamente (semântica multi-join). leave é best-effort em um lote e retorna erro apenas quando o processo não era membro de nenhum dos grupos nomeados.

Permissões: pg.join / pg.leave em cada nome de grupo

Listando Membros

local members, err = group:get_members("workers")        -- todos os nós
local local_members, err = group:get_local_members("workers")  -- apenas este nó

Retorna: string[], error — array de strings PID (vazio para grupo desconhecido)

Permissões: pg.get_members / pg.get_local_members no nome do grupo

Listando Grupos

local groups, err = group:which_groups()         -- todos os grupos no cluster
local local_groups, err = group:which_local_groups()  -- grupos com membro local

Retorna: string[], error — nomes de grupos que atualmente têm pelo menos um membro

Permissões: pg.which_groups / pg.which_local_groups

Broadcast

Envia uma mensagem para todos os membros de um grupo. Cada membro a recebe sob topic do processo chamador — trate com process.listen(topic).

local ok, err = group:broadcast("workers", "task", {id = 42})   -- todos os nós
local ok, err = group:broadcast_local("workers", "task", {id = 42})  -- apenas este nó

Retorna: boolean, error

Permissões: pg.broadcast / pg.broadcast_local no nome do grupo

Monitorando um Grupo

monitor inscreve-se em eventos de entrada/saída para um grupo e retorna os membros atuais atomicamente — nenhuma mudança de associação pode ocorrer entre o snapshot e a inscrição.

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

for _, pid in ipairs(members) do
    -- membros atuais no momento da inscrição
end

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

sub:close()  -- cancelar inscrição; sub:close({flush = true}) drena eventos enfileirados primeiro

Retorna: pg.Subscription, string[], error — a inscrição e um snapshot dos membros atuais

Permissão: pg.monitor no nome do grupo

Observando Todos os Grupos

events inscreve-se em mudanças de associação em todos os grupos do escopo e retorna um snapshot de todos os grupos com seus membros.

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

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

Retorna: pg.Subscription, table, error

Permissão: pg.events

Campos de Evento

Eventos entregues em um channel de inscrição contêm:

Channels de inscrição têm buffer (capacidade 64); se um consumidor lento encher o buffer, eventos adicionais para essa inscrição são descartados.

Liberando

group:release()

Libera a instância imediatamente. Idempotente; após a liberação, cada método retorna um erro. A limpeza também ocorre automaticamente quando o processo sai.

Retorna: boolean

Permissões

Erros

Veja Error Handling para trabalhar com erros.

Veja Também

• Grupos de Processos - Tipo de entrada de escopo e configuração
• Cluster - Associação e o modelo de clustering
• Gerenciamento de Processos - Criando e enviando mensagens para processos individuais