큐

큐 시스템은 비동기 메시지 발행자, 드라이버, 큐, 컨슈머 및 핸들러 함수를 연결합니다.

이 페이지는 설정 및 동작 레퍼런스입니다. 완전한 문서를 보여 주지 않는 YAML 펜스는 기존 엔트리 목록용 조각이며, 외부 드라이버 예제는 브로커나 AWS 호환 서비스가 이미 존재한다고 가정합니다.

아키텍처

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

• 드라이버 - 백엔드 구현 (memory, AMQP, SQS)
• 큐 - 드라이버에 바인딩된 논리적 큐
• 컨슈머 - 동시성 설정으로 큐를 핸들러에 연결
• 워커 풀 - 동시 메시지 프로세서

여러 큐가 드라이버를 공유할 수 있습니다. 여러 컨슈머가 같은 큐에서 처리할 수 있습니다.

엔트리 종류

드라이버 설정

메모리 드라이버

개발 및 단일 노드 배포용 인프로세스 드라이버. 외부 의존성 없음.

- 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 메시지 속성으로 그대로 전달됩니다.

큐 설정

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

코덱

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

확인 응답

핸들러가 명시적으로 settle하지 않으면 컨슈머는 함수 호출 결과에 따라 settle합니다:

false를 포함한 일반 반환 값은 확인 응답 동작을 선택하지 않습니다. msg:ack() 또는 msg:nack()을 호출해 명시적으로 settle하세요. Settlement는 단일 실행이며 먼저 도착한 호출이 우선합니다.

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는 Queue 모듈을 참조하세요.

정상 종료

컨슈머 중지 시:

1. 새 메시지 수신 중지
2. 워커 컨텍스트 취소
3. 처리 중인 메시지 완료 대기(타임아웃 적용)
4. 워커가 제시간에 완료되지 않으면 오류 반환

참고

• Queue 모듈 - Lua API 참조
• 큐 컨슈머 가이드 - 컨슈머 패턴 및 워커 풀
• 슈퍼비전 - 컨슈머 라이프사이클 관리