Spring AMQP項目將Spring的核心概念應用于基于AMQP的消息傳遞解決方案的開發。 我們提供了一個“模板”作為發送和接收消息的高級抽象。 我們還為消息驅動的 POJO 提供支持。 這些庫有助于管理 AMQP 資源，同時促進依賴關系注入和聲明性配置的使用。 在所有這些情況下，你可以看到與 Spring 框架中的 JMS 支持的相似之處。 有關其他與項目相關的信息，請訪問Spring AMQP項目主頁。

(資料圖片僅供參考)

2. 最新消息

2.1. 自 2.4 以來 3.0 的變化

2.1.1. Java 17， Spring Framework 6.0

此版本需要 Spring Framework 6.0 和 Java 17

2.1.2. 遠程處理

不再支持遠程處理功能（使用 RMI）。

2.1.3. 觀察

現在支持使用千分尺啟用計時器觀察和跟蹤。 有關更多信息，請參閱千分尺觀察。

2.1.4. 原生鏡像

提供了對創建本機映像的支持。 有關詳細信息，請參閱本機映像。

2.1.5. 異步兔子模板

2.1.6. 流支持更改

現在支持超級流和單個活躍使用者。

有關更多信息，請參閱使用 RabbitMQ 流插件。

2.1.7. 變更??@RabbitListener??

批處理偵聽器現在可以同時使用 . 批處理消息傳遞適配器現在可確保該方法適用于使用批處理。 將容器工廠設置為 時，該屬性也設置為 。 有關詳細信息，請參閱批處理@RabbitListener。??Collection????List????consumerBatchEnabled????true????batchListener????true??

??MessageConverter??s 現在可以返回空值;這目前由 . 有關詳細信息，請參閱從消息轉換??Optional.empty()????Jackson2JsonMessageConverter??

現在，您可以通過容器工廠而不是通過 上的屬性配置 。 有關詳細信息，請參閱回復管理。??ReplyPostProcessor????@RabbitListener??

2.1.8. 連接工廠變更

中的默認值現在為 。這會導致在提供多個地址時連接到隨機主機。 有關詳細信息，請參閱連接到集群。??addressShuffleMode????AbstractConnectionFactory????RANDOM??

不再使用 RabbitMQ 庫來確定哪個節點是隊列的領導者。 有關詳細信息，請參閱隊列相關性和LocalizedQueueConnectionFactory。??LocalizedQueueConnectionFactory????http-client??

3. 簡介

參考文檔的第一部分是對Spring AMQP和基本概念的高級概述。 它包括一些代碼片段，可幫助您盡快啟動并運行。

3.1. 不耐煩的人快速瀏覽

3.1.1. 簡介

這是開始春季AMQP的五分鐘導覽。

先決條件：安裝并運行 RabbitMQ 代理 （https://www.rabbitmq.com/download.html）。 然后獲取 spring-rabbit JAR 及其所有依賴項 - 最簡單的方法是在構建工具中聲明依賴項。 例如，對于 Maven，您可以執行類似于以下內容的操作：

  org.springframework.amqp  spring-rabbit  3.0.0

對于 Gradle，您可以執行類似以下步驟的操作：

compile "org.springframework.amqp:spring-rabbit:3.0.0"

兼容性

Spring 框架版本的最低依賴關系是 5.2.0。

Java 客戶機庫的最低版本為 5.7.0。??amqp-client??

流隊列的最小 Java 客戶機庫為 0.7.0。??stream-client??

非常非常快

本節提供最快的介紹。

首先，添加以下語句以使本節后面的示例正常工作：??import??

以下示例使用普通的命令式 Java 發送和接收消息：

ConnectionFactory connectionFactory = new CachingConnectionFactory();AmqpAdmin admin = new RabbitAdmin(connectionFactory);admin.declareQueue(new Queue("myqueue"));AmqpTemplate template = new RabbitTemplate(connectionFactory);template.convertAndSend("myqueue", "foo");String foo = (String) template.receiveAndConvert("myqueue");

請注意，原生 Java Rabbit 客戶端中還有一個。 我們在前面的代碼中使用了 Spring 抽象。 它緩存通道（以及可選的連接）以供重用。 我們依賴于代理中的默認交換（因為在發送中沒有指定任何交換），以及所有隊列通過其名稱默認綁定到默認交換（因此，我們可以在發送中使用隊列名稱作為路由密鑰）。 這些行為在 AMQP 規范中定義。??ConnectionFactory??

使用 XML 配置

以下示例與前面的示例相同，但將資源配置外部化為 XML：

ApplicationContext context =    new GenericXmlApplicationContext("classpath:/rabbit-context.xml");AmqpTemplate template = context.getBean(AmqpTemplate.class);template.convertAndSend("myqueue", "foo");String foo = (String) template.receiveAndConvert("myqueue");

缺省情況下，聲明會自動查找 、 的 bean，并代表用戶將它們聲明給代理。 因此，您不需要在簡單的 Java 驅動程序中顯式使用該 Bean。 有很多選項可用于配置 XML 架構中組件的屬性。 您可以使用 XML 編輯器的自動完成功能來瀏覽它們并查看其文檔。??????Queue????Exchange????Binding??

