Spring和RabbitMQ消息队列(AMQP)整合详解

╰半夏微凉° 2021-12-23 02:55 929阅读 0赞

# Spring和RabbitMQ消息队列(AMQP)整合详解 #

## 官方主页 ##

[Spring AMQP][]

## 一、概述 ##

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。对于消息中间件，常见的角色大致也就有Producer（生产者）、Consumer（消费者）

常见的消息中间件产品：

（1）ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。这里介绍的是ActiveMQ的使用。

（2）RabbitMQ

AMQP协议的领导实现，支持多种场景。淘宝的MySQL集群内部有使用它进行通讯，OpenStack开源云平台的通信组件，最先在金融行业得到运用。

（3）ZeroMQ

史上最快的消息队列系统

（4）Kafka

Apache下的一个子项目 。特点：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统。适合处理海量数据。

**Jms**

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

**AMQP**

AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。意味着我们可以使用Java的AMQP provider，同时使用一个python的producer加一个rubby的consumer。从这一点看，AQMP可以用http来进行类比，不关心实现的语言，只要大家都按照相应的数据格式去发送报文请求，不同语言的client均可以和不同语言的server链接。

上一篇[《Spring和ActiveMq消息队列整合详解》][Spring_ActiveMq]介绍了ActiveMq的整合

本篇通过介绍下RabbitMQ的整合过程。

**建议访问[首发地址][Link 1]查看，自动生成目录树方便查看章节。**

