Queue-Konsumenten

Queue-Consumer liefern Nachrichten über einen konfigurierbaren Worker-Pool aus einer Queue an Funktionshandler.

Übersicht

flowchart LR
    subgraph Consumer
        QD[Queue Driver] --> DC[Delivery Channel
prefetch=10]
        DC --> WP[Worker Pool
concurrency]
        WP --> FH[Function Handler]
        FH --> AN[Ack/Nack]
    end

Konfiguration

Entry-Definition

- name: order_consumer
  kind: queue.consumer
  queue: app:orders
  func: app:process_order
  concurrency: 5
  prefetch: 20
  lifecycle:
    auto_start: true
    requires:
      - app:orders

Handler-Funktion

Die Handler-Funktion erhält den Body, nachdem der Codec der Queue ihn dekodiert hat. Mit queue.message() greifen Sie auf die aktuelle Zustellung und deren Metadaten zu:

-- process_order.lua
local queue = require("queue")
local logger = require("logger")

local function main(order)
    local msg, msg_err = queue.message()
    if msg_err then
        return nil, msg_err
    end

    logger:info("processing order", {
        message_id = msg:id(),
        order_id = order.id
    })

    return {processed = true, order_id = order.id}
end

return {main = main}

- name: process_order
  kind: function.lua
  source: file://process_order.lua
  method: main
  modules:
    - queue
    - logger

Bestätigung

Sofern der Handler die Zustellung nicht explizit abschließt, verwendet der Consumer das Ergebnis des Funktionsaufrufs:

Gewöhnliche Rückgabewerte einschließlich false wählen das Bestätigungsverhalten nicht aus. Verwenden Sie msg:ack() oder msg:nack() für einen expliziten Abschluss. Der Abschluss erfolgt genau einmal: Der erste gewinnt. Bei AMQP mit auto_ack: true bestätigt der Broker bei Zustellung; ein späterer Handlerfehler kann daher keine erneute Zustellung durch den Broker auslösen.

Der Handler kann die Nachricht mit queue.message() und msg:ack() / msg:nack() selbst quittieren; der Consumer überspringt dann sein eigenes Ack/Nack.

Worker-Pool

• Worker laufen als nebenläufige Goroutinen
• Jeder Worker verarbeitet eine Nachricht auf einmal
• Worker beziehen Nachrichten aus einem gemeinsamen Delivery-Channel. Der nächste freie Worker erhält die nächste Nachricht; Reihenfolge oder Rotation zwischen Workern ist nicht garantiert.
• Der Prefetch-Puffer erlaubt dem Treiber, Nachrichten vor der Verarbeitung zu liefern.

Beispiel

concurrency: 3
prefetch: 10

Flow:
1. Driver delivers up to 10 messages to buffer
2. 3 workers pull from buffer concurrently
3. As workers finish, buffer refills
4. Backpressure when all workers busy and buffer full

Kontrolliertes Herunterfahren

Beim Herunterfahren:

1. Keine neuen Lieferungen mehr annehmen
2. Worker-Kontexte abbrechen
3. Auf laufende Nachrichten warten (mit Timeout)
4. Timeout-Fehler zurückgeben wenn Worker nicht fertig werden

Queue-Deklaration

# Queue driver (memory for dev/test)
- name: queue_driver
  kind: queue.driver.memory
  lifecycle:
    auto_start: true

# Queue definition
- name: orders
  kind: queue.queue
  driver: app:queue_driver
  queue_name: orders        # Override name (default: entry name)
  codec: json/plain         # Payload codec (optional; json/plain is the default)
  dead_letter:              # Accepted configuration; not enforced by built-in drivers
    queue: app:dlq
    max_attempts: 5
  driver_options:
    memory:
      max_length: 10000     # Memory driver: bounded queue size

Memory-Treiber

Der integrierte In-Memory-Treiber ist für Entwicklung und Tests vorgesehen:

• Kind: queue.driver.memory
• Nachrichten im Speicher gehalten
• Nack versucht, eine Kopie der Nachricht am Ende der Queue einzureihen; dieser Versuch kann scheitern, wenn die begrenzte Queue voll ist
• Keine Persistenz über Neustarts hinweg

Siehe auch

• Message Queue — Referenz des Queue-Moduls
• Queue-Konfiguration — Queue-Treiber und Entry-Definitionen
• Supervision — Consumer-Lebenszyklus
• Prozessverwaltung — Prozessstart und Kommunikation