Producer → [Exchange] → [Queue] → Consumer

// Dial accepts a string in the AMQP URI format and returns a new Connection
// over TCP using PlainAuth.
conn, _ := amqp.Dial("amqp://user:pass@localhost:5672")
defer conn.Close()

ch, _ := conn.Channel()
defer ch.Close()

// Default Exchange를 사용하여 "task_queue"에 직접 메시지 전송
ch.PublishWithContext(ctx,
    "",           // exchange: 빈 문자열 = Default Exchange
    "task_queue",  // routing key = 큐 이름
    false, false,
    amqp.Publishing{Body: []byte("hello")},
)

Exchange(direct) --[routing_key="error"]--> error_queue
                 --[routing_key="info"]---> info_queue

routing_key="order.created"  →  "order.*" 패턴의 큐에 매칭
routing_key="order.item.shipped" → "order.#" 패턴의 큐에 매칭

// 1. Exchange 선언
ch.ExchangeDeclare(
    "log_exchange", // name
    "direct",       // type: direct, fanout, topic, headers
    true,           // durable
    false,          // autoDelete
    false,          // internal
    false,          // noWait
    nil,            // args
)

// 2. Queue 선언
ch.QueueDeclare("error_queue", true, false, false, false, nil)
ch.QueueDeclare("info_queue", true, false, false, false, nil)

// 3. Binding: Exchange와 Queue를 routing key로 연결
ch.QueueBind("error_queue", "error", "log_exchange", false, nil)
ch.QueueBind("info_queue", "info", "log_exchange", false, nil)

Publisher --[routing_key="error"]--> log_exchange ---> error_queue
Publisher --[routing_key= "info"]--> log_exchange ---> info_queue

ch.QueueBind(
    "error_queue",  // queue: 바인딩할 큐 이름
    "error",        // key: 라우팅에 사용할 binding key
    "log_exchange", // exchange: 바인딩할 Exchange 이름
    false,          // noWait
    nil,            // args
)

ch.QueueBind("all_logs", "error", "log_exchange", false, nil)
ch.QueueBind("all_logs", "warn", "log_exchange", false, nil)
ch.QueueBind("all_logs", "info", "log_exchange", false, nil)

q, err := ch.QueueDeclare(
    "task_queue", // name: 큐 이름
    true,         // durable: 서버 재시작 후에도 유지
    false,        // autoDelete: 마지막 consumer 해제 시 자동 삭제 여부
    false,        // exclusive: 선언한 connection에서만 접근 가능 여부
    false,        // noWait: 서버 응답을 기다리지 않고 바로 리턴
    nil,          // args: 추가 인자 (amqp.Table)
)

amqp.Table{
    "x-message-ttl":             int32(60000),   // 메시지 TTL (밀리초)
    "x-dead-letter-exchange":    "",              // DLX 지정
    "x-dead-letter-routing-key": "main_queue",    // DLX routing key
    "x-max-length":              int32(10000),    // 큐 최대 메시지 수
}

ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
defer cancel()

body, _ := json.Marshal(payload)

err := ch.PublishWithContext(ctx,
    "",           // exchange
    "task_queue", // routing key
    false,        // mandatory: 라우팅 실패 시 에러 반환 여부
    false,        // immediate: 즉시 소비 불가능 시 에러 반환 여부 (deprecated)
    amqp.Publishing{
        DeliveryMode: amqp.Persistent, // 메시지를 디스크에 저장
        ContentType:  "application/json",
        Body:         body,
    },
)

msgs, _ := ch.Consume(
    "task_queue", // queue
    "",           // consumer tag (빈 문자열이면 서버가 자동 생성)
    false,        // auto-ack: false면 수동 확인 필요
    false,        // exclusive: consumer 하나가 Queue를 독점
    false,        // no-local
    false,        // no-wait
    nil,          // args
)

for d := range msgs {
    result, err := process(d.Body)
    if err != nil {
        d.Reject(false) // false = requeue 안 함
        continue
    }
    d.Ack(false) // false = 이 메시지만 확인 (multiple=false)
}

ch.Qos(
    1,     // prefetch count: 한 번에 1개만 받음
    0,     // prefetch size: 제한 없음 (바이트 단위)
    false, // global: false면 이 channel의 consumer에만 적용
)

ch.QueueDeclare(
    "task_queue", true, false, false, false,
    amqp.Table{
        "x-dead-letter-exchange":    "dlx_exchange",
        "x-dead-letter-routing-key": "dead_letter_queue",
    },
)

amqp.Table{
    "x-dead-letter-exchange":    "",            // Default Exchange
    "x-dead-letter-routing-key": "task_queue",  // 이 큐로 dead letter가 전달됨
}

