前言:

想像一下 LINE Bot 已經有上萬名好友。這個 LINE Bot 設計了許多互動指令，平常一秒鐘只有幾十個訊息湧入還撐得住。但如果某一天突然有數萬名使用者同時發訊息，你的 /webhook endpoint 就會瞬間被 LINE Server 傳來的 request 給 overload。這種「瞬間高流量」(High Concurrency) 的情境該如何妥善的處理？

What is Message Queue?

Message Queue（訊息佇列）是一種 非同步（asynchronous）通訊技術。
它的角色就像「城市裡的郵局」：

• 每個人（Producer）把信丟到郵局的信箱。
• 郵局（Message Queue）會保管並排好順序 (具備 persistence 服務)。
• 收件人（Consumer）不一定要馬上領信，有空時再慢慢處理。
  這樣，就算收件人暫時不在，信件也不會不見，寄件人也不用等太久。

假設有三個微服務 A → B → C。

• 如果沒有 MQ：
  A 必須等 B 成功，B 又要等 C 成功。如果 B crash，A 的所有訊息都卡死。
• 有了　Message Queue 這個個郵局：
  A 把訊息像寄信一樣丟到郵局，B 有空來領。同理， B 也去郵局寄信，C 有空再來領，那麼就算 B 一時斷掉，訊息也不會不見。

👉 decoupling 就算中間某個暫時 crash 了，訊息也不會遺失。

相關的 terms

名詞

• Producer：寄件人，把信寄到郵局。
• Message Queue：郵局的信箱，把信保存好。
• Consumer/ Worker：收件人，隨時來郵局取信。
  動作:
• Publish/Produce：把訊息交給郵局（send message）。
• Consume：把信領出來，開始處理（process message）。

Consumer Distribution

https://bunny.net/academy/http/what-is-a-message-queue/

調整一條 Queue 的訊息要怎麼分配給多個 Worker。

• 一對一（point-to-point）
  每則訊息只給一個 Worker 處理。就像快遞一個包裹只送到一個地址。
• 一對多（publish-subscribe）
  同一則訊息可以被多個 Queue 或 Consumer 接收。就像新聞廣播，一則新聞同時被多個聽眾聽到。
• 競爭消費（competing consumers）
  多個 Worker 同時從同一個 Queue 搶任務，每個訊息只會被其中一個處理。就像排班的外送員，一張訂單只會被一位外送員接走。

Delayed Queue

https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-delay-queues.html
訊息延後一段時間才被處理。

• 電商訂單，30 分鐘未付款就自動取消。
• 遊戲伺服器延遲任務（例如：冷卻時間後才觸發）。

Dead Letter Queue, DLQ

https://www.baeldung.com/kafka-spring-dead-letter-queue
https://aws.amazon.com/what-is/dead-letter-queue/
收集「失敗的訊息」，避免無限 loop 卡死系統。

• message 一直 deserialize 失敗，就丟到 DLQ 供後台排查。
• 就像郵局寄不出去的信，最後會放到 DLQ 。

使用時機

1 High Concurrency 高併發流量
→ Queue 保管這些 request，後端慢慢處理，避免網站直接 crash。

2 Service Decoupling 解耦服務
A 服務只管把訊息丟到 Queue，不必管 B/C 有沒有 crash，系統更有彈性以及容錯。

3 Non-Real-Time Tasks 非即時
上傳圖片 → 轉檔 → 壓縮 → 存雲端，使用 Queue 分派給多個 Worker 處理，就像多位工人同時在工廠作業。

4 Reliability 可靠性
像銀行轉帳，訊息 (交易) 不能消失。Queue 支援 ack / retry 機制，確保送達。

⚠️ Crash Scenarios

如果 Consumer crash 了，郵局裡的信還在，等收件人回來就能繼續處理。

如果 Message Queue crash 了，通常 Message Queue 裡的資料已經持久化（persistence）了，存在硬碟或資料庫中。因此當 Queue 服務重啟後，訊息還在不會遺失。可是，在這段 crash 的時間內，Publisher 的確無法把新訊息送進 Queue（就像郵局大門暫時關閉），但重點是已經進去的訊息不會掉，重啟後就能繼續處理。