使用爪噠配置

以下示例重復與前面的示例相同的示例，但使用了 Java 中定義的外部配置：

ApplicationContext context =    new AnnotationConfigApplicationContext(RabbitConfiguration.class);AmqpTemplate template = context.getBean(AmqpTemplate.class);template.convertAndSend("myqueue", "foo");String foo = (String) template.receiveAndConvert("myqueue");........@Configurationpublic class RabbitConfiguration {    @Bean    public CachingConnectionFactory connectionFactory() {        return new CachingConnectionFactory("localhost");    }    @Bean    public RabbitAdmin amqpAdmin() {        return new RabbitAdmin(connectionFactory());    }    @Bean    public RabbitTemplate rabbitTemplate() {        return new RabbitTemplate(connectionFactory());    }    @Bean    public Queue myQueue() {       return new Queue("myqueue");    }}

具有 Spring 引導自動配置和異步 POJO 偵聽器

Spring 引導會自動配置基礎結構 bean，如以下示例所示：

@SpringBootApplicationpublic class Application {    public static void main(String[] args) {        SpringApplication.run(Application.class, args);    }    @Bean    public ApplicationRunner runner(AmqpTemplate template) {        return args -> template.convertAndSend("myqueue", "foo");    }    @Bean    public Queue myQueue() {        return new Queue("myqueue");    }    @RabbitListener(queues = "myqueue")    public void listen(String in) {        System.out.println(in);    }}

4. 參考資料

參考文檔的這一部分詳細介紹了構成Spring AMQP的各種組件。主要章節介紹了開發 AMQP 應用程序的核心類。 本部分還包括有關示例應用程序的章節。

4.1. 使用彈簧 AMQP

本章探討了使用 Spring AMQP 開發應用程序的基本組件接口和類。

4.1.1. AMQP 抽象

Spring AMQP 由兩個模塊組成（每個模塊在發行版中由一個 JAR 表示）：和 。 “spring-amqp”模塊包含該軟件包。 在該包中，您可以找到表示核心 AMQP“模型”的類。 我們的目的是提供不依賴于任何特定 AMQP 代理實現或客戶端庫的通用抽象。 最終用戶代碼可以跨供應商實現更具可移植性，因為它可以僅針對抽象層進行開發。 然后，這些抽象由特定于代理的模塊實現，例如“spring-rabbit”。 目前只有一個 RabbitMQ 實現。 但是，除了RabbitMQ之外，這些抽象已經在.NET中使用Apache Qpid進行了驗證。 由于 AMQP 在協議級別運行，原則上可以將 RabbitMQ 客戶端與任何支持相同協議版本的代理一起使用，但我們目前不測試任何其他代理。??spring-amqp????spring-rabbit????org.springframework.amqp.core??

本概述假定您已經熟悉 AMQP 規范的基礎知識。 如果沒有，請查看其他資源中列出的資源

??Message??

0-9-1 AMQP 規范未定義類或接口。 相反，在執行諸如 之類的操作時，內容將作為字節數組參數傳遞，其他屬性將作為單獨的參數傳入。 Spring AMQP將類定義為更通用的AMQP域模型表示的一部分。 該類的目的是將主體和屬性封裝在單個實例中，以便 API 可以反過來更簡單。 下面的示例演示類定義：??Message????basicPublish()????Message????Message????Message??

public class Message {    private final MessageProperties messageProperties;    private final byte[] body;    public Message(byte[] body, MessageProperties messageProperties) {        this.body = body;        this.messageProperties = messageProperties;    }    public byte[] getBody() {        return this.body;    }    public MessageProperties getMessageProperties() {        return this.messageProperties;    }}

該接口定義了幾個常見屬性，例如“messageId”、“timestamp”、“內容類型”等。 您還可以通過調用該方法，使用用戶定義的“標頭”擴展這些屬性。??MessageProperties????setHeader(String key, Object value)??

交換

該接口表示 AMQP 交換，即消息生產者發送到的內容。 經紀商虛擬主機中的每個交易所都有一個唯一的名稱以及一些其他屬性。 以下示例顯示了該接口：??Exchange????Exchange??

public interface Exchange {    String getName();    String getExchangeType();    boolean isDurable();    boolean isAutoDelete();    Map getArguments();}

如您所見，an 也有一個由 中定義的常量表示的“類型”。 基本類型包括：、、 和 。 在核心包中，您可以找到每種類型的接口實現。 這些類型的行為在如何處理隊列綁定方面有所不同。 例如，交換允許隊列由固定路由密鑰（通常是隊列的名稱）綁定。 交換支持具有路由模式的綁定，這些模式可能分別包含“*”和“#”通配符，分別表示“恰好一個”和“零個或多個”。 交易所發布到綁定到它的所有隊列，而不考慮任何路由密鑰。 有關這些和其他 Exchange 類型的詳細信息，請參閱其他資源。??Exchange????ExchangeTypes????direct????topic????fanout????headers????Exchange????Exchange????Direct????Topic????Fanout??

隊列

該類表示消息使用者從中接收消息的組件。 與各種類一樣，我們的實現旨在成為此核心 AMQP 類型的抽象表示。 以下清單顯示了該類：??Queue????Exchange????Queue??

public class Queue  {    private final String name;    private volatile boolean durable;    private volatile boolean exclusive;    private volatile boolean autoDelete;    private volatile Map arguments;    /**     * The queue is durable, non-exclusive and non auto-delete.     *     * @param name the name of the queue.     */    public Queue(String name) {        this(name, true, false, false);    }    // Getters and Setters omitted for brevity}

請注意，構造函數采用隊列名稱。 根據實現，管理模板可能會提供用于生成唯一命名隊列的方法。 此類隊列可用作“回復”地址或其他臨時情況。 因此，自動生成隊列的“獨占”和“自動刪除”屬性都將設置為“true”。

捆綁

假設生產者發送到交易所，消費者從隊列接收，將隊列連接到交易所的綁定對于通過消息傳遞連接這些生產者和消費者至關重要。 在Spring AMQP中，我們定義了一個類來表示這些連接。 本節回顧將隊列綁定到交易所的基本選項。??Binding??

您可以使用固定路由密鑰將隊列綁定到 ，如以下示例所示：??DirectExchange??

new Binding(someQueue, someDirectExchange, "foo.bar");

您可以使用路由模式將隊列綁定到 ，如以下示例所示：??TopicExchange??

new Binding(someQueue, someTopicExchange, "foo.*");

您可以將隊列綁定到沒有路由密鑰的 ，如以下示例所示：??FanoutExchange??

new Binding(someQueue, someFanoutExchange);

我們還提供了一個促進“流暢 API”樣式，如以下示例所示：??BindingBuilder??

Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");

就其本身而言，類的實例僅保存有關連接的數據。 換句話說，它不是一個“活動”組件。 但是，正如稍后將在配置代理中看到的那樣，該類可以使用實例來實際觸發代理上的綁定操作。 此外，正如您在同一部分中所看到的，您可以通過在類中使用 Spring 的注釋來定義實例。 還有一個方便的基類，它進一步簡化了生成與 AMQP 相關的 Bean 定義的方法，并識別隊列、交換和綁定，以便在應用程序啟動時在 AMQP 代理上聲明它們。??Binding????AmqpAdmin????Binding????Binding????@Bean????@Configuration??

也在核心包中定義。 作為實際 AMQP 消息傳遞中涉及的主要組件之一，它在其自己的部分中進行了詳細討論（請參閱AmqpTemplate）。??AmqpTemplate??

4.1.2. 連接和資源管理

雖然我們在上一節中描述的 AMQP 模型是通用的并且適用于所有實現，但當我們進入資源管理時，細節特定于代理實現。 因此，在本節中，我們將重點介紹僅存在于“spring-rabbit”模塊中的代碼，因為此時，RabbitMQ 是唯一受支持的實現。

管理與 RabbitMQ 代理的連接的核心組件是接口。 實現的職責是提供 的實例，該實例是 的包裝器。??ConnectionFactory????ConnectionFactory????org.springframework.amqp.rabbit.connection.Connection????com.rabbitmq.client.Connection??

選擇連接工廠

有三家連接工廠可供選擇

前兩個是在 2.3 版中添加的。

對于大多數用例，應使用 。 如果要確保嚴格的消息排序而不需要使用作用域內操作，則可以使用 。 如果要使用相關的發布者確認，或者希望通過其打開多個連接，則應使用 。??PooledChannelConnectionFactory????ThreadChannelConnectionFactory????CachingConnectionFactory????CacheMode??

所有三個工廠都支持簡單的發布者確認。

將 配置為使用單獨的連接時，現在可以從版本 2.3.2 開始，將發布連接工廠配置為其他類型。 默認情況下，發布工廠的類型相同，在主工廠上設置的任何屬性也會傳播到發布工廠。??RabbitTemplate??

??PooledChannelConnectionFactory??

此工廠基于 Apache Pool2 管理單個連接和兩個通道池。 一個池用于事務通道，另一個池用于非事務通道。 池是默認配置的;提供回調以配置池;有關更多信息，請參閱 Apache 文檔。??GenericObjectPool??

Apache jar 必須位于類路徑上才能使用此工廠。??commons-pool2??

@BeanPooledChannelConnectionFactory pcf() throws Exception {    ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();    rabbitConnectionFactory.setHost("localhost");    PooledChannelConnectionFactory pcf = new PooledChannelConnectionFactory(rabbitConnectionFactory);    pcf.setPoolConfigurer((pool, tx) -> {        if (tx) {            // configure the transactional pool        }        else {            // configure the non-transactional pool        }    });    return pcf;}

??ThreadChannelConnectionFactory??

此工廠管理單個連接和兩個連接，一個用于事務通道，另一個用于非事務通道。 此工廠確保同一線程上的所有操作都使用相同的通道（只要它保持打開狀態）。 這有助于嚴格的消息排序，而無需作用域內操作。 為避免內存泄漏，如果應用程序使用許多生存期較短的線程，則必須調用工廠的線程來釋放通道資源。 從版本 2.3.7 開始，線程可以將其通道傳輸到另一個線程。 有關詳細信息，請參閱多線程環境中的嚴格消息排序。??ThreadLocal????closeThreadChannel()??

??CachingConnectionFactory??

提供的第三個實現是 ，默認情況下，它建立可由應用程序共享的單個連接代理。 共享連接是可能的，因為使用 AMQP 進行消息傳遞的“工作單元”實際上是一個“通道”（在某些方面，這類似于 JMS 中的連接和會話之間的關系）。 連接實例提供了一個方法。 該實現支持緩存這些通道，并根據通道是否為事務性為通道維護單獨的緩存。 創建 的實例時，可以通過構造函數提供“主機名”。 您還應該提供“用戶名”和“密碼”屬性。 若要配置通道緩存的大小（默認值為 25），可以調用該方法。??CachingConnectionFactory????createChannel????CachingConnectionFactory????CachingConnectionFactory????setChannelCacheSize()??

從版本 1.3 開始，您可以配置緩存連接以及僅通道。 在這種情況下，每次調用 都會創建一個新連接（或從緩存中檢索空閑連接）。 關閉連接會將其返回到緩存（如果尚未達到緩存大小）。 在此類連接上創建的通道也會被緩存。 在某些環境中使用單獨的連接可能很有用，例如從 HA 群集使用，在 與負載均衡器配合使用，以連接到不同的集群成員等。 要緩存連接，請將 設置為 。??CachingConnectionFactory????createConnection()????cacheMode????CacheMode.CONNECTION??

從版本 1.5.5 開始，提供了一個名為的新屬性。 設置此屬性時，它將限制允許的連接總數。 設置后，如果達到限制，則用于等待連接變為空閑狀態。 如果超過該時間，則拋出 。??connectionLimit????channelCheckoutTimeLimit????AmqpTimeoutException??

重要的是要了解緩存大小（默認情況下）不是限制，而只是可以緩存的通道數。 如果緩存大小為 10，則實際上可以使用任意數量的通道。 如果使用的通道超過 10 個，并且它們都返回到緩存中，則緩存中將有 10 個通道。 其余部分是物理關閉的。

從版本 1.6 開始，默認通道緩存大小已從 1 增加到 25。 在高容量、多線程環境中，小緩存意味著以高速率創建和關閉通道。 增加默認緩存大小可以避免此開銷。 您應該通過 RabbitMQ 管理 UI 監控正在使用的通道，并考慮進一步增加緩存大小，如果您 看到許多頻道正在創建和關閉。 緩存僅按需增長（以滿足應用程序的并發要求），因此此更改不會 影響現有的小批量應用程序。

從版本 1.4.2 開始，具有一個名為 的屬性。 當此屬性大于零時，將限制可在連接上創建的通道數。 如果達到限制，調用線程將阻塞，直到通道可用或達到此超時，在這種情況下，將拋出 a。??CachingConnectionFactory????channelCheckoutTimeout????channelCacheSize????AmqpTimeoutException??

以下示例演示如何創建新的 ：??connection??

CachingConnectionFactory connectionFactory = new CachingConnectionFactory("somehost");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");Connection connection = connectionFactory.createConnection();

使用 XML 時，配置可能類似于以下示例：

通過使用 rabbit 命名空間可以快速方便地創建 A，如下所示：??ConnectionFactory??

在大多數情況下，此方法更可取，因為框架可以為您選擇最佳默認值。 創建的實例是 . 請記住，通道的默認緩存大小為 25。 如果要緩存更多通道，請通過設置“channelCacheSize”屬性來設置更大的值。 在 XML 中，它如下所示：??CachingConnectionFactory??

此外，使用命名空間，您可以添加“channel-cache-size”屬性，如下所示：

默認緩存模式為 ，但您可以將其配置為緩存連接。 在下面的示例中，我們使用：??CHANNEL????connection-cache-size??

可以使用命名空間提供主機和端口屬性，如下所示：

或者，如果在群集環境中運行，則可以使用 addresses 屬性，如下所示：

有關 的信息，請參閱連接到群集。??address-shuffle-mode??

以下示例具有自定義線程工廠，該工廠的線程名稱前綴為 ：??rabbitmq-??

地址解析器

從版本 2.1.15 開始，您現在可以使用 解析連接地址。 這將覆蓋 和 屬性的任何設置。??AddressResolver????addresses????host/port??

命名連接

從版本 1.7 開始，提供了用于注入 . 生成的名稱用于目標 RabbitMQ 連接的特定于應用程序的標識。 如果 RabbitMQ 服務器支持連接名稱，則連接名稱將顯示在管理 UI 中。 此值不必是唯一的，也不能用作連接標識符，例如，在 HTTP API 請求中。 此值應該是人類可讀的，并且是鍵下的一部分。 您可以使用簡單的 Lambda，如下所示：??ConnectionNameStrategy????AbstractionConnectionFactory????ClientProperties????connection_name??

connectionFactory.setConnectionNameStrategy(connectionFactory -> "MY_CONNECTION");

該參數可用于通過某些邏輯區分目標連接名稱。 默認情況下，的 、表示對象的十六進制字符串和內部計數器用于生成 . 命名空間組件也隨屬性一起提供。??ConnectionFactory????beanName????AbstractConnectionFactory????connection_name????????connection-name-strategy??

的實現將連接名稱設置為應用程序屬性。 可以將其聲明為 并將其注入到連接工廠中，如以下示例所示：??SimplePropertyValueConnectionNameStrategy????@Bean??

@Beanpublic SimplePropertyValueConnectionNameStrategy cns() {    return new SimplePropertyValueConnectionNameStrategy("spring.application.name");}@Beanpublic ConnectionFactory rabbitConnectionFactory(ConnectionNameStrategy cns) {    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();    ...    connectionFactory.setConnectionNameStrategy(cns);    return connectionFactory;}

該屬性必須存在于應用程序上下文的 .??Environment??

阻止的連接和資源限制

連接可能會被阻止，無法與內存警報對應的代理進行交互。 從版本 2.0 開始，可以為 提供要通知連接阻止和取消阻止事件的實例。 此外，分別通過其內部實現發出 a 和 。 這些允許您提供應用程序邏輯，以便對代理上的問題做出適當的反應，并（例如）采取一些糾正措施。??org.springframework.amqp.rabbit.connection.Connection????com.rabbitmq.client.BlockedListener????AbstractConnectionFactory????ConnectionBlockedEvent????ConnectionUnblockedEvent????BlockedListener??

從版本 2.0.2 開始，具有自動使用第二個連接工廠的配置選項，除非正在使用事務。 有關詳細信息，請參閱使用單獨的連接。 對于發布者連接與主策略相同，并附加到調用方法的結果中。??RabbitTemplate????ConnectionNameStrategy????.publisher??

從版本 1.7.7 開始，提供了 an，當無法創建 a 時會拋出該 （例如，因為達到限制并且緩存中沒有可用通道）。 您可以在 中使用此異常在 進行一些退避后恢復操作。??AmqpResourceNotAvailableException????SimpleConnection.createChannel()????Channel????channelMax????RetryPolicy??

配置基礎客戶端連接工廠

使用 Rabbit 客戶端的實例。 在 上設置等效屬性時，會傳遞許多配置屬性（、 和 等）。 若要設置其他屬性（例如），可以定義 Rabbit 工廠的實例，并使用 的相應構造函數提供對它的引用。 使用命名空間（如前所述）時，需要在屬性中提供對已配置工廠的引用。 為方便起見，提供了一個工廠 Bean 來幫助在 Spring 應用程序上下文中配置連接工廠，如下一節所述。??CachingConnectionFactory????ConnectionFactory????host????port????userName????password????requestedHeartBeat????connectionTimeout????CachingConnectionFactory????clientProperties????CachingConnectionFactory????connection-factory??

??RabbitConnectionFactoryBean??和配置 SSL

從版本 1.4 開始，提供了方便的功能，以便使用依賴關系注入在基礎客戶端連接工廠上方便地配置 SSL 屬性。 其他二傳手委托給底層工廠。 以前，您必須以編程方式配置 SSL 選項。 以下示例演示如何配置：??RabbitConnectionFactoryBean????RabbitConnectionFactoryBean??

有關配置 SSL 的信息，請參閱RabbitMQ 文檔。 省略 和 配置以在不進行證書驗證的情況下通過 SSL 進行連接。 下一個示例演示如何提供密鑰和信任存儲區配置。??keyStore????trustStore??

該屬性是一個 Spring 指向包含以下鍵的屬性文件：??sslPropertiesLocation????Resource??

keyStore=file:/secret/keycert.p12trustStore=file:/secret/trustStorekeyStore.passPhrase=secrettrustStore.passPhrase=secret

和是春天指著商店。 通常，此屬性文件由操作系統保護，應用程序具有讀取訪問權限。??keyStore????truststore????Resources??

從 Spring AMQP 版本 1.5 開始，您可以直接在工廠 Bean 上設置這些屬性。 如果同時提供了離散屬性 和，則后者中的屬性將覆蓋 離散值。??sslPropertiesLocation??

連接到群集

要連接到群集，請在 上配置屬性：??addresses????CachingConnectionFactory??

@Beanpublic CachingConnectionFactory ccf() {    CachingConnectionFactory ccf = new CachingConnectionFactory();    ccf.setAddresses("host1:5672,host2:5672,host3:5672");    return ccf;}

從版本 3.0 開始，每當建立新連接時，基礎連接工廠將通過選擇隨機地址來嘗試連接到主機。 若要恢復到以前嘗試從第一個連接到最后一個的行為，請將該屬性設置為 。??addressShuffleMode????AddressShuffleMode.NONE??

從版本 2.3 開始，添加了隨機模式，這意味著在創建連接后，第一個地址將移動到末尾。 如果您希望從所有節點上的所有分片消費，您可能希望將此模式與RabbitMQ 分片插件一起使用，并具有合適的并發性。??INORDER????CacheMode.CONNECTION??

@Beanpublic CachingConnectionFactory ccf() {    CachingConnectionFactory ccf = new CachingConnectionFactory();    ccf.setAddresses("host1:5672,host2:5672,host3:5672");    ccf.setAddressShuffleMode(AddressShuffleMode.INORDER);    return ccf;}

路由連接工廠

從 1.3 版開始，引入了 。 此工廠提供了一種機制，用于為多個映射配置映射，并在運行時確定某些映射的目標。 通常，實現會檢查線程綁定上下文。 為方便起見，Spring AMQP 提供了 ，它從 獲取當前線程綁定。 以下示例顯示了如何在 XML 和 Java 中配置 a：??AbstractRoutingConnectionFactory????ConnectionFactories????ConnectionFactory????lookupKey????SimpleRoutingConnectionFactory????lookupKey????SimpleResourceHolder????SimpleRoutingConnectionFactory??

public class MyService {    @Autowired    private RabbitTemplate rabbitTemplate;    public void service(String vHost, String payload) {        SimpleResourceHolder.bind(rabbitTemplate.getConnectionFactory(), vHost);        rabbitTemplate.convertAndSend(payload);        SimpleResourceHolder.unbind(rabbitTemplate.getConnectionFactory());    }}

使用后取消綁定資源非常重要。 有關更多信息，請參見 的JavaDoc。??AbstractRoutingConnectionFactory??

從版本 1.4 開始，支持 SpEL 和屬性，它們在每個 AMQP 協議交互操作（、、或）上進行評估，解析為所提供的值。 您可以使用 Bean 引用，例如在表達式中。 對于操作，要發送的消息是根評估對象。 對于操作，是根評估對象。??RabbitTemplate????sendConnectionFactorySelectorExpression????receiveConnectionFactorySelectorExpression????send????sendAndReceive????receive????receiveAndReply????lookupKey????AbstractRoutingConnectionFactory????@vHostResolver.getVHost(#root)????send????receive????queueName??

路由算法如下：如果選擇器表達式被計算或被計算到，或者提供的不是 的實例，則一切都像以前一樣工作，依賴于提供的實現。 如果評估結果不是 ，但沒有目標，并且 配置了 。 在 的情況下，它確實回退到基于 的實現。 但是，如果拋出 ，則拋出 an。??null????null????ConnectionFactory????AbstractRoutingConnectionFactory????ConnectionFactory????null????ConnectionFactory????lookupKey????AbstractRoutingConnectionFactory????lenientFallback = true????AbstractRoutingConnectionFactory????routing????determineCurrentLookupKey()????lenientFallback = false????IllegalStateException??

命名空間支持還提供組件上的 and 屬性。??send-connection-factory-selector-expression????receive-connection-factory-selector-expression??????

此外，從版本 1.4 開始，您可以在偵聽器容器中配置路由連接工廠。 在這種情況下，隊列名稱列表將用作查找鍵。 例如，如果使用 配置容器，則查找鍵為 （請注意，鍵中沒有空格）。??setQueueNames("thing1", "thing2")????[thing1,thing]"??

從版本 1.6.9 開始，可以通過在偵聽器容器上使用向查找鍵添加限定符。 例如，這樣做可以偵聽具有相同名稱但位于不同虛擬主機中的隊列（每個虛擬主機都有一個連接工廠）。??setLookupKeyQualifier??

例如，對于查找鍵限定符和偵聽隊列的容器，您可以向其注冊目標連接工廠的查找鍵可以是 。??thing1????thing2????thing1[thing2]??

從版本 2.4.4 開始，可以禁用此驗證。 如果確認和返回之間的值需要不相等，則可以使用 來關閉驗證。 請注意，添加到的第一個連接工廠將確定 和 的一般值。??AbstractRoutingConnectionFactory#setConsistentConfirmsReturns????AbstractRoutingConnectionFactory????confirms????returns??

如果您遇到要檢查的某些消息確認/返回而其他消息則不會確認/返回的情況，這可能會很有用。 例如：

@Beanpublic RabbitTemplate rabbitTemplate() {    final com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();    cf.setHost("localhost");    cf.setPort(5672);    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(cf);    cachingConnectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);    PooledChannelConnectionFactory pooledChannelConnectionFactory = new PooledChannelConnectionFactory(cf);    final Map connectionFactoryMap = new HashMap<>(2);    connectionFactoryMap.put("true", cachingConnectionFactory);    connectionFactoryMap.put("false", pooledChannelConnectionFactory);    final AbstractRoutingConnectionFactory routingConnectionFactory = new SimpleRoutingConnectionFactory();    routingConnectionFactory.setConsistentConfirmsReturns(false);    routingConnectionFactory.setDefaultTargetConnectionFactory(pooledChannelConnectionFactory);    routingConnectionFactory.setTargetConnectionFactories(connectionFactoryMap);    final RabbitTemplate rabbitTemplate = new RabbitTemplate(routingConnectionFactory);    final Expression sendExpression = new SpelExpressionParser().parseExpression(            "messageProperties.headers["x-use-publisher-confirms"] ?: false");    rabbitTemplate.setSendConnectionFactorySelectorExpression(sendExpression);}

這樣，帶有標頭的消息將通過緩存連接發送，您可以確保消息傳遞。 有關確保郵件傳遞的詳細信息，請參閱發布者確認并返回。??x-use-publisher-confirms: true??

隊列相關性和??LocalizedQueueConnectionFactory??

在集群中使用 HA 隊列時，為了獲得最佳性能，您可能需要連接到物理代理 潛在顧客隊列所在的位置。 可以配置多個代理地址。 這是為了進行故障轉移，客戶端會嘗試按順序進行連接。 使用管理插件提供的 REST API 來確定哪個節點是隊列的引導節點。 然后，它創建（或從緩存中檢索）僅連接到該節點的節點。 如果連接失敗，則確定新的引導節點，使用者連接到該節點。 配置了默認連接工廠，以防無法確定隊列的物理位置，在這種情況下，它會正常連接到群集。??CachingConnectionFactory????LocalizedQueueConnectionFactory????CachingConnectionFactory????LocalizedQueueConnectionFactory??

是 和 使用隊列名稱作為查找鍵，如上面的路由連接工廠中所述。??LocalizedQueueConnectionFactory????RoutingConnectionFactory????SimpleMessageListenerContainer??

以下示例配置顯示了如何配置工廠：

@Autowiredprivate ConfigurationProperties props;@Beanpublic CachingConnectionFactory defaultConnectionFactory() {    CachingConnectionFactory cf = new CachingConnectionFactory();    cf.setAddresses(this.props.getAddresses());    cf.setUsername(this.props.getUsername());    cf.setPassword(this.props.getPassword());    cf.setVirtualHost(this.props.getVirtualHost());    return cf;}@Beanpublic LocalizedQueueConnectionFactory queueAffinityCF(        @Qualifier("defaultConnectionFactory") ConnectionFactory defaultCF) {    return new LocalizedQueueConnectionFactory(defaultCF,            StringUtils.commaDelimitedListToStringArray(this.props.getAddresses()),            StringUtils.commaDelimitedListToStringArray(this.props.getAdminUris()),            StringUtils.commaDelimitedListToStringArray(this.props.getNodes()),            this.props.getVirtualHost(), this.props.getUsername(), this.props.getPassword(),            false, null);}

請注意，前三個參數是 、 和 的數組。 這些是位置性的，因為當容器嘗試連接到隊列時，它會使用管理 API 來確定哪個節點是隊列的引導節點，并連接到與該節點位于同一陣列位置的地址。??addresses????adminUris????nodes??

添加到類路徑：??WebFlux??

例 1.馬文

  org.springframework.amqp  spring-rabbit

例 2.格拉德爾

compile "org.springframework.amqp:spring-rabbit"

您還可以通過實現和重寫其 和（可選）方法來使用其他 REST 技術。??LocalizedQueueConnectionFactory.NodeLocator????createClient, ``restCall????close??

lqcf.setNodeLocator(new NodeLocator() {    @Override    public MyClient createClient(String userName, String password) {        ...    }    @Override    public HashMap restCall(MyClient client, URI uri) {        ...    });});

該框架提供 和 ，默認值如上所述。??WebFluxNodeLocator????RestTemplateNodeLocator??

發布商確認并返回

通過將屬性設置為“true”和“true”來支持已確認（具有相關性）和返回的消息。??CachingConnectionFactory????publisherConfirmType????ConfirmType.CORRELATED????publisherReturns??

設置這些選項后，工廠創建的實例將包裝在 中，用于方便回調。 當獲得這樣的通道時，客戶端可以向 . 該實現包含用于路由確認或返回到相應偵聽器的邏輯。 以下各節將進一步介紹這些功能。??Channel????PublisherCallbackChannel????PublisherCallbackChannel.Listener????Channel????PublisherCallbackChannel??

另請參閱??作用域內操作??。??simplePublisherConfirms??

連接和通道偵聽器

連接工廠支持注冊和實現。 這允許您接收連接和通道相關事件的通知。 （建立連接時 使用 A 執行聲明 - 有關詳細信息，請參閱交換、隊列和綁定的自動聲明）。 以下清單顯示了接口定義：??ConnectionListener????ChannelListener????ConnectionListener????RabbitAdmin????ConnectionListener??

@FunctionalInterfacepublic interface ConnectionListener {    void onCreate(Connection connection);    default void onClose(Connection connection) {    }    default void onShutDown(ShutdownSignalException signal) {    }}

從版本 2.0 開始，可以為對象提供實例，以便在連接阻止和取消阻止事件時收到通知。 以下示例顯示了通道偵聽器接口定義：??org.springframework.amqp.rabbit.connection.Connection????com.rabbitmq.client.BlockedListener??

@FunctionalInterfacepublic interface ChannelListener {    void onCreate(Channel channel, boolean transactional);    default void onShutDown(ShutdownSignalException signal) {    }}

請參閱發布是異步的 — 如何檢測成功和失敗，了解您可能希望注冊 的一種方案。??ChannelListener??

記錄通道關閉事件

版本 1.5 引入了一種機制，使用戶能夠控制日志記錄級別。

使用默認策略記錄通道關閉，如下所示：??CachingConnectionFactory??

若要修改此行為，可以將自定義注入到 in 屬性中。??ConditionalExceptionLogger????CachingConnectionFactory????closeExceptionLogger??

另請參閱使用者事件。

運行時緩存屬性

與 1.6 版本相同，現在通過該方法提供緩存統計信息。 這些統計信息可用于調整緩存，以在生產中對其進行優化。 例如，高水位線可用于確定是否應增加緩存大小。 如果它等于緩存大小，則可能需要考慮進一步增加。 下表描述了這些屬性：??CachingConnectionFactory????getCacheProperties()????CacheMode.CHANNEL??

表 1.CacheMode.CHANNEL 的緩存屬性

connectionName

channelCacheSize

localPort

idleChannelsTx

idleChannelsNotTx

idleChannelsTxHighWater

idleChannelsNotTxHighWater

下表描述了這些屬性：??CacheMode.CONNECTION??

表 2.CacheMode.CONNECTION 的緩存屬性

connectionName:

openConnections

channelCacheSize

connectionCacheSize

idleConnections

idleConnectionsHighWater

idleChannelsTx:

idleChannelsNotTx:

idleChannelsTxHighWater:

idleChannelsNotTxHighWater:

屬性（或）也包括在內。??cacheMode????CHANNEL????CONNECTION??

圖1.JVisualVM 示例

RabbitMQ 自動連接/拓撲恢復

自Spring AMQP的第一個版本以來，該框架在代理發生故障時提供了自己的連接和通道恢復。 此外，如配置代理中所述，在重新建立連接時，會重新聲明任何基礎結構 Bean（隊列和其他）。 因此，它不依賴于庫現在提供的自動恢復。 默認情況下，已啟用自動恢復。 兩種恢復機制之間存在一些不兼容之處，因此默認情況下，Spring 將底層的屬性設置為 。 即使該屬性是 ，Spring 也會通過立即關閉任何恢復的連接來有效地禁用它。??RabbitAdmin????amqp-client????amqp-client????automaticRecoveryEnabled????RabbitMQ connectionFactory????false????true??

4.1.3. 添加自定義客戶端連接屬性

現在允許您訪問基礎連接工廠以允許，例如， 設置自定義客戶端屬性。 以下示例演示如何執行此操作：??CachingConnectionFactory??

connectionFactory.getRabbitConnectionFactory().getClientProperties().put("thing1", "thing2");

查看連接時，這些屬性將顯示在 RabbitMQ 管理 UI 中。

4.1.4.??AmqpTemplate??

與 Spring 框架和相關項目提供的許多其他高級抽象一樣，Spring AMQP 提供了一個發揮核心作用的“模板”。 定義主操作的接口稱為 。 這些操作涵蓋了發送和接收消息的一般行為。 換句話說，它們不是任何實現所獨有的——因此名稱中的“AMQP”。 另一方面，該接口的實現與AMQP協議的實現相關聯。 與本身是接口級 API 的 JMS 不同，AMQP 是一種線級協議。 該協議的實現提供自己的客戶端庫，因此模板接口的每個實現都依賴于特定的客戶端庫。 目前，只有一個實現：。 在下面的示例中，我們經常使用 . 但是，當您查看配置示例或實例化模板或調用資源庫的任何代碼摘錄時，您可以看到實現類型（例如，）。??AmqpTemplate????RabbitTemplate????AmqpTemplate????RabbitTemplate??

如前所述，該接口定義了發送和接收消息的所有基本操作。 我們將分別在發送消息和接收消息中探討消息發送和接收。??AmqpTemplate??

另請參閱異步兔子模板。

添加重試功能

從版本 1.3 開始，您現在可以將 配置為使用 來幫助處理代理連接問題。 有關完整信息，請參閱春季重試項目。 以下只是一個使用指數退避策略和默認值的示例，該策略在將異常拋給調用方之前進行了三次嘗試。??RabbitTemplate????RetryTemplate????SimpleRetryPolicy??

下面的示例使用 XML 命名空間：

以下示例使用 Java 中的注釋：??@Configuration??

@Beanpublic RabbitTemplate rabbitTemplate() {    RabbitTemplate template = new RabbitTemplate(connectionFactory());    RetryTemplate retryTemplate = new RetryTemplate();    ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();    backOffPolicy.setInitialInterval(500);    backOffPolicy.setMultiplier(10.0);    backOffPolicy.setMaxInterval(10000);    retryTemplate.setBackOffPolicy(backOffPolicy);    template.setRetryTemplate(retryTemplate);    return template;}

從版本 1.4 開始，除了屬性之外，該選項在 上受支持。 它用作 的第二個參數。??retryTemplate????recoveryCallback????RabbitTemplate????RetryTemplate.execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback)??

retryTemplate.execute(    new RetryCallback() {        @Override        public Object doWithRetry(RetryContext context) throws Exception {            context.setAttribute("message", message);            return rabbitTemplate.convertAndSend(exchange, routingKey, message);        }    }, new RecoveryCallback