ch.QueueDeclare(
    "delay_60s", true, false, false, false,
    amqp.Table{
        "x-message-ttl": int32(60000), // 60초 후 만료 (밀리초 단위)
    },
)

ch.PublishWithContext(ctx, "", "some_queue", false, false,
    amqp.Publishing{
        Expiration: "60000", // 문자열로 밀리초 지정
        Body:       body,
    },
)

// 문제가 있는 코드: sleep 동안 Consumer가 멈춘다
case ActionRetry:
    d.Reject(false)
    sleepSeconds := min(MaxBackoff, tryCount*tryCount)
    time.Sleep(time.Duration(sleepSeconds) * time.Second)  // 블로킹!
    republish(ch, payload)

Main Queue ←──────────────── [Default Exchange]
    │                              ↑
    │ (Retry 필요)                 │ (TTL 만료 → dead letter)
    ↓                              │
Retry Queue (TTL=3s)   ────────────┘
Retry Queue (TTL=9s)   ────────────┘
Retry Queue (TTL=27s)  ────────────┘
    ...
    ...

const MaxBackoff = 3600 // 최대 대기시간 (초)

// tryCount → delay(3^n): 1→3s, 2→9s, 3→27s, 4→81s, 5→243s, 6→729s, 7→2187s, 8+→3600s
var RetryQueues = []struct {
    Name string
    TTL  int // milliseconds
}{
    {"retry_3s", 3_000},
    {"retry_9s", 9_000},
    {"retry_27s", 27_000},
    {"retry_81s", 81_000},
    {"retry_243s", 243_000},
    {"retry_729s", 729_000},
    {"retry_2187s", 2_187_000},
    {"retry_3600s", 3_600_000},
}

func retryQueueIndex(tryCount int) int {
    delays := []int{3, 9, 27, 81, 243, 729, 2187}
    if tryCount <= 0 {
        return 0
    }
    backoff := 1
    for i := 0; i < tryCount; i++ {
        backoff *= 3 // 3^tryCount
    }
    if backoff >= MaxBackoff {
        return len(RetryQueues) - 1
    }
    for i, d := range delays {
        if backoff <= d {
            return i
        }
    }
    return len(RetryQueues) - 1
}

const MainQueue = "task_queue"

func declareRetryInfra(ch *amqp.Channel) error {
    for _, rq := range RetryQueues {
        if _, err := ch.QueueDeclare(
            rq.Name,
            true, false, false, false,
            amqp.Table{
                "x-message-ttl":             int32(rq.TTL),
                "x-dead-letter-exchange":    "",         // Default Exchange
                "x-dead-letter-routing-key": MainQueue,  // 만료 후 메인 큐로
            },
        ); err != nil {
            return err
        }
    }
    return nil
}

func sendRetry(ch *amqp.Channel, payload TaskPayload) error {
    idx := retryQueueIndex(payload.TryCount)
    queueName := RetryQueues[idx].Name

    log.Printf("retry tryCount=%d → %s", payload.TryCount, queueName)

    ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
    defer cancel()

    body, err := json.Marshal(payload)
    if err != nil {
        return err
    }

    return ch.PublishWithContext(ctx,
        "",        // Default Exchange
        queueName, // routing key = retry 큐 이름
        false, false,
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,
            ContentType:  "application/json",
            Body:         body,
        },
    )
}

for d := range msgs {
    payload := parsePayload(d.Body)
    result := process(payload)

    switch result {
    case ActionOK:
        d.Ack(false)

    case ActionFailed:
        d.Reject(false) // 버림 (재시도 안 함)

    case ActionRetry:
        d.Reject(false)          // 현재 메시지를 큐에서 제거
        payload.TryCount++
        sendRetry(ch, payload)   // TTL 큐에 넣고 즉시 리턴 → 블로킹 없음
    }
}

1. Producer가 task_queue에 메시지 발행
2. Consumer가 메시지를 수신하여 처리
3. 처리 실패 (ActionRetry):
   a. d.Reject(false)로 원본 메시지 제거
   b. tryCount를 증가시키고 sendRetry 호출
   c. tryCount에 따라 적절한 retry 큐 선택 (예: retry_27s)
   d. Default Exchange를 통해 retry 큐에 메시지 발행
4. retry 큐에서 TTL동안 대기 (e.g. 27s)
5. TTL 만료 → dead letter 발생
6. x-dead-letter-exchange=""로 Default Exchange에 전달
7. x-dead-letter-routing-key="task_queue"로 메인 큐에 도착
8. Consumer가 다시 수신 → 2번으로 돌아감

amqp.Table{
    "x-message-ttl":          int32(60000),
    "x-dead-letter-exchange": "",
}

rabbitmqctl set_policy my-ttl "^retry_.*" \
  '{"message-ttl": 60000, "dead-letter-exchange": ""}' \
  --apply-to queues