RabbitMQ 的消息持久化与 Spring AMQP 的实现详解

// 参数1 exchange ：交换器名
// 参数2 type ：交换器类型
// 参数3 durable ：是否持久化
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);

// 参数1 queue ：队列名
// 参数2 durable ：是否持久化
// 参数3 exclusive ：仅创建者可以使用的私有队列，断开后自动删除
// 参数4 autoDelete : 当所有消费客户端连接断开后，是否自动删除队列
// 参数5 arguments
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

// 参数1 exchange ：交换器
// 参数2 routingKey ： 路由键
// 参数3 props ： 消息的其他参数,其中 MessageProperties.PERSISTENT_TEXT_PLAIN 表示持久化
// 参数4 body ： 消息体
channel.basicPublish("", queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

// 参数1 name ：交互器名
// 参数2 durable ：是否持久化
// 参数3 autoDelete ：当所有消费客户端连接断开后，是否自动删除队列
new TopicExchange(name, durable, autoDelete)

// 参数1 name ：队列名
// 参数2 durable ：是否持久化
// 参数3 exclusive ：仅创建者可以使用的私有队列，断开后自动删除
// 参数4 autoDelete : 当所有消费客户端连接断开后，是否自动删除队列
new Queue(name, durable, exclusive, autoDelete);

rabbitTemplate.convertAndSend(exchange, routeKey, message);

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

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

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

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);
 }
}