代码可以在Spring组件化构建[https://www.pomit.cn/java/spring/spring.html][https_www.pomit.cn_java_spring_spring.html]中的RabbitMQ组件中查看，并下载。

首发地址：  
[品茗IT-同步发布][Link 1]

[品茗IT][IT] 提供在线支持：

[一键快速构建Spring项目工具][https_www.pomit.cn_java_spring_spring.html]

[一键快速构建SpringBoot项目工具][SpringBoot]

[一键快速构建SpringCloud项目工具][SpringCloud]

[一站式Springboot项目生成][Springboot]

[Mysql一键生成Mybatis注解Mapper][Mysql_Mybatis_Mapper]

**如果大家正在寻找一个java的学习环境，或者在开发中遇到困难，可以[加入我们的java学习圈，点击即可加入][java]，共同学习，节约学习时间，减少很多在学习中遇到的难题。**

## 二、环境配置 ##

### 2.1 RabbitMQ的安装部署 ###

RabbitMQ可以在[RabbitMQ官网][RabbitMQ] 下载并安装。

这里只说windows的安装过程：

*  1.下载rabbitmq-server-3.7.15.exe，下载完成后，双击安装，会提示先安装erlang，erlang的官方地址是：[https://www.erlang.org/downloads][https_www.erlang.org_downloads]
 *  2.安装完erlang，然后安装RabbitMQ，安装过程没有任何需要注意的，就是一路往下。
 *  3.RabbitMQ安装完成后。找到RabbitMQ的路径：

比如我的windows10会出现这个：

![在这里插入图片描述][aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjA4MjM5Nzc2MDAwLnBuZw]

然后我点击RabbitMQ Service - start:

![在这里插入图片描述][aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjA4NDUxNjI1MjE2LnBuZw]

为了启动管理服务，查看RabbitMQ的界面，我们需要打开RabbitMQ Command Prompt命令行，然后切换到RabbitMQ安装目录下的sbin目录，输入命令`rabbitmq-plugins.bat enable rabbitmq_management`:

![在这里插入图片描述][aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjExMzE5OTE0NzUyLnBuZw]

这里不建议自己启动命令提示符，或者使用powershell，因为会出现`ERLANG_HOME not set correctly.`错误。

浏览器输入`http://127.0.0.1:15672/`进行访问：

![在这里插入图片描述][aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjEyMDQwMzA2OTQ0LnBuZw]

使用默认账号登录：guest/guest

![在这里插入图片描述][aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjEyMTI1OTcwNjg4LnBuZw]

到这里，安装工作搞一段落了。

### 2.2 Web项目建立 ###

本文假设你已经引入Spring必备的一切了，已经是个Spring项目了,如果不会搭建，可以打开这篇文章看一看[《Spring和Spring Mvc 5整合详解》][Spring_Spring Mvc 5]。

本项目将RabbitMQ的exchange三种模式的生产者和消费者都放在一个项目中，通过调用web接口发送消息，并监听每个队列的消息。

#### 2.2.1 maven依赖 ####

<?xml version="1.0"?>
    <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>cn.pomit</groupId>
    		<artifactId>SpringWork</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>RabbitMQ</artifactId>
    	<packaging>jar</packaging>
    	<name>RabbitMQ</name>
    	<url>http://maven.apache.org</url>
    	<properties>
    		<spring-rabbit.version>2.1.7.RELEASE</spring-rabbit.version>
    		<jsonlib.version>2.4</jsonlib.version>
    		<spring.security.version>5.1.0.RELEASE</spring.security.version>
    	</properties>
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-web</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-webmvc</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.amqp</groupId>
    			<artifactId>spring-rabbit</artifactId>
    			<version>${spring-rabbit.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    		</dependency>
    	</dependencies>
    </project>

父模块可以在[https://www.pomit.cn/spring/SpringWork/pom.xml][https_www.pomit.cn_spring_SpringWork_pom.xml]获取。

## 三、Direct模式 ##

### 3.1 配置 ###

spring-rabbitmq.xml:

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/rabbit https://www.springframework.org/schema/rabbit/spring-rabbit.xsd http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
    
    	<bean id="annotationPropertyConfigurerAmqp" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
    		<property name="order" value="1" />
    		<property name="ignoreUnresolvablePlaceholders" value="true" />
    		<property name="locations">
    			<list>
    				<value>classpath:rabbit.properties</value>
    			</list>
    		</property>
    	</bean>
    
    	<bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
    
    	<rabbit:connection-factory id="connectionFactory" host="${rabbit.host}" port="${rabbit.port}" username="${rabbit.user}" password="${rabbit.password}" />
    
    	<rabbit:admin connection-factory="connectionFactory" />
    
    	
    	<rabbit:queue durable="true" auto-delete="false" exclusive="false" name="${rabbit.direct.queue}" />
    	
    	<rabbit:direct-exchange name="${rabbit.direct.exchange}" durable="true" auto-delete="false">
    		<rabbit:bindings>
    			<rabbit:binding queue="${rabbit.direct.queue}" key="${rabbit.direct.key}" />
    		</rabbit:bindings>
    	</rabbit:direct-exchange>
    	
    	<rabbit:template id="amqpDirectTemplate" exchange="${rabbit.direct.exchange}" routing-key="${rabbit.direct.key}" connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
    
    	<rabbit:annotation-driven />
    
    	<bean id="rabbitListenerContainerFactory" class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="concurrentConsumers" value="3" />
    		<property name="maxConcurrentConsumers" value="10" />
    		<property name="messageConverter" ref="jsonMessageConverter" />
    
    	</bean>
    </beans>

这里面的配置可以分为连接、管理、队列、exchange、发送、容器、数据转换这几类：

*  连接：connectionFactory定义了连接工厂。
 *  管理：rabbit:admin 是rabbit对所有队列、exchange的管理，这个过程是由Spring自动完成的。
 *  队列：rabbit:queue定义了一个队列，队列只是负责接收消息。
 *  exchange：交换器，rabbit:direct-exchange定义了一个点对点的传输过程。这个exchange要和队列进行绑定，routing-key是路由标识，如果使用rabbit:fanout-exchange，则不能有routing-key。rabbit:topic-exchange模式下可以有routing-key，而且routing-key可以使用模糊匹配。
 *  发送：rabbit:template负责发送消息，和exchange和routing-key（fanout-exchange不必须）绑定。
 *  容器：jmsContainer是将消息队列和监听bean整合起来，这样就保证用的时候能找到对应的bean。
 *  数据转换：jsonMessageConverter的bean使用Jackson2JsonMessageConverter，可以将发送和接收消息自动转为实体。

**<rabbit:annotation-driven /> 以开启注解，我们把监听者用注解来启动，用配置文件不够灵活。**

rabbit.properties配置文件：

rabbit.host=127.0.0.1
    rabbit.port=5672
    rabbit.user=guest
    rabbit.password=guest
    
    rabbit.direct.queue=spring.queue.direct
    rabbit.direct.exchange=spring.exchange.direct
    rabbit.direct.key=spring.key.direct

### 3.2 发送方（生产者） ###

我们使用web服务进行调用发送消息。

RabbitWeb :

package cn.pomit.springwork.rabbitmq.web;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    
    import cn.pomit.springwork.rabbitmq.dto.ParamReq;
    import cn.pomit.springwork.rabbitmq.service.PomitDirectMessageService;
    
    @RestController
    @RequestMapping("/rabbit")
    public class RabbitWeb { 
    	@Autowired
    	PomitDirectMessageService pomitDirectMessageService;
    
    	@RequestMapping(value = "/direct", method = {  RequestMethod.GET })
    	public String direct() { 
    		ParamReq paramReq = new ParamReq();
    		paramReq.setContent("asdasdxbcvbcb");
    		paramReq.setMessageType("direct");
    		pomitDirectMessageService.sendMessage(paramReq);
    		return "0000";
    	}
    }

PomitDirectMessageService :

package cn.pomit.springwork.rabbitmq.service;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Service;
    
    import cn.pomit.springwork.rabbitmq.dto.ParamReq;
    
    @Service
    public class PomitDirectMessageService { 
    	@Autowired
    	@Qualifier("amqpDirectTemplate")
    	AmqpTemplate amqpTemplate;
    	
    	public void sendMessage(ParamReq paramReq){ 
    		amqpTemplate.convertAndSend(paramReq);
    	}
    }

这里面的amqpTemplate使用配置文件中的amqpDirectTemplate。点对点进行发送。

### 3.3 监听者（消费者） ###

为了灵活方便监听数据，我们使用注解来完成这一处理，而且配置文件中配置了messageConverter之后，Spring能够自动将数据转为实体。

PomitMessageListener：

package cn.pomit.springwork.rabbitmq.listener;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import cn.pomit.springwork.rabbitmq.dto.ParamReq;
    
    @Component
    public class PomitMessageListener{ 
    
    	@RabbitListener(queues = "${rabbit.direct.queue}")
    	public void handleMessage(ParamReq paramReq) { 
    		
    		System.out.println(paramReq);
    	}
    
    }

## 四、Fanout模式 ##

在上面的Direct模式中，稍微加点东西，就可以变为fanout模式。

### 4.1 配置 ###

在spring-rabbit.xml中加入下面配置：

<rabbit:queue durable="true" auto-delete="false" exclusive="false" name="${rabbit.common.queue}" />
    	<rabbit:queue durable="true" auto-delete="false" exclusive="false" name="${rabbit.common.queue2}" />
    
    
    	
    	<rabbit:fanout-exchange name="${rabbit.fanout.exchange}" durable="true" auto-delete="false">
    		<rabbit:bindings>
    			<rabbit:binding queue="${rabbit.common.queue}" />
    			<rabbit:binding queue="${rabbit.common.queue2}" />
    		</rabbit:bindings>
    	</rabbit:fanout-exchange>
    	
    	<rabbit:template id="amqpFanoutTemplate" exchange="${rabbit.fanout.exchange}" connection-factory="connectionFactory" message-converter="jsonMessageConverter" />

在rabbit.pproperities中配置文件中加入：

rabbit.common.queue=spring.queue.common
    rabbit.common.queue.pattern=spring.key.*
    rabbit.fanout.exchange=spring.exchange.fanout

### 4.2 发送方（生产者） ###

我们使用web服务进行调用发送消息。RabbitWeb中加入发送fanout模式消息的接口：

@Autowired
    PomitTopicMessageService pomitTopicMessageService;

@RequestMapping(value = "/fanout", method = {  RequestMethod.GET })
    public String fanout() { 
        ParamReq paramReq = new ParamReq();
        paramReq.setContent("asdasdxbcvbcb");
        paramReq.setMessageType("fanout");
        pomitFanoutMessageService.sendMessage(paramReq);
        return "0000";
    }

PomitFanoutMessageService:

package cn.pomit.springwork.rabbitmq.service;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Service;
    
    import cn.pomit.springwork.rabbitmq.dto.ParamReq;
    
    @Service
    public class PomitFanoutMessageService { 
    	@Autowired
    	@Qualifier("amqpFanoutTemplate")
    	AmqpTemplate amqpTemplate;
    	
    	public void sendMessage(ParamReq paramReq){ 
    		amqpTemplate.convertAndSend(paramReq);
    	}
    }

这里面的amqpTemplate使用配置文件中的amqpFanoutTemplate。广播进行发送。

### 4.3 监听者（消费者） ###

为了灵活方便监听数据，我们使用注解来完成这一处理，而且配置文件中配置了messageConverter之后，Spring能够自动将数据转为实体。

在PomitMessageListener中加入监听即可：

@RabbitListener(queues = "${rabbit.common.queue}")
    	public void handleCommonMessage(ParamReq paramReq) { 
    		
    		System.out.println("一、" + paramReq);
    	}
    
    	@RabbitListener(queues = "${rabbit.common.queue2}")
    	public void handleCommonMessage2(ParamReq paramReq) { 
    		
    		System.out.println("二、" + paramReq);
    	}

## 五、Topic模式 ##

在上面的Fanout模式中，稍微加点东西，就可以变为Topic模式。

### 5.1 配置 ###

在spring-rabbit.xml中加入下面topic的配置：

![在这里插入图片描述][aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjc2Nzk0MDExOTA0LnBuZw]

如图所示，可以个fanout模式和direct模式公用同一个队列，但是这里的topic绑定队列要求填写pattern。

pattern是routing-key的模糊匹配写法，你可以直接写成routing-key，这样发送的时候如果绑定了routing-key，就能直接接收到，也可以写成xxx.*,*.xxx，这样只要能模糊匹配到发送的routing-key，就能接收到，否则绑定的队列是接收不到的。

在rabbit.pproperities中配置文件中加入：

rabbit.common.queue2=spring.queue.common2
    rabbit.common.queue2.pattern=*.key.topic
    rabbit.topic.exchange=spring.exchange.topic
    rabbit.topic.key=spring.key.topic

### 5.2 发送方（生产者） ###

我们使用web服务进行调用发送消息。RabbitWeb中加入发送fanout模式消息的接口：

@Autowired
    PomitTopicMessageService pomitTopicMessageService;

@RequestMapping(value = "/topic1", method = {  RequestMethod.GET })
    public String topic1() { 
    	ParamReq paramReq = new ParamReq();
    	paramReq.setContent("asdasdxbcvbcb");
    	paramReq.setMessageType("topic1");
    	pomitTopicMessageService.sendMessage(paramReq);
    	return "0000";
    }
    
    @RequestMapping(value = "/topic2", method = {  RequestMethod.GET })
    public String topic2() { 
    	ParamReq paramReq = new ParamReq();
    	paramReq.setContent("asdasdxbcvbcb");
    	paramReq.setMessageType("topic2");
    	pomitTopicMessageService.sendMessage2(paramReq);
    	return "0000";
    }

PomitTopicMessageService:

404 not found

这里面的amqpTemplate使用配置文件中的两个amqpTemplateTopic、amqpTemplateTopic2。分别发送不同的routing-key进行测试。

### 5.3 监听者（消费者） ###

为了灵活方便监听数据，我们使用注解来完成这一处理，而且配置文件中配置了messageConverter之后，Spring能够自动将数据转为实体。

PomitMessageListener的监听可以不变，因为队列并没有新增，还是那三个。

## 六、过程中用到的实体 ##

详细完整的实体，可以访问[品茗IT-博客《Spring和RabbitMQ消息队列(AMQP)整合详解》][Link 1]进行查看

全部代码可以在Spring组件化构建[https://www.pomit.cn/java/spring/spring.html][https_www.pomit.cn_java_spring_spring.html]中的RabbitMQ组件中查看，并下载。

### 快速构建项目 ###

[Spring组件化构建][https_www.pomit.cn_java_spring_spring.html]

[SpringBoot组件化构建][SpringBoot]

[SpringCloud服务化构建][SpringCloud]

喜欢这篇文章么，喜欢就加入我们一起讨论SpringBoot技术吧！  
![品茗IT交流群][IT 1]

[Spring AMQP]: https://docs.spring.io/spring-amqp/docs/2.1.7.RELEASE/reference/html/
[Spring_ActiveMq]: https://www.pomit.cn/p/178648639553792
[Link 1]: https://www.pomit.cn/p/191678573195520
[https_www.pomit.cn_java_spring_spring.html]: https://www.pomit.cn/java/spring/spring.html
[IT]: https://www.pomit.cn
[SpringBoot]: https://www.pomit.cn/java/spring/springboot.html
[SpringCloud]: https://www.pomit.cn/java/spring/springcloud.html
[Springboot]: https://www.pomit.cn/java/java/sqlToSpringboot.html
[Mysql_Mybatis_Mapper]: https://www.pomit.cn/java/java/sqlToMybatis.html
[java]: https://jq.qq.com/?_wv=1027&k=52sgH1J
[RabbitMQ]: https://www.rabbitmq.com/
[https_www.erlang.org_downloads]: https://www.erlang.org/downloads
[aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjA4MjM5Nzc2MDAwLnBuZw]: /images/20211223/4bcafb2c5e7b435cab3117593d756422.png
[aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjA4NDUxNjI1MjE2LnBuZw]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjA4NDUxNjI1MjE2LnBuZw
[aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjExMzE5OTE0NzUyLnBuZw]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjExMzE5OTE0NzUyLnBuZw
[aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjEyMDQwMzA2OTQ0LnBuZw]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjEyMDQwMzA2OTQ0LnBuZw
[aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjEyMTI1OTcwNjg4LnBuZw]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjEyMTI1OTcwNjg4LnBuZw
[Spring_Spring Mvc 5]: https://www.pomit.cn/p/181499739820288
[https_www.pomit.cn_spring_SpringWork_pom.xml]: https://www.pomit.cn/spring/SpringWork/pom.xml
[aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjc2Nzk0MDExOTA0LnBuZw]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93d3cucG9taXQuY24vdXBsb2FkL3BpY3R1cmUvMjAxOTA2MjYvMTkxNjc2Nzk0MDExOTA0LnBuZw
[IT 1]: https://img-blog.csdnimg.cn/20190430181302249.png