Spring Boot集成RabbitMQ以及队列模式操作

Spring Boot集成RabbitMQ以及队列模式操作

目录前言一、场景描述二、准备工作三、发布/订阅模式（Fanout） 生产者 消费者四、Work模式4.1 轮询模式 生产者 消费者xmhXOcKxt4.2 公平分发 生产者 消费者 生产者 消费者五、路由模式（Direct）六、主题模式（Topic） 小结

前言

本篇博客将会通过我们的实际场景来演示如何在Spring Boot中集成RabbitMQ以及如何对各种队列模式进行操作。

一、场景描述

我们通过模仿用户下订单时，订单系统分别通过短信，邮件或微信进行推送消息，如下图：

二、准备工作

（1）创建两个Spring Boot项目分别对应生产者和消费者。

（2）导入依赖。

org.springframework.boot

spring-boot-starter-amqp

org.springframework.boot

spring-boot-starter-web

（3）定义生产者的配置文件。

application.yml:

server:

port: 8021

spring:

#给项目来个名字

application:

name: rabbitmq-provider

#配置rabbitMq 服务器

rabbitmq:

host: 服务器地址

port: 5672

username: yixin

password: 123456

#虚拟host 可以不设置,使用server默认host

virtual-host: /

（4）定义消费者的配置文件。

application.yml:

server:

port: 8022

spring:

#给项目来个名字

application:

name: rabbitmq-consumer

#配置rabbitMq 服务器

rabbitmq:

host: 服务器地址

port: 5672

username: yixin

password: 123456

#虚拟host 可以不设置,使用server默认host

virtual-host: /

三、发布/订阅模式（Fanout）

生产者

项目名：rabbitmq-fanout-provider

（1）编写配置类。

package com.yixin.config;

import org.springframework.amqp.core.*;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class FanoutRabbitConfig {

/**

* 创建三个队列 ：fanout.ShortMessage fanout.Email fanout.WeChat

* 将三个队列都绑定在交换机 fanoutExchange 上

* 因为是扇型交换机, 路由键无需配置,配置也不起作用

*/

@Bean

public Queue queueShortMessage() {

// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效

// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable

// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。

// return new Queue("TestDirectQueue",true,true,false);

//一般设置一下队列的持久化就好,其余两个就是默认false

return new Queue("fanout.ShortMessage",true);

}

@Bean

public Queue queueEmail() {

return new Queue("fanout.Email",true);

}

@Bean

public Queue queueWeChat() {

return new Queue("fanout.WeChat",true);

}

//Fanout交换机 起名：fanoutExchange

@Bean

FanoutExchange fanoutExchange() {

return new FanoutExchange("fanoutExchange");

}

//绑定 将队列和交换机绑定

@Bean

Binding bindingExchangeShortMessage() {

return BindingBuilder.bind(queueShortMessage()).to(fanoutExchange());

}

@Bean

Binding bindingExchangeEmail() {

return BindingBuilder.bind(queueEmail()).to(fanoutExchange());

}

@Bean

Binding bindingExchangeWeChat() {

return BindingBuilder.bind(queueWeChat()).to(fanoutExchange());

}

}

（2）编写一个Controller类进行推送消息。

package com.yixin.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.HashMap;

import java.util.Map;

import java.util.UUID;

@RestController

public class OrderController {

@Autowired

RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法

@GetMapping("/sendFanoutMessage")

public String sendFanoutMessage() {

String messageId = String.valueOf(UUID.randomUUID());

String messageData = "用户成功下单了！";

String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

Map map = new HashMap<>();

map.put("messageId", messageId);

map.put("messageData", messageData);

map.put("createTime", createTime);

rabbitTemplate.convertAndSend("fanoutExchange", null, map);

return "ok";

}

}

消费者

项目名：rabbitmq-fanout-consumer

邮件系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "fanout.Email")

public class Email {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("邮件系统收到消息 : " +testMessage.toString());

}

}

短信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "fanout.ShortMessage")

public class ShortMessage {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("短信系统收到消息 : " +testMessage.toString());

}

}

微信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "fanout.WeChat")

public class WeChat {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("微信系统收到消息 : " +testMessage.toString());

}

}

现在我们已经创建好了，将rabbitmq-fanout-provider和rabbitmq-fanout-consumer项目都跑起来，在浏览器输入：http://localhost:8021/sendFanoutMessage 进行发送消息。

消费者的控制台情况如下：

四、Work模式

