Nachrichten-Queue

Veröffentlichen und Konsumieren von Nachrichten aus verteilten Queues. Unterstützt mehrere Backends einschließlich RabbitMQ und andere AMQP-kompatible Broker.

Für Queue-Konfiguration siehe Queue.

Laden

local queue = require("queue")

Nachrichten veröffentlichen

Senden Sie Nachrichten an eine Queue per ID:

local ok, err = queue.publish("app:tasks", {
    action = "send_email",
    user_id = 456,
    template = "welcome"
})
if err then
    return nil, err
end

Parameter	Typ	Beschreibung
`queue_id`	string	Queue-Identifikator (Format: “namespace:name”)
`data`	any	Nachrichtendaten (Tables, Strings, Zahlen, Booleans)
`headers`	table	Optionale Nachrichten-Header

Gibt zurück: boolean, error

Nachrichten-Header

Header ermöglichen Routing, Priorisierung und Tracing:

queue.publish("app:notifications", {
    type = "order_shipped",
    order_id = order.id
}, {
    priority = "high",
    correlation_id = request_id
})

Zugriff auf Zustellungskontext

Innerhalb eines Queue-Consumers auf die aktuelle Nachricht zugreifen:

local msg, err = queue.message()
if err then
    return nil, err
end

local msg_id = msg:id()
local priority = msg:header("priority")
local all_headers = msg:headers()

Gibt zurück: Message, error

Nur verfügbar beim Verarbeiten von Queue-Nachrichten im Consumer-Kontext.

Nachrichten-Methoden

Methode	Gibt zurück	Beschreibung
`id()`	`string, error`	Eindeutiger Nachrichten-Identifikator
`header(key)`	`any, error`	Einzelner Header-Wert (nil wenn fehlend)
`headers()`	`table, error`	Alle Nachrichten-Header
`ack()`	`boolean, error`	Verarbeitung bestaetigen (single-shot)
`nack()`	`boolean, error`	Fehlschlag fuer Redelivery oder Dead-Letter melden (single-shot)

Die Runtime fuehrt bei Handler-Erfolg automatisch ack aus und bei Handler-Fehler automatisch nack. Rufen Sie ack/nack nur auf, um frueher abzuschliessen.

Queue-Info

local stats, err = queue.info("app:tasks")
-- stats kann enthalten: message_count, consumer_count, ready (treiberabhaengig)

Gibt zurueck: table, error

Consumer-Muster

Queue-Consumer werden als Entry-Points definiert, die den Payload direkt empfangen:

entries:
  - kind: queue.consumer
    id: email_worker
    queue: app:emails
    method: handle_email

function handle_email(payload)
    local msg = queue.message()

    logger:info("Processing", {
        message_id = msg:id(),
        to = payload.to
    })

    local ok, err = email.send(payload.to, payload.template, payload.data)
    if err then
        return nil, err  -- Nachricht wird erneut eingereiht oder dead-lettered
    end
end

Berechtigungen

Queue-Operationen unterliegen der Sicherheitsrichtlinienauswertung.

Aktion	Ressource	Beschreibung
`queue.publish`	-	Allgemeine Berechtigung zum Veröffentlichen von Nachrichten
`queue.publish.queue`	Queue-ID	Zu spezifischer Queue veröffentlichen

Beide Berechtigungen werden geprüft: zuerst die allgemeine Berechtigung, dann die queue-spezifische.

Fehler

Bedingung	Art	Wiederholbar
Queue-ID leer	`errors.INVALID`	nein
Nachrichtendaten leer	`errors.INVALID`	nein
Kein Zustellungskontext	`errors.INVALID`	nein
Veröffentlichung nicht erlaubt	`errors.INVALID`	nein
Veröffentlichung fehlgeschlagen	`errors.INTERNAL`	nein

Siehe Fehlerbehandlung für die Arbeit mit Fehlern.

Siehe auch

Queue-Konfiguration - Queue-Treiber und Entry-Definitionen
Queue-Consumer-Leitfaden - Consumer-Muster und Worker-Pools
Prozessverwaltung - Prozess-Spawning und Kommunikation
Channels - Inter-Prozess-Kommunikationsmuster
Funktionen - Asynchrone Funktionsaufrufe