Queue

Wippy 提供队列系统，用于异步消息处理，支持可配置的驱动和消费者。

架构

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

• Driver - 后端实现（memory、AMQP、SQS）
• Queue - 绑定到驱动的逻辑队列
• Consumer - 连接队列到处理器，带并发设置
• Worker Pool - 并发消息处理器

多个队列可以共享一个驱动。多个消费者可以从同一队列处理消息。

Entry 类型

驱动配置

内存驱动

用于开发和单节点部署的进程内驱动。无外部依赖。

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

AMQP 驱动

用于 RabbitMQ 和 AMQP 0-9-1 兼容的 broker。

- 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:
    enabled: true
    server_name: "rabbit.example.com"
    cert_env: "AMQP_CLIENT_CERT"
    key_env: "AMQP_CLIENT_KEY"
    ca_env: "AMQP_CA_CERT"
    insecure_skip_verify: false

内联 cert/key/ca 字段携带 PEM 内容；*_env 变体通过 env registry 解析。两种来源在每个字段上互斥。insecure_skip_verify 禁用证书验证（仅用于开发）。

SQS 驱动

用于 AWS SQS 和 SQS 兼容的 endpoint（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: 345600
  default_delay_seconds: 0
  lifecycle:
    auto_start: true

队列在首次使用时由驱动自动创建。在发布时使用 SQS 前缀的 header（sqs.*）来寻址 SQS 特定属性；像 correlation_id 和 content_type 这样的中性键在可能的情况下会被翻译为 SQS 系统属性。

队列配置

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

消费者配置

- 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
    depends_on:
      - app.queue:tasks

amqp 消费者选项：

Worker Pool

Worker 作为并发 goroutine 运行：

concurrency: 3, prefetch: 10

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

处理函数

消费者处理器以解码后的消息体作为第一个参数。使用 queue.message() 访问投递元数据（id、headers）。

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

local function main(body)
    local msg = queue.message()
    logger:info("processing", {
        id = msg:id(),
        correlation_id = msg:header("correlation_id")
    })

    local ok, err = process_task(body)
    if err then
        return false  -- nack: redelivery or DLQ
    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

确认机制

Runtime 根据处理器返回值自动 settle：

仅在需要提前 settle 时显式调用 msg:ack() 或 msg:nack()。Settlement 是单次的：先到达的调用获胜。

Dead-Letter 路由

当队列上配置了 dead_letter 时，nack 超过 max_attempts 的消息会被路由到 DLQ，驱动会设置 x_dead_letter_reason 和 x_original_queue header。发布者不得设置任何 x_* header——这些保留给 DLQ 簿记使用。

发布消息

从 Lua 代码发布：

local queue = require("queue")

queue.publish("app.queue:tasks", {
    id = "task-123",
    action = "process",
    data = payload
})

参见 Queue 模块 了解完整 API。

优雅关闭

消费者停止时：

1. 停止接收新消息
2. 取消 worker 上下文
3. 等待正在处理的消息（有超时）
4. 如果 worker 未及时完成则返回错误

参见

• Queue 模块 - Lua API 参考
• Queue 消费者指南 - 消费者模式和 worker 池
• Supervision - 消费者生命周期管理