快捷導(dǎo)航

RabbitMQ 的消息持久化與 Spring AMQP 的實(shí)現(xiàn)詳解

更新時(shí)間：2019年08月22日 09:16:34 作者：梁桂釗

這篇文章主要介紹了RabbitMQ 的消息持久化與 Spring AMQP 的實(shí)現(xiàn)剖析詳解,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

前言

要從奔潰的 RabbitMQ 中恢復(fù)的消息，我們需要做消息持久化。如果消息要從 RabbitMQ 奔潰中恢復(fù)，那么必須滿足三點(diǎn)，且三者缺一不可。

交換器必須是持久化。
隊(duì)列必須是持久化的。
消息必須是持久化的。

原生的實(shí)現(xiàn)方式

原生的 RabbitMQ 客戶端需要完成三個(gè)步驟。

第一步，交換器的持久化。

// 參數(shù)1 exchange ：交換器名
// 參數(shù)2 type ：交換器類型
// 參數(shù)3 durable ：是否持久化
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);

第二步，隊(duì)列的持久化。

// 參數(shù)1 queue ：隊(duì)列名
// 參數(shù)2 durable ：是否持久化
// 參數(shù)3 exclusive ：僅創(chuàng)建者可以使用的私有隊(duì)列，斷開(kāi)后自動(dòng)刪除
// 參數(shù)4 autoDelete : 當(dāng)所有消費(fèi)客戶端連接斷開(kāi)后，是否自動(dòng)刪除隊(duì)列
// 參數(shù)5 arguments
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

第三步，消息的持久化。

// 參數(shù)1 exchange ：交換器
// 參數(shù)2 routingKey ： 路由鍵
// 參數(shù)3 props ： 消息的其他參數(shù),其中 MessageProperties.PERSISTENT_TEXT_PLAIN 表示持久化
// 參數(shù)4 body ： 消息體
channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

Spring AMQP 的實(shí)現(xiàn)方式

Spring AMQP 是對(duì)原生的 RabbitMQ 客戶端的封裝。一般情況下，我們只需要定義交換器的持久化和隊(duì)列的持久化。

其中，交換器的持久化配置如下。

// 參數(shù)1 name ：交互器名
// 參數(shù)2 durable ：是否持久化
// 參數(shù)3 autoDelete ：當(dāng)所有消費(fèi)客戶端連接斷開(kāi)后，是否自動(dòng)刪除隊(duì)列
new TopicExchange(name, durable, autoDelete)

此外，還需要再配置隊(duì)列的持久化。

// 參數(shù)1 name ：隊(duì)列名
// 參數(shù)2 durable ：是否持久化
// 參數(shù)3 exclusive ：僅創(chuàng)建者可以使用的私有隊(duì)列，斷開(kāi)后自動(dòng)刪除
// 參數(shù)4 autoDelete : 當(dāng)所有消費(fèi)客戶端連接斷開(kāi)后，是否自動(dòng)刪除隊(duì)列
new Queue(name, durable, exclusive, autoDelete);

至此，RabbitMQ 的消息持久化配置完畢。

那么，消息的持久化難道不需要配置么？確實(shí)如此，我們來(lái)看下源碼。

一般情況下，我們會(huì)通過(guò)這種方式發(fā)送消息。

rabbitTemplate.convertAndSend(exchange, routeKey, message);

其中，調(diào)用了 convertAndSend(String exchange, String routingKey, final Object object) 方法。

@Override
public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException {
 convertAndSend(exchange, routingKey, object, (CorrelationData) null);
}

接著，用調(diào)用了 convertAndSend(String exchange, String routingKey, final Object object, CorrelationData correlationData) 方法。

public void convertAndSend(String exchange, String routingKey, final Object object, CorrelationData correlationData) throws AmqpException {
  send(exchange, routingKey, convertMessageIfNecessary(object), correlationData);
 }

此時(shí)，最關(guān)鍵的方法出現(xiàn)了，它是 convertMessageIfNecessary(final Object object)。

protected Message convertMessageIfNecessary(final Object object) {
 if (object instanceof Message) {
  return (Message) object;
 }
 return getRequiredMessageConverter().toMessage(object, new MessageProperties());
}

其中，關(guān)鍵的是 MessageProperties 類,它持久化的策略是 MessageDeliveryMode.PERSISTENT，因此它會(huì)初始化時(shí)默認(rèn)消息是持久化的。

public class MessageProperties implements Serializable {
 public MessageProperties() {
  this.deliveryMode = DEFAULT_DELIVERY_MODE;
  this.priority = DEFAULT_PRIORITY;
 }
 static {
  DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;
  DEFAULT_PRIORITY = Integer.valueOf(0);
 }
}

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: