RabbitMQ 高级特性介绍

RabbitMQ 是一个基于AMQP（高级消息队列协议）的开源消息队列系统，是在互联网中广泛应用的 middleware 中的佼佼者。它的主要特性是，使得应用程序能够通过消息交互而不是直接耦合。消息本质上是一些数据，这意味着它们可以非常快地在应用程序之间传递。RabbitMQ 拥有诸多优秀的特性，本文即将对 RabbitMQ 的特性进行全面介绍。

一、消息可靠性

RabbitMQ 拥有强大的消息可靠性，包括持久化、消息确认、事务等多个方面。

1、持久化


try {
  channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, body.getBytes("UTF-8"));
} catch ...

上述代码中，将消息属性设置为 PERSISTENT_TEXT_PLAIN 后，消息将会被持久化到 RabbitMQ 中，确保即使 RabbitMQ 的服务意外崩溃，消息也能被恢复并传递。

2、消息确认


channel.confirmSelect();
channel.basicPublish(exchangeName, routingKey, null, body.getBytes("UTF-8"));
if (channel.waitForConfirms()) {
  System.out.println("Message Publish Success.");
}

为了确保消息能够准确地传递，消息的发送者需要向 RabbitMQ 请求确认消息是否成功发送。上述示例中，我们使用了 channel.confirmSelect() 开启了消息确认模式，并且使用 channel.waitForConfirms() 进行消息状态的确认。

二、灵活的消息路由策略

RabbitMQ 支持复杂的消息路由策略，无论是多重绑定还是通配符路由，都可以轻松配置实现。

1、Direct Exchange


// 创建 exchange
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
// 绑定 queue
channel.queueBind(queueName, exchangeName, routingKey);
// 发送消息
channel.basicPublish(exchangeName, routingKey, null, body.getBytes("UTF-8"));

上述代码片段展示了直接交换机的使用方式。通过 exchangeDeclare() 创建直接交换机，通过 queueBind() 绑定队列。在发送消息时使用交换机名和路由键，指定消息路由规则。

2、Topic Exchange


// 创建 exchange
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
// 绑定 queue
channel.queueBind(queueName, exchangeName, routingPattern);
// 发送消息
channel.basicPublish(exchangeName, routingKey, null, body.getBytes("UTF-8"));

Topic Exchange 是 RabbitMQ 更灵活的交换机类型，支持通配符规则进行路由。上述代码展示了 Topic Exchange 的操作方式，通过为 exchangeDeclare() 指定交换机类型为 TOPIC 创建 Topic Exchange，消息在发送时使用 routingKey 来匹配 routingPattern。

三、负载均衡和消息处理

在 RabbitMQ 中，可以通过多重绑定和消费者数量控制来实现负载均衡。

1、多重绑定


channel.queueBind(queueName, exchangeName, routingKey1);
channel.queueBind(queueName, exchangeName, routingKey2);
channel.queueBind(queueName, exchangeName, routingKey3);

通过为队列使用多个 routingKey 进行绑定，可以将多个消息源的消息传递给同一个队列。这样，队列内消息数量就会增加，自动根据消费者的数量进行负载均衡。

2、消费者数量控制


channel.basicQos(1);
channel.basicConsume(queueName, false, consumer);

通过 basicQos() 方法控制消费者数量，可以确保每个消费者最多只能处理一个消息。这样，当队列中的消息数量增加时，系统仅仅会增加更多的消费者用于消费消息，而不会因为单个消费者过多导致消息堆积。

四、Spring Boot 集成

RabbitMQ 支持在 Spring Boot 中方便的集成使用。

1、依赖管理



  org.springframework.boot
  spring-boot-starter-amqp

在项目的 pom.xml 文件中添加以上依赖，即可引入 Spring Boot 对 RabbitMQ 的支持。

2、配置 Application


@RestController
@SpringBootApplication
public class Application {

  public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }

  @Bean
  public Queue queue() {
    return new Queue("test", true);
  }

  @RabbitListener(queues = "test")
  public void processMessage(String content) {
    System.out.println("Received message: " + content);
  }
}

在 Application 中，通过 @Bean 创建了一个 Queue 对象，并通过 @RabbitListener 注解来监听该队列，一旦有消息到达，就会为其调用 processMessage() 方法进行处理。

五、总结

RabbitMQ 提供了强大的消息传递能力，支持消息可靠性、灵活的消息路由策略、负载均衡和消息处理等特性。此外，在 Spring Boot 中集成 RabbitMQ 也非常便捷。当您需要进行应用间消息传递的时候，RabbitMQ 显然是一个不错的选择。

原创文章，作者：小蓝，如若转载，请注明出处：https://www.506064.com/n/150913.html