Prozessgruppen

Prozessgruppen organisieren Prozesse unter dynamischen Namen und senden Nachrichten an Gruppenmitglieder im gesamten Cluster. Ein Prozess kann mehreren Gruppen beitreten; die clusterweite Mitgliedschaft ist letztlich konsistent.

Diese Seite ist eine API-Referenz. Ihre Snippets setzen einen vorhandenen pg.scope, einen ausführbaren Eintrag mit Prozesskontext und Richtlinien voraus, die die dokumentierten Operationen erlauben. Die Blöcke demonstrieren einzelne Aufrufe oder partielle Abonnementabläufe und keine eigenständige Anwendung.

Den Scope-Entry-Kind und seine Konfiguration beschreibt Prozessgruppen. Das umfassendere Clustering-Modell behandelt der Cluster-Leitfaden.

Laden

local pg = require("pg")

Fügen Sie pg zur modules:-Liste des ausführbaren Eintrags hinzu, bevor Sie das Modul laden.

Einen Scope öffnen

Eine Prozessgruppe gehört zu einem Scope, der durch einen Registry-Eintrag vom Typ pg.scope dargestellt wird. Öffnen Sie den Scope, um eine Instanz für Gruppenoperationen zu erhalten:

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

Rückgabewerte: pg.Instance, error

Berechtigung: pg.open auf der Scope-id

Die Instanz wird bei der Bereinigung des Ausführungsframes automatisch freigegeben. Mit release() geben Sie sie früher frei. Alle anderen Operationen sind Methoden der Instanz und verwenden die Syntax :.

Beitreten und Verlassen

Die folgenden Aufrufe sind unabhängige Formen. Wählen Sie den Einzelgruppen- oder Batch-Beitritt, den die Anwendung benötigt, und kombinieren Sie ihn mit den entsprechenden Leave-Operationen.

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

Rückgabewerte: boolean, error

Ein Prozess kann derselben Gruppe mehrfach beitreten und muss sie ebenso oft verlassen, um vollständig auszuscheiden. Bei einem Batch arbeitet leave nach Best Effort und gibt nur dann einen Fehler zurück, wenn der Prozess keiner der genannten Gruppen angehörte.

Berechtigungen: pg.join / pg.leave auf jedem Gruppennamen

Mitglieder auflisten

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

Rückgabewerte: string[], error — ein Array von PID-Strings; bei einer unbekannten Gruppe leer

Berechtigungen: pg.get_members / pg.get_local_members auf dem Gruppennamen

Gruppen auflisten

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

Rückgabewerte: string[], error — Gruppennamen, die aktuell mindestens ein Mitglied haben

Berechtigungen: pg.which_groups / pg.which_local_groups

Broadcasts

Ein Broadcast sendet eine Nachricht vom aufrufenden Prozess unter topic an jedes Gruppenmitglied. Mitglieder empfangen sie mit 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

Rückgabewerte: boolean, error

Berechtigungen: pg.broadcast / pg.broadcast_local auf dem Gruppennamen

Eine Gruppe überwachen

monitor abonniert Beitritts- und Austrittsereignisse einer Gruppe und gibt einen atomaren Snapshot ihrer aktuellen Mitglieder zurück. Zwischen Snapshot und Einrichtung des Abonnements kann keine Mitgliedschaftsänderung unbeobachtet bleiben.

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

Rückgabewerte: pg.Subscription, string[], error — das Abonnement und ein Snapshot der aktuellen Mitglieder

Berechtigung: pg.monitor auf dem Gruppennamen

Alle Gruppen beobachten

events abonniert Mitgliedschaftsänderungen jeder Gruppe im Scope und gibt einen Snapshot zurück, der Gruppen ihren Mitgliedern zuordnet.

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()

Rückgabewerte: pg.Subscription, table, error

Berechtigung: pg.events

Ereignisfelder

Über den Channel eines Abonnements gelieferte Ereignisse enthalten:

Abonnement-Channels sind gepuffert (Kapazität 64). Füllt ein langsamer Consumer den Puffer, bleiben weitere Ereignisse geordnet in der Prozess-Mailbox und werden zugestellt, sobald der Consumer den Channel leert. Das Abonnement hält also an, statt Ereignisse zu verwerfen.

Freigeben

group:release()

release gibt die Instanz sofort frei und ist idempotent. Nach der Freigabe gibt jede andere Gruppenoperation einen Fehler zurück. Die Bereinigung erfolgt außerdem automatisch am Ende des Ausführungsframes.

Rückgabewert: boolean

Berechtigungen

Fehler

Siehe Fehlerbehandlung für den Umgang mit Fehlern.

Siehe auch

• Prozessgruppen - Scope-Entry-Kind und Konfiguration
• Cluster - Mitgliedschaft, Benennung und Clustering-Modell
• Prozessverwaltung - Einzelne Prozesse starten und Nachrichten senden