消息容器介紹

1. 前言

Hello，大家好。本小節(jié)會為同學們介紹 RabbitMQ 在 Spring 生態(tài)中的消息容器，消息容器是 RabbitMQ 在 Spring 生態(tài)中的第三個核心元素，消息容器指的不是一種特定的容納消息的容器，而是一系列提供容納消息、監(jiān)聽消息等其他對消息指標進行監(jiān)控的工具，接下來就讓我們來看看到底什么是消息容器吧。

本節(jié)主要內(nèi)容：

• 消息容器基礎概念概述；

• 常用消息容器基礎配置概述。

2. 消息容器基礎概念概述

基礎概念：

消息容器，即容納消息的容器。通過對上述部分小節(jié)的學習，我們已經(jīng)知道，在 原生 RabbitMQ 知識體系中，并沒有單獨地一個消息容器的概念，只有消息隊列的概念。

但是，在 Spring-AMQP 中，消息容器的概念和原生 RabbitMQ 知識體系中的消息隊列的概念完全不一樣。Spring-AMQP 中的消息容器指的是：可以提供對消息隊列、消息、消費者、消息簽收模式進行控制，以及消息的全方位監(jiān)聽的一系列工具的統(tǒng)稱。

我們知道，在 RabbitMQ 中，充當核心角色的就是我們應用程序中的數(shù)據(jù)，也就是消息，那么，對消息以及消息隊列進行全方位監(jiān)控的工具，在 Spring-AMQP 中就被稱為消息容器。

消息容器的主要作用就是對經(jīng)過 RabbitTemplate 消息模板發(fā)送到 RabbitMQ Server 中的消息進行監(jiān)聽，當然，要確保 RabbitTemplate 和消息容器所使用的是同一個 RabbitAdmin 構造的連接，這樣才能對消息進行監(jiān)控。

在介紹完消息容器的基礎概念之后，下面讓我們來看一下如何對消息容器進行簡單的配置吧。

3. 常用消息容器基礎配置概述

還是像上節(jié)小節(jié)一樣，要想在 Spring 中使用消息容器，需要將 Spring-AMQP 和 AMQP-Stater 的依賴先引入進來，方便起見，同學們可以直接拷貝下放代碼：

3.1 引入消息容器

以 Maven 引入方式為例，引入代碼如下所示：

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>3.6.5</version>
</dependency>

在將這兩個依賴進行引入之后，我們就可以對消息容器進行配置了。

3.2 常用消息容器基礎配置

在 Spring-AMQP 中，消息容器共有這五種：AbstractMessageListenerContainer、DirectMessageListenerContainer、DirectReplyToMessageListenerContainer、SimpleMessageListenerContainer，以及 MessageListenerContainer 接口。

在這五種消息容器中，常用的消息容器只有一種，就是 SimpleMessageListenerContainer ，由于 SimpleMessageListenerContainer 消息容器配置起來簡單方便，所支持的配置的屬性比較全面，所以該消息容器是所有消息容器中使用頻率最高的，也是實際工作中使用最多的一種消息容器。

本節(jié)以 SimpleMessageListenerContainer 消息容器為例，來介紹一下 SimpleMessageListenerContainer 消息容器該如何使用吧。

Tips: SimpleMessageListenerContainer 消息容器是眾多消息容器中最基本的消息容器，我們把 SimpleMessageListenerContainer 消息容器的基本使用了解之后，其他的消息容器就無師自通了，只不過其他的消息容器所提供的配置屬性或多或少罷了。

初始化 SimpleMessageListenerContainer 消息容器

像 RabbitAdmin 和 RabbitTemplate 一樣，要想使用消息容器，需要先對消息容器進行初始化，這個初始化過程非常簡單，初始化 SimpleMessageListenerContainer 消息容器的代碼如下所示：

代碼實現(xiàn)：

@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
  SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
  return simpleMessageListenerContainer;
}

代碼解釋：

第 1 行，我們使用 Spring 的 Bean 注解將我們聲明的 simpleMessageListenerContainer 方法注入到 Spring 容器中，這樣 Spring 容器就可以監(jiān)聽到我們注入的配置。

第 2 行，我們使用 Spring-AMQP 中的 SimpleMessageListenerContainer 類，來聲明了一個名為 simpleMessageListenerContainer 的方法，用來對 simpleMessageListenerContainer 消息容器進行初始化。

第 3 行，我們實例化了一個 simpleMessageListenerContainer 實例，該實例是 Spring-AMQP 中對 simpleMessageListenerContainer 消息容器進行初始化的實例，要想使用 SimpleMessageListenerContainer ，就必須要初始化該實例。

第 4 行，我們將初始化好的 simpleMessageListenerContainer 實例進行返回。

SimpleMessageListenerContainer 基本使用