4.1 轮询模式

解释：所谓轮询分发就是有两个消费者监听同一个队列，那么当我们发大量消息的时候，交换器会将消息平均分配给两个消费者，就算其中一个消费者的处理效率比另一个高，也同样只能分配一样的消息数量。

生产者

（1）编写配置类。

package com.yixin.config;

import org.springframework.amqp.core.*;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class WorkRabbitMQConfig {

@Bean

public Queue queueWork() {

return new Queue("queue_work",true);

}

@Bean

FanoutExchange workExchange() {

return new FanoutExchange("WorkExchange");

}

//绑定 将队列和交换机绑定

@Bean

Binding bindingExchangeShortMessage() {

return BindingBuilder.bind(queueWork()).to(workExchange());

}

}

（2）编写Controller类进行推送消息。

package com.yixin.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.HashMap;

import java.util.Map;

import java.util.UUID;

@RestController

public class SendMessageController {

@Autowired

RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法

@GetMapping("/sendWorkMessage")

public String sendDirectMessage() {

for(int i=0;i<10;i++) {

String message="收到消息："+i;

//将消息携带绑定键值：shortmessage 发送到交换机DirectExchange

rabbitTemplate.convertAndSend("WorkExchange", "", message);

System.out.println("发送成功:"+i);

}

return "消息发送成功！";

}

}

消费者

邮件系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "queue_work")

public class Email {

@RabbitHandler

public void process(String testMessage) {

try {

Thread.sleep(100);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("邮件系统收到消息 : " +testMessage);

}

}

短信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "queue_work")

public class ShortMessage {

@RabbitHandler

public void process(String testMessage) {

//休眠300毫秒，表示效率相比Email低

try {

Thread.sleep(300);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("短信系统收到消息 : " +testMessage);

}

}

我们把生产者和消费者项目都启动起来，访问 http://localhost:8021/sendWorkMessage 进行推送消息：

查看消费者的消费信息：

可以发现两者的工作效率不一样，但分配到的数量确实一样的，邮件收到的是偶数，短信收到的是奇数，这就是轮询分发！

4.2 公平分发

解读：公平分发就是根据谁执行的效率高，那么就给其多分发消息进行处理，正所谓能者多劳。

实现公平分发很简单，在基于轮询分发的基础上，我们只需要在消费者项目的配置文件中加入以下代码：

spring:

listener:

simple:

prefetch: 1

表示将预处理模式更改为每次读取1条消息,在消费者未回执确认之前,不在进行下一条消息的投送。

故我们的消费者的配置文件整体如下：

application.yml:

server:

port: 8022

spring:

application:

name: rabbitmq-consumer

rabbitmq:

host: 服务器地址

port: 5672

username: yixin

password: 123456

virtual-host: /

listener:

simple:

prefetch: 1

我们重新启动生产者和消费者项目，消费情况如下：

现在可以发现，10条消息，邮件系统消费了7条，因为邮件系统比短信系统的执行效率更高！

五、路由模式（Direct）

生产者

项目名：rabbitmq-direct-provider

（1）编写配置类。

package com.yixin.config;

import org.springframework.amqp.core.*;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class DirectRabbitConfig {

/**

* 创建三个队列 ：direct.ShortMessage direct.Email direct.WeChat

* 将三个队列都绑定在交换机 DirectExchange 上

* 因为是扇型交换机, 路由键无需配置,配置也不起作用

*/

@Bean

public Queue queueShortMessage() {

return new Queue("direct.ShortMessage",true);

}

@Bean

public Queue queueEmail() {

return new Queue("direct.Email",true);

}

@Bean

public Queue queueWeChat() {

return new Queue("direct.WeChat",true);

}

//Direct交换机 起名：DirectExchange

@Bean

DirectExchange DirectExchange() {

return new DirectExchange("DirectExchange",true,false);

}

//绑定 将队列和交换机绑定,并设置用于匹配键

@Bean

Binding bindingExchangeShortMessage() {

return BindingBuilder.bind(queueShortMessage()).to(DirectExchange()).with("shortmessage");

}

@Bean

Binding bindingExchangeEmail() {

return BindingBuilder.bind(queueEmail()).to(DirectExchange()).with("email");

}

@Bean

Binding bindingExchangeWeChat() {

return BindingBuilder.bind(queueWeChat()).to(DirectExchange()).with("wechat");

}

}

（2）编写Controller类进行推送消息。

package com.yixin.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.HashMap;

import java.util.Map;

import java.util.UUID;

@RestController

public class SendMessageController {

@Autowired

RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法

@GetMapping("/sendDirectMessage")

public String sendDirectMessage() {

String messageId = String.valueOf(UUID.randomUUID());

String messageData = "Direct:用户成功下单了！";

String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

Map map=new HashMap<>();

map.put("messageId",messageId);

map.put("messageData",messageData);

map.put("createTime",createTime);

//将消息携带绑定键值：shortmessage 发送到交换机DirectExchange

rabbitTemplate.convertAndSend("DirectExchange", "shortmessage", map);

return "消息发送成功！";

}

}

消费者

项目名：rabbitmq-direct-consumer

邮件系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "direct.Email")