RabbitMQ 簡單使用介紹

Dependency and configuration files:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

docker-compose.yml:

version: "3.8"
services:
  rabbitmq:
    image: rabbitmq:3.13-management
    container_name: rabbitmq
    ports:
      - "5672:5672"      # AMQP
      - "15672:15672"    # Management UI
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

application.yml:

server:
  port: 8080

spring:
  rabbitmq:
    host: localhost       
    port: 5672
    username: guest
    password: guest

mq:
  queue: line.events.queue

Logic diagram:

LINE Server
   |
   | POST /webhook  (JSON)
   v
WebhookController
   |-- rabbit.convertAndSend("", queueName, payload)
   '-- 200 OK
   .
RabbitMQ (Default Exchange)
   |-- route by routingKey == queueName
   v
Queue: line.events.queue
   v
LineEventConsumer (@RabbitListener)
   '-- onMessage(String body)  // TODO: DB/Reply API

com/vanillasky/demo/
├─ MinApp.java            
├─ RabbitConfig.java      # 宣告 durable queue: line.events.queue
├─ WebhookController.java # /webhook → publish to default exchange → 200 OK
└─ LineEventConsumer.java # @RabbitListener 收訊息 → println

RabbitConfig:

@Configuration
public class RabbitConfig {

  @Bean
  public Queue queue(@Value("${mq.queue}") String queueName) {
    // durable queue；Spring 會自動用 RabbitAdmin 宣告
    return new Queue(queueName, true);
  }
}

WebhookController:

@RestController
@RequestMapping("/webhook")
public class WebhookController {

  private final RabbitTemplate rabbit;
  private final String queueName;

  public WebhookController(RabbitTemplate rabbit, @Value("${mq.queue}") String queueName) {
    this.rabbit = rabbit;
    this.queueName = queueName;
  }

  @PostMapping
  public ResponseEntity<Void> handle(@RequestBody String payload,
                                     @RequestHeader(value = "X-Line-Signature", required = false) String sig) {
    // TODO:　簽名驗證
    // 直接用 Default Exchange 把訊息送到 queue（routingKey = queue 名稱）
    rabbit.convertAndSend("", queueName, payload);
    return ResponseEntity.ok().build();
  }
}

LineEventConsumer:

@Component
public class LineEventConsumer {

  @RabbitListener(queues = "${mq.queue}")
  public void onMessage(String body) {
    // bussiness logic: DB／LINE Reply API 
    System.out.println("[consumer] got: " + body);
  }
}

DemoApp :

@SpringBootApplication
public class DemoApp {
  public static void main(String[] args) {
    SpringApplication.run(MinApp.class, args);
  }
}

測試:

curl -X POST http://localhost:8080/webhook \
  -H "Content-Type: application/json" \
  -d '{"events":[{"type":"message","message":{"type":"text","text":"hi"}}]}'

[consumer] got: {"events":[{"type":"message","message":{"type":"text","text":"hi"}}]}

不該用 MQ 的時候

• 需要即時回應（Real-time response）
  MQ 本質是 async ，而 queue 的排隊就會造成延遲。若　business logic　要求 "馬上回應"，例如使用者提交表單馬上要看到結果，就不該用 MQ。

　Message queues are not for real-time communications … when you want an immediate response. (Contrast Security)
https://www.contrastsecurity.com/security-influencers/what-is-a-message-queue-importance-use-cases-and-vulnerabilities-contrast-security

• 需要強一致性的交易（Strong consistency across services）
  如果　business logic 必須保證「一個操作跨多個資源同時成功或同時失敗」，MQ 的 async 會讓一致性控制的邏輯非常複雜。

　Don’t design events as delayed commands … it creates hidden coupling and　complexity. (Ben Morris – Event-driven anti-patterns)
https://www.ben-morris.com/event-driven-architecture-and-message-design-anti-patterns-and-pitfalls/

-　系統簡單、流量低（Overengineering risk）
如果系統規模很小，request 量低，直接用同步呼叫即可。導入 MQ 會難以 maintain 。

Engineers warn about overusing MQ where a simple DB + background job is enough. https://news.ycombinator.com/item?id=40723302