知識庫 / Reactive RSS 訂閱

4.5. 測試

有了我們消費者和發佈者端點都已配置完成，現在我們可以使用我們的示例應用程序進行一些測試。

我們需要在 application.yml 中定義 RabbitMQ 服務器連接詳細信息以及至少一個目的地，它應該如下所示：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    
destinations:
  queues:
    NYSE:
      exchange: nyse
      routing-key: NYSE

spring.rabbitmq.* 屬性定義了用於連接到我們運行在本地 Docker 容器中的 RabbitMQ 服務器所需的基本屬性。請注意，上述 IP 地址僅為示例，在特定設置中可能不同。

隊列定義使用 destinations.queues.<name>.*，其中 <name> 用作目標名稱。這裏聲明瞭一個名為“NYSE”的目標，該目標將向 RabbitMQ 中的“nyse”交換器發送消息，並使用“NYSE”路由鍵。

通過命令行或從我們的 IDE 啓動服務器後，我們就可以開始發送和接收消息。我們將使用 curl 工具，這是一個在 Windows、Mac & Linux 操作系統上常用的工具。

以下列表顯示瞭如何將消息發送到我們的目標以及服務器的預期響應：

$ curl -v -d "Test message" http://localhost:8080/queue/NYSE
* timeout on name lookup is not supported
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 8080 (#0)
> POST /queue/NYSE HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.49.1
> Accept: */*
> Content-Length: 12
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 12 out of 12 bytes
< HTTP/1.1 202 Accepted
< content-length: 0
<
* Connection #0 to host localhost left intact

執行完該命令後，我們可以驗證消息是否已通過 RabbitMQ 接收並準備好供消費，通過以下命令進行驗證：

$ docker exec rabbitmq rabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
NYSE    1

現在我們可以使用 curl 命令讀取消息：

$ curl -v http://localhost:8080/queue/NYSE
* timeout on name lookup is not supported
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /queue/NYSE HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.49.1
> Accept: */*
>
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< Content-Type: text/event-stream;charset=UTF-8
<
data:Test message

data:No news is good news...

... same message repeating every 5 secs

如我們所見，首先我們獲取之前存儲的消息，然後開始接收我們的假消息，每 5 秒發送一次。

如果再次運行列出隊列的命令，現在可以看到沒有消息存儲在隊列中：

$ docker exec rabbitmq rabbitmqctl list_queues

Timeout: 60.0 seconds ...
Listing queues for vhost / ...
NYSE    0

5. 場景 2：發佈-訂閲 (Publish-Subscribe)

另一個常見的消息應用程序場景是發佈-訂閲模式，其中一條消息必須發送給多個消費者。

RabbitMQ 提供了兩種支持這種類型的交換 (exchange) 模式：扇出 (Fan-out) 和主題 (Topic)。

這兩種交換模式的主要區別在於，後一種模式允許我們根據註冊時提供的路由鍵模式 (例如 “alarm.mailserver.*”) 過濾接收的消息，而前者則簡單地將傳入的消息複製到所有綁定的隊列。

RabbitMQ 還支持 Header 交換，這允許進行更復雜的消息過濾，但其使用超出了本文的範圍。

5.1. 目的地配置

在啓動時，我們使用另一個 <iPostConstruct> 方法定義 Pub/Sub 目的地，正如在點對點場景中所做的那樣。

唯一的區別在於，我們只創建 <emExchanges>，而沒有 <emQueues> – 這些隊列將在以後按需創建並綁定到 <emExchange> 上，因為我們希望為每個客户端創建一個獨有的 <emQueue>。

@PostConstruct
public void setupTopicDestinations(
    destinationsConfig.getTopics()
      .forEach((key, destination) -> {
          Exchange ex = ExchangeBuilder
            .topicExchange(destination.getExchange())
            .durable(true)
            .build();
            amqpAdmin.declareExchange(ex);
      });
}

5.2. publisher 端點

客户端將使用位於 /topic/{name} 路徑上的 publisher 端點，以發佈消息，這些消息將被髮送到所有已連接的客户端。

如前所述，我們使用一個 @PostMapping 註解，它返回一個 Mono 對象，該對象包含消息發送後的狀態。

@PostMapping(value = "/topic/{name}")
public Mono<ResponseEntity<?>> sendMessageToTopic(
  @PathVariable String name, @RequestBody String payload) {

    DestinationInfo d = destinationsConfig
      .getTopics()
      .get(name);
    
    if (d == null) {
        return Mono.just(ResponseEntity.notFound().build());
    }      
    
   return Mono.fromCallable(() -> {
       amqpTemplate.convertAndSend(
         d.getExchange(), d.getRoutingKey(),payload);   
            return ResponseEntity.accepted().build();
        });
    }

5.3. 訂閲者端點

我們的訂閲者端點位於 /topic/{name}，並向連接的客户端產生一個 Flux 消息流。

這些消息包括接收到的消息和每 5 秒生成的一個虛擬消息：

@GetMapping(
  value = "/topic/{name}",
  produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<?> receiveMessagesFromTopic(@PathVariable String name) {
    DestinationInfo d = destinationsConfig.getTopics()
        .get(name);
    if (d == null) {
        return Flux.just(ResponseEntity.notFound()
            .build());
    }
    Queue topicQueue = createTopicQueue(d);
    String qname = topicQueue.getName();
    MessageListenerContainer mlc = messageListenerContainerFactory.createMessageListenerContainer(qname);
    Flux<String> f = Flux.<String> create(emitter -> {
        mlc.setupMessageListener((MessageListener) m -> {
            String payload = new String(m.getBody());
            emitter.next(payload);
        });
        emitter.onRequest(v -> {
            mlc.start();
        });
        emitter.onDispose(() -> {
            amqpAdmin.deleteQueue(qname);
            mlc.stop();
        });            
      });
    
    return Flux.interval(Duration.ofSeconds(5))
        .map(v -> "No news is good news")
        .mergeWith(f);
}

這段代碼與之前案例基本相同，僅有以下差異：首先，我們為每個新訂閲者創建一個新的 Queue。

我們通過調用 createTopicQueue() 方法來實現，該方法使用來自 DestinationInfo 實例的信息，創建一個專有、不可持久化的隊列，然後使用配置的路由鍵將其綁定到 Exchange。

private Queue createTopicQueue(DestinationInfo destination) {

    Exchange ex = ExchangeBuilder
      .topicExchange(destination.getExchange())
      .durable(true)
      .build();
    amqpAdmin.declareExchange(ex);
    Queue q = QueueBuilder
      .nonDurable()
      .build();     
    amqpAdmin.declareQueue(q);
    Binding b = BindingBuilder.bind(q)
      .to(ex)
      .with(destination.getRoutingKey())
      .noargs();        
    amqpAdmin.declareBinding(b);
    return q;
}

需要注意的是，儘管我們再次聲明瞭 Exchange，RabbitMQ 不會創建新的一個，因為我們在啓動時就已經聲明瞭它。

第二個差異在於我們傳遞給 onDispose() 方法的 lambda 函數，該函數會在訂閲者斷開連接時也會刪除 Queue。

5.3. 測試

為了測試 Pub/Sub 場景，首先需要在 application.yml 中定義一個主題目的地，如下所示：

destinations:
## ... queue destinations omitted      
  topics:
    weather:
      exchange: alerts
      routing-key: WEATHER

這裏，我們定義了一個主題端點，該端點將在 /topic/weather 位置可用。該端點將用於將消息發佈到 RabbitMQ 的“alerts”交換，並使用“WEATHER”路由鍵。

啓動服務器後，我們可以使用 rabbitmqctl 命令來驗證交換是否已創建。

$ docker exec docker_rabbitmq_1 rabbitmqctl list_exchanges
Listing exchanges for vhost / ...
amq.topic       topic
amq.fanout      fanout
amq.match       headers
amq.headers     headers
        direct
amq.rabbitmq.trace      topic
amq.direct      direct
alerts  topic

現在，如果我們發出 list_bindings 命令，我們可以看到沒有與“alerts”交換相關的隊列：

$ docker exec rabbitmq rabbitmqctl list_bindings
Listing bindings for vhost /...
        exchange        NYSE    queue   NYSE    []
nyse    exchange        NYSE    queue   NYSE    []

讓我們啓動幾個訂閲者，它們將訂閲我們的目標，通過在每個命令殼中打開兩個命令殼併發出以下命令來實現：

$ curl -v http://localhost:8080/topic/weather
* timeout on name lookup is not supported
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /topic/weather HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.49.1
> Accept: */*
>
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< Content-Type: text/event-stream;charset=UTF-8
<
data:No news is good news...

# ... same message repeating indefinitely

最後，我們再次使用 curl 向我們的訂閲者發送警報：

$ curl -v -H "Content-Type: application/json" -d "Hurricane approaching!" http://localhost:8080/topic/weather
* timeout on name lookup is not supported
*   Trying 127.0.0.1...
* Connected to localhost (127.0.0.1) port 8080 (#0)
> POST /topic/weather HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.49.1
> Accept: */*
> Content-Length: 22
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 22 out of 22 bytes
< HTTP/1.1 202 Accepted
< content-length: 0
<
* Connection #0 to host localhost left intact

一旦我們發送消息，我們幾乎可以立即在每個訂閲者的設備上看到“颶風逼近！”的消息。

如果現在檢查可用的綁定，我們可以看到每個訂閲者都有一個隊列：

$ docker exec rabbitmq rabbitmqctl list_bindings
Listing bindings for vhost /...
        exchange        IBOV    queue   IBOV    []
        exchange        NYSE    queue   NYSE    []
        exchange        spring.gen-i0m0pbyKQMqpz2_KFZCd0g       
  queue   spring.gen-i0m0pbyKQMqpz2_KFZCd0g       []
        exchange        spring.gen-wCHALTsIS1q11PQbARJ7eQ       
  queue   spring.gen-wCHALTsIS1q11PQbARJ7eQ       []
alerts  exchange        spring.gen-i0m0pbyKQMqpz2_KFZCd0g     
  queue   WEATHER []
alerts  exchange        spring.gen-wCHALTsIS1q11PQbARJ7eQ     
  queue   WEATHER []
ibov    exchange        IBOV    queue   IBOV    []
nyse    exchange        NYSE    queue   NYSE    []
quotes  exchange        NYSE    queue   NYSE    []

當我們在訂閲者的殼層上按下 Ctrl-C 時，我們的網關最終會檢測到客户端已斷開連接，並移除這些綁定。

6. 結論

在本文中，我們演示瞭如何使用 spring-amqp 模塊創建一個簡單的響應式應用程序，並與之交互以使用 RabbitMQ 服務器。

僅需幾行代碼，我們就能創建一個功能完善的 HTTP 到 AMQP 網關，它支持點對點和發佈訂閲集成模式，並且可以輕鬆地通過添加標準 Spring 功能來擴展以添加額外的功能，例如安全功能。