Spring Boot 构建应用——整合消息中间件 RabbitMQ

RabbitMQ 是消息中间件的一种，实现了 AMQP 标准。消息中间件的工作过程可以用生产者-消费者模型来表示。生产者发消息到消息队列中去，消费者监听指定的消息队列，并且当消息队列收到消息之后，接收消息队列传来的消息，并且给予相应的处理。消息队列常用于分布式系统之间互相信息的传递。

对于 RabbitMQ 来说，这三个基本模块以外，还添加了一个模块，交换机 (Exchange)。它使得生产者和消息队列之间产生了隔离，生产者将消息发送给交换机，而交换机则根据调度策略把相应的消息转发给对应的消息队列。

交换机的主要作用是接收相应的消息并且绑定到指定的队列，有三种类型的交换机：

交换机	说明
direct	默认的交换机 (一对一)。即创建消息队列的时候，指定一个BindingKey，当生产者发送消息的时候，指定对应的Key，当Key和消息队列的BindingKey一致的时候，消息将会被发送到该消息队列中。
fanout	路由广播的形式，将会把消息发给绑定它的全部队列，即便设置了key，也会被忽略。生产者其实仅关注Exchange与Route Key, 消费者仅关注Queue
topic	Topic转发信息主要是依据通配符，队列和交换机的绑定主要是依据一种模式(通配符+字符串)，而当发送消息的时候，只有指定的Key和该模式相匹配的时候，消息才会被发送到该消息队列中。

Spring Boot 整合 RabbitMQ 非常简单，需要添加 Maven 起步依赖：

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

一些核心类如下：

org.springframework.amqp.core.Queue: 队列
org.springframework.amqp.core.Binding: 建立交换机与队列的绑定关系
org.springframework.amqp.core.DirectExchange: Direct交换机
org.springframework.amqp.core.TopicExchange: Topic交换机
org.springframework.amqp.core.FanoutExchange: Fanout交换机
org.springframework.amqp.support.converter.MessageConverter: 消息转换器, 如将Java类转换JSON类型发送至Broker, 从Broker处获取JSON消息转换为Java类型
org.springframework.amqp.core.AmqpTemplate: 多用于生产者端发布消息
org.springframework.amqp.core.AmqpAdmin: 用于Exchange, Queue等的动态管理

然后通过 application.properties 中的 spring.rabbitmq.* 前缀配置属性（生产者、消费者应用都一样）：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
spring.rabbitmq.virtual-host=/admin

下面看一下在三种类型的交换机下的不同实现。

1.Spring Boot整合RabbitMQ (direct交换机)

在 Java 配置类中注册 bean（生产者和消费者的是一样的，因为监听的是同一个队列，所以队列名要先约定好）：

@Configuration
public class RabbitConfiguration {
//队列名
public static final String TRADE_QUEUE = "funds";
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public Queue queue() {
return new Queue(TRADE_QUEUE);
}
}

1.生产者生产消息

在 SpringBoot 中，我们使用 AmqpTemplate 去发送消息（调用 send 方法即发送）：

@Component
public class HelloProducer {
@Autowired
private AmqpTemplate amqpTemplate;
public void send() {
amqpTemplate.convertAndSend(RabbitConfiguration.TRADE_QUEUE, "Hello, Rabbit!");
}
}

2.消费者消费消息

配置监听器监听指定的 Queue，当消息队列有消息的时候予以接收：

@Component
public class HelloConsumer {
@RabbitListener(queues = {RabbitConfiguration.TRADE_QUEUE})
public void processBootTask(String content) {
System.out.println(content);
}
}

2.Spring Boot整合RabbitMQ (fanout交换机)

要使用 fanout 交换机的话，那么生产者与消费者的配置就不一样了。

1.生产者生产消息

在 Java 配置类中注册 bean：

@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_FANOUT_EXCHANGE = "admin.fanout";
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(DEFAULT_FANOUT_EXCHANGE);
}
}

在 SpringBoot 中，我们使用 AmqpTemplate 去发送消息：

@Component
public class HelloProducer {
@Autowired
private AmqpTemplate amqpTemplate;
public void send() {
//参数一:交换机名称，参数二:发送的key，参数三:内容
amqpTemplate.convertAndSend(RabbitConfiguration.DEFAULT_FANOUT_EXCHANGE, "", "Hello, Rabbit!");
}
}

2.消费者消费消息

在 Java 配置类中注册 bean：

@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_FANOUT_EXCHANGE = "admin.fanout";
public static final String FANOUT_QUEUE = "admin-" + UUID.randomUUID();
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
//配置广播路由器
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(DEFAULT_FANOUT_EXCHANGE);
}
//配置临时队列
@Bean
public Queue randomQueue() {
return new Queue(FANOUT_QUEUE);
}
@Bean
public Binding bindingExchange() {
return BindingBuilder.bind(randomQueue()).to(fanoutExchange());
}
}

配置监听器，监听 Queue，当消息队列有消息时，监听器就会接收到消息（也可配置多个队列接收）：

@Component
public class HelloConsumer {
@RabbitListener(queues = "#{rabbitConfiguration.FANOUT_QUEUE}")
public void processBootTask(String content) {
System.out.println(content);
}
}

3.Spring Boot整合RabbitMQ (topic交换机)

使用 topic 交换机，生产者与消费者的配置也不一样。

1.生产者生产消息

在 Java 配置类中注册 bean：

@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_TOPIC_EXCHANGE = "admin.topic";
public static final String TOPIC_ROUTE_KEY = "A.B.C";
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(DEFAULT_TOPIC_EXCHANGE);
}
}

在 SpringBoot 中，我们使用 AmqpTemplate 去发送消息：

@Component
public class HelloProducer {
@Autowired
private AmqpTemplate amqpTemplate;
public void send() {
//参数一:交换机名称，参数二:发送的key，参数三:内容
amqpTemplate.convertAndSend(RabbitConfiguration.DEFAULT_TOPIC_EXCHANGE, RabbitConfiguration.TOPIC_ROUTE_KEY, "Hello, Rabbit!");
}
}

RabbitMQ 将会根据参数二去寻找有没有匹配此规则的队列，如果有则把消息给它，如果有且不止一个，则把消息分发给匹配的队列 (每个队列都有消息)。

2.消费者消费消息

在 Java 配置类中注册 bean：

@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_TOPIC_EXCHANGE = "admin.topic";
public static final String TOPIC_QUEUE = "admin-" + UUID.randomUUID();
//*表示一个词，#表示零个或多个词
public static final String TOPIC_ROUTE_KEY = "#.#";
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
//配置主题路由器
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(DEFAULT_TOPIC_EXCHANGE);
}
//配置临时队列
@Bean
public Queue randomQueue() {
return new Queue(TOPIC_QUEUE);
}
@Bean
public Binding bindingExchange() {
return BindingBuilder.bind(randomQueue()).to(topicExchange()).with(TOPIC_ROUTE_KEY);
}
}

配置监听器，监听 Queue，当消息队列匹配此规则时，监听器就会接收到消息：

@Component
public class HelloConsumer {
@RabbitListener(queues = "#{rabbitConfiguration.TOPIC_QUEUE}")
public void processBootTask(String content) {
System.out.println(content);
}
}

生产端指定 Route Key 为 A.B.C, 下面是消费端绑定 Route Key 的不同情况：

消费端绑定的Route Key	是否匹配
A.B.C	Yes
#	Yes
A.#	Yes
.	No
A.*	No
A.B.*	Yes
A.*.C	Yes