キューコンシューマ

キューコンシューマは、設定可能なワーカープールを介して、キューから関数ハンドラへメッセージを配信します。

概要

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

設定

エントリ定義

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

ハンドラ関数

ハンドラ関数は、キューのコーデックによるデコード後のボディを受け取ります。現在の配信とそのメタデータへアクセスするには queue.message() を使用します。

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

確認応答

ハンドラが配信を明示的に確定しない限り、コンシューマは関数呼び出しの結果を使用します。

false を含む通常の戻り値は、確認応答の動作を選択しません。明示的に確定するには msg:ack() または msg:nack() を呼び出します。確定は1回限りで、最初の確定が優先されます。AMQPで auto_ack: true の場合、ブローカーが配信時に確認応答するため、その後ハンドラが失敗してもブローカーによる再配信は発生しません。

ハンドラはqueue.message()とmsg:ack() / msg:nack()でメッセージ自身を確定できます。その場合、コンシューマは自身のack/nackをスキップします。

ワーカープール

• ワーカーは並行goroutineとして実行されます。
• 各ワーカーは一度に1つのメッセージを処理します。
• ワーカーは共有配信チャネルから取得します。次の空きワーカーが次のメッセージを受け取り、ワーカー間での順序やローテーションは保証されません。
• プリフェッチバッファにより、ドライバは処理に先行してメッセージを配信できます。

例

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

グレースフルシャットダウン

シャットダウン時、コンシューマは次の処理を行います。

1. 新しい配信の受け入れを停止
2. ワーカーコンテキストをキャンセル
3. 停止タイムアウトまで、処理中のハンドラを待機
4. ワーカーが終了しない場合はタイムアウトエラーを返す

キュー宣言

# 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

メモリドライバ

組み込みのインメモリドライバは、開発およびテスト用です。

• kindは queue.driver.memory です。
• メッセージはメモリに保存されます。
• Nackは、複製したメッセージをキュー末尾へ再エンキューしようとします。有界キューが満杯の場合、この試行は失敗することがあります。
• メッセージは再起動をまたいで永続化されません。

関連項目

• メッセージキュー — キューモジュールのリファレンス
• キュー設定 — キュードライバとエントリ定義
• スーパービジョン — コンシューマのライフサイクル
• プロセス管理 — プロセスの生成と通信