配置 SimpleMessageListenerContainer 消息容器，和之前的配置 RabbitAdmin 以及 RabbitTemplate 不同，我們只需要直接在上述初始化方法中去填充我們需要的配置屬性即可，這里以基本常用屬性為例，代碼如下所示：

simpleMessageListenerContainer.setQueues(new Queue("test_001"), new Queue("test_002"), new Queue("test_003"));
simpleMessageListenerContainer.setConcurrentConsumers(1);
simpleMessageListenerContainer.setMaxConcurrentConsumers(3);
simpleMessageListenerContainer.setDefaultRequeueRejected(false);
simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
simpleMessageListenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {
  @Override
  public String createConsumerTag(String queue) {
  return queue + "_" + "created consumer tag";
  }
});
simpleMessageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
  @Override
  public void onMessage(Message message, Channel channel) throws Exception {
  String msg = new String(message.getBody());
  // do someting...
  }
  simpleMessageListenerContainer.setAutoStartup(true)
});

代碼解釋：

第 1-2 行，我們使用 simpleMessageListenerContainer 的 setQueues 方法，來設置我們需要進行監(jiān)聽的隊列，這里新建了三個隊列，名稱分別為：test_001 ，test_002 ，test_003。

第 3-4 行，我們分別使用了 simpleMessageListenerContainer 的 setConcurrentConsumers 方法和 setMaxConcurrentConsumers 方法，來分別設置當前用于消費消息的消費者個數(shù)，以及最大允許用于消費消息的消費者個數(shù)，這里分別被設置成了 1 和 3 ，表示：當前用于消費消息的消費者個數(shù)為 1 個，最大允許進行消費消息的消費者個數(shù)為 3 個。

第 5 行，我們使用 simpleMessageListenerContainer 的 setDefaultRequeueRejected 方法，來設置是否開啟重回隊列， 關于什么是重回隊列，在前面小節(jié)中都有詳細介紹，這里不再贅述。當 setDefaultRequeueRejected 方法的值為 true 時，表示開啟重回隊列，當為 false 時，表示關閉重回隊列，這里表示不適使用重回隊列，即該方法的值為 false。

第 6 行，我們使用 simpleMessageListenerContainer 的 setAcknowledgeMode 方法，來設置當前消息的接收模式， 即對應原生 RabbitMQ 中的消息簽收模式，是自動簽收還是手動簽收，這里，AcknowledgeMode.AUTO 表示自動簽收消息。

第 7-10 行，我們使用 simpleMessageListenerContainer 的 setConsumerTagStrategy 方法，來為消費者設置一個標簽，設置標簽需要通過 new ConsumerTagStrategy 匿名內(nèi)部類的方式來實現(xiàn)，通過重寫其中的 createConsumerTag 方法來設置消費者的 Tag 標簽。

第 13-17 行，我們使用 simpleMessageListenerContainer 的 setMessageListener 方法，來添加一種監(jiān)聽器， 添加監(jiān)聽器的方式也是通過匿名內(nèi)部類實現(xiàn)；new ChannelAwareMessageListener 監(jiān)聽器監(jiān)聽的是，當消息經(jīng)過 channel 時的情況，實現(xiàn)該監(jiān)聽器需要重寫其中的 onMessage 方法，我們可以將業(yè)務邏輯填充在 onMessage 方法中。

setMessageListener 方法可以添加任意一種 Spring-AMQP 中支持的監(jiān)聽器，不單單只有 ChannelAwareMessageListener 監(jiān)聽器這一種，其他的監(jiān)聽器就交給同學們自行探索吧。

第 19 行，我們使用 simpleMessageListenerContainer 的 setAutoStartup 方法，來設置 simpleMessageListenerContainer 消息容器的啟動模式，這里被設置為了 true ，表示，當 Spring 容器啟動時，simpleMessageListenerContainer 消息容器也隨之啟動。

Tips: 1. setQueues 方法中的參數(shù)只有一個，就是消息隊列，其參數(shù)類型為 Queue… ，該參數(shù)理論上可以支持無限個隊列被添加進去，從而對這些隊列進行監(jiān)聽；
2. AcknowledgeMode 的消息簽收類型，不只有 AUTO 這一種，還有 NONE 和 MANUAL ，分別表示不設置和手動監(jiān)聽。
3. simpleMessageListenerContainer 消息容器是支持熱加載的一款消息容器，即當我們的應用程序處于運行狀態(tài)時，我們手動改變了某一配置屬性的值，這種改動是立即生效的，不需要我們重啟我們的應用程序，這點同學們注意。

4. 小結

本小節(jié)詳細為同學們介紹了 Spring 生態(tài)中的消息容器這一概念，并通過結合常用的 SimpleMessageListenerContainer 消息容器，來介紹了消息容器的基本配置方法和常用配置屬性，希望同學們可以對 Spring-AMQP 中的常用消息容器 SimpleMessageListenerContainer 有一個基礎的認知，這樣，我們以后在使用其他消息容器時，才會做到融會貫通。