キュー

キューシステムは、非同期メッセージのパブリッシャー、ドライバー、キュー、コンシューマー、ハンドラー関数を接続します。

このページは設定および動作のリファレンスです。YAML のコードブロックは、完全なドキュメントを示す場合を除き、既存のエントリリストに配置する断片です。外部ドライバーの例では、ブローカーまたは AWS 互換サービスがすでに存在することを前提としています。

アーキテクチャ

flowchart LR
    P[Publisher] --> D[Driver]
    D --> Q[Queue]
    Q --> C[Consumer]
    C --> W[Worker Pool]
    W --> F[Function]

• ドライバー — バックエンド実装（memory、AMQP、SQS）
• キュー — ドライバーに関連付けられた論理キュー
• コンシューマー — 並行処理設定を使ってキューとハンドラーを接続
• ワーカープール — メッセージを並行処理

複数のキューで 1 つのドライバーを共有できます。複数のコンシューマーで同じキューを処理できます。

エントリ種別

ドライバー設定

メモリドライバー

インプロセスドライバーは開発環境および単一ノードのデプロイ向けで、外部依存関係はありません。

- name: memory_driver
  kind: queue.driver.memory
  lifecycle:
    auto_start: true

AMQP ドライバー

RabbitMQ および AMQP 0-9-1 互換ブローカー用です。

- name: amqp_driver
  kind: queue.driver.amqp
  url: "amqp://guest:guest@localhost:5672/"
  vhost: "/"
  connection_name: "wippy-service"
  heartbeat: "10s"
  connection_timeout: "30s"
  reconnect_delay: "1s"
  reconnect_max_delay: "30s"
  default_message_ttl: "1h"
  default_queue_expiry: "24h"
  prefetch_count: 10
  lifecycle:
    auto_start: true

tls の下に TLS を設定します。

  tls:
    enabled: true
    server_name: "rabbit.example.com"
    cert: ${env:app.env:amqp_cert}
    key:  ${env:app.env:amqp_key}
    ca:   ${env:app.env:amqp_ca}
    insecure_skip_verify: false

cert/key/ca は PEM コンテンツを保持します。インライン、file:// 経由、または env レジストリで解決される ${env:NAME} プレースホルダ経由で指定できます。insecure_skip_verify は証明書検証を無効化します（開発用のみ）。従来の cert_env/key_env/ca_env ディレクティブも同じように解決されますが非推奨です。${env:NAME} を使用してください。

SQS ドライバー

AWS SQS および SQS 互換エンドポイント（LocalStack、ElasticMQ）用です。認証情報、リージョン、その他の AWS SDK 設定は、共有の config.aws リソースから取得されます。

- name: aws_config
  kind: config.aws
  region: us-east-1
  access_key_id: ${env:app:AWS_ACCESS_KEY_ID}
  secret_access_key: ${env:app:AWS_SECRET_ACCESS_KEY}

- name: sqs_driver
  kind: queue.driver.sqs
  config: app:aws_config
  endpoint: "http://localhost:9324"
  message_retention_period: 86400
  default_delay_seconds: 0
  lifecycle:
    auto_start: true

キューは初回使用時にドライバによって自動作成されます。発行時に SQS 固有のフィールドを指定するには SQS プレフィックス付きヘッダを使用してください。sqs.delay_seconds、sqs.message_group_id、sqs.message_deduplication_id は型付きの SQS メッセージフィールドにマッピングされます。それ以外のヘッダ（correlation_id や content_type のような中立的なキー、および任意の sqs.message_attributes.* キー）は SQS メッセージ属性としてそのまま運ばれます。

キュー設定 {id="queue-configuration"}

- name: tasks
  kind: queue.queue
  driver: app.queue:memory_driver
  codec: json/plain
  queue_name: "app_tasks"
  driver_options:
    memory:
      max_length: 500
  dead_letter:
    queue: app.queue:tasks_dlq
    max_attempts: 5

ドライバーオプション

driver_options のキーはドライバー名でスコープされます。ドライバーは自身のサブバッグだけを読み取ります。他のキーは休眠状態となるため、必要に応じて単一のキューエントリで複数のドライバーの設定を宣言できます。

memory:

amqp:

コーデック {id="codecs"}

codec は、メッセージ本文をブローカーに渡す前のシリアライズ方法を選択します。ペイロード形式を表す文字列で、デフォルトは json/plain です。

AMQP ドライバーは、発行するメッセージに対応する content-type（application/json または application/msgpack）を設定します。不明なコーデックは発行時ではなく、キューの宣言時に失敗します。

コンシューマー設定

- name: task_consumer
  kind: queue.consumer
  queue: app.queue:tasks
  func: app.queue:task_handler
  concurrency: 4
  prefetch: 20
  auto_ack: false
  driver_options:
    amqp:
      consumer_tag: "worker-1"
      exclusive: false
  lifecycle:
    auto_start: true
    requires:
      - app.queue:tasks

amqp コンシューマーオプション:

ワーカープール

ワーカーは並行して実行されます。

concurrency: 3, prefetch: 10

1. Driver delivers up to 10 messages to the shared buffer
2. 3 workers pull from the buffer and can each hold an active delivery
3. As workers finish, buffer refills
4. Backpressure when all workers busy and buffer full

ハンドラー関数

コンシューマーハンドラーは、デコード済みのメッセージ本文を最初の引数として受け取ります。配信メタデータ（id、headers）には queue.message() でアクセスします。

local queue = require("queue")
local logger = require("logger")

local function main(body)
    local msg, msg_err = queue.message()
    if msg_err then return nil, msg_err end
    local message_id, id_err = msg:id()
    if id_err then return nil, id_err end
    local correlation_id, header_err = msg:header("correlation_id")
    if header_err then return nil, header_err end

    logger:info("processing", {
        id = message_id,
        correlation_id = correlation_id
    })

    local ok, err = process_task(body)
    if err then
        return nil, err  -- nack: ドライバに応じて再配信
    end
    return true          -- ack: remove from queue
end

return { main = main }

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

確認応答

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

false を含む通常の戻り値では、確認応答の動作は選択されません。明示的に確定するには msg:ack() または msg:nack() を呼び出します。確定は一度だけ行われ、最初に到達した呼び出しが優先されます。

Dead-Letter ルーティング

Dead-letter ルーティングはまだ実装されていません。dead_letter ブロック（キュー設定を参照）は設定として受け付けられますが、現時点で試行回数を数え、nack されたメッセージを設定済みの DLQ にルーティングし、x_dead_letter_* ヘッダを設定する組み込みドライバはありません。nack されたメッセージはドライバ自身のポリシーに従って再配信されます。x_* ヘッダ名前空間は将来の DLQ 記録用に予約されているため、発行者は x_* ヘッダを設定しないでください。

メッセージの発行

Lua コードから発行します。

local queue = require("queue")

local published, publish_err = queue.publish("app.queue:tasks", {
    id = "task-123",
    action = "process",
    data = payload
})
if publish_err then return nil, publish_err end
return published

Lua の発行 API とメッセージ API については、キューモジュールを参照してください。

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

コンシューマーの停止時には次の処理を行います。

1. 新しい配信の受け付けを停止
2. ワーカーコンテキストをキャンセル
3. 処理中のメッセージを待機（タイムアウトあり）
4. ワーカーが時間内に終了しない場合はエラーを返す

関連項目

• キューモジュール - Lua API リファレンス
• キューコンシューマーガイド - コンシューマーのパターンとワーカープール
• スーパービジョン - コンシューマーのライフサイクル管理