public class Email {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("邮件系统收到消息 : " +testMessage.toString());

}

}

短信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "direct.ShortMessage")

public class ShortMessage {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("短信系统收到消息 : " +testMessage.toString());

}

}

微信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "direct.WeChat")

public class WeChat {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("微信系统收到消息 : " +testMessage.toString());

}

}

我们将生产者和消费者项目都跑起来，在浏览器输入：http://localhost:8021/sendDirectMessage 进行发送消息。

我们的消费者消费情况如下：

发现确实是只有我们的短信系统收到消息了，测试成功！

六、主题模式（Topic）

生产者

项目名：rabbitmq-topic-provider

（1）编写配置类。

package com.yixin.config;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.core.TopicExchange;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class TopicRabbitConfig {

@Bean

public Queue queueShortMessage() {

return new Queue("topic.ShortMessage");

}

@Bean

public Queue queueEmail() {

return new Queue("topic.Email");

}

@Bean

public Queue queueWeChat() {

return new Queue("topic.WeChat");

}

@Bean

TopicExchange exchange() {

return new TopicExchange("TopicExchange");

}

//将queueShortMessage和TopicExchange绑定,而且绑定的键值为topic.shortmessage

//这样只要是消息携带的路由键是topic.shortmessage,才会分发到该队列

@Bean

Binding bindingExchangeShortMessage() {

return BindingBuilder.bind(queueShortMessage()).to(exchange()).with("topic.shortmessage");

}

//将queueEmail和TopicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#

// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列

@Bean

Binding bindingExchangeEmail() {

return BindingBuilder.bind(queueEmail()).to(exchange()).with("topic.#");

}

//只要是消息携带的路由键是topic.wechat,才会分发到该队列

@Bean

Binding bindingExchangeWeChat() {

return BindingBuilder.bind(queueEmail()).to(exchange()).with("topic.wechat");

}

}

（2）编写Controller类进行推送消息。

package com.yixin.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.HashMap;

import java.util.Map;

import java.util.UUID;

@RestController

public class SendMessageController {

@Autowired

RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法

@GetMapping("/sendTopicMessage")

public String sendDirectMessage() {

String messageId = String.valueOf(UUID.randomUUID());

String messageData = "Topic:用户成功下单了！";

String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

Map map=new HashMap<>();

map.put("messageId",messageId);

map.put("messageData",messageData);

map.put("createTime",createTime);

//将消息携带绑定键值：topic.shortmessage 发送到交换机TopicExchange

rabbitTemplate.convertAndSend("TopicExchange", "topic.shortmessage", map);

return "消息发送成功！";

}

}

消费者

项目名：rabbitmq-topic-consumer

邮件系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "topic.Email")

public class Email {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("邮件系统收到消息 : " +testMessage.toString());

}

}

短信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "topic.ShortMessage")

public class ShortMessage {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("短信系统收到消息 : " +testMessage.toString());

}

}

微信系统：

package com.yixin.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

import java.util.Map;

@Component

@RabbitListener(queues = "topic.WeChat")

public class WeChat {

@RabbitHandler

public void process(Map testMessage) {

System.out.println("微信系统收到消息 : " +testMessage.toString());

}

}

现在把我们的项目生产者和消费者项目启动起来，并输入http://localhost:8021/sendTopicMessage 进行推送消息。

对我们的消费者接受消息的情况进行查看：

可以发现只有我们的邮件和短信系统收到了通知，测试成功！

小结

以上就是【一心同学】整理的如何在【Spring Boot】中集成【RabbitMQ】并且通过【场景演示】在Spring Boot中对各种【队列模式】的使用，大家可以自己动手演示一遍，记忆更加深刻。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。