SpringBoot入门建站全系列（十八）整合RabbitMQ（AMQP类消息队列）

ゞ浴缸里的玫瑰 2021-11-29 14:16 287阅读 0赞

## SpringBoot入门建站全系列（十八）整合RabbitMQ（AMQP类消息队列） ##

#### 一、概述 ####

#### 1.1 简介 ####

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。对于消息中间件，常见的角色大致也就有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链接。

本篇将通过介绍RabbitMQ的三种模式的配合使用来介绍下Springboot如何整合RabbitMQ。

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

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

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

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

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

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

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

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

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

#### 1.2 RabbitMQ基础概念 ####

说基础概念，这里就不多废话了，想必大家都已经百度过了，我这里只说下关于本篇的用到的一些概念。

**几个概念：queue、exchange、Binding、routingKey**:

queue:队列，和用什么模式无关。

exchange：交换机，就是个代理，决定你用哪种模式，每个模式需要定义一个exchange。

Binding：队列绑定，需要将queue与exchange绑定，可以交叉绑定，一个Binding，就代表一个路径，比如queue绑定到direct模式的exchange，这个queue就可以接收direct模式的消息；queue绑定到topic模式的exchange，这个queue就可以接收topic模式的消息。

routingKey：路由key，等于一条消息的归类。通过这个匹配到与exchange绑定的queue。因为绑定的时候是指定了routingKey的类型。

\*\*三种模式：direct、fanout、topic。\*\*不是说只有这三种，是这里只说这三种常见的。

direct模式： 一对一模式，发送路径是固定的，一个routingKey只会转发到绑定到指定exchange并绑定到exchange的queue。

fanout模式： 一对多，没有routingKey，消息发送的时候发送到exchange，因为没有routingKey，exchange会转发到绑定到该exchange的所有queue。

topic模式：一对多，可以绑定多个key，也可以绑定模式匹配的key。消息发送的时候发送到exchange，exchange会根据绑定的所有key和模式转发到相应的queue。所以这个模式叫做订阅。

### 二、配置 ###

本文假设你已经引入spring-boot-starter-web。已经是个SpringBoot项目了,如果不会搭建，可以打开这篇文章看一看[《SpringBoot入门建站全系列（一）项目建立》][SpringBoot 1]。

#### 2.1 Maven依赖 ####

使用RabbitMQ可以使用spring-boot-starter-amqp，方便快捷，一般springboot对大多数开源项目都做了整合，提供了专用的stater。

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

#### 2.2 配置文件 ####

在application.properties 中需要配置RabbitMQ的信息，也可以配置自定义的配置，如：

spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    
    rabbit.direct.queue=spring.queue.direct
    rabbit.direct.exchange=spring.exchange.direct
    rabbit.direct.key=spring.key.direct
    
    rabbit.common.queue=spring.queue.common
    rabbit.common.queue.pattern=spring.key.*
    
    rabbit.common.queue2=spring.queue.common2
    rabbit.common.queue2.pattern=*.key.topic
    
    rabbit.fanout.exchange=spring.exchange.fanout
    
    rabbit.topic.exchange=spring.exchange.topic
    
    rabbit.topic.key=spring.key.topic

这里面，

*  spring.rabbitmq.\*是springboot自动装配的配置，rabbitmq的相关信息。
 *  rabbit.direct.\*是我自定义的配置，是direct模式下需要的配置。
 *  rabbit.fanout.\*是我自定义的配置，是fanout模式下需要的配置。
 *  rabbit.topic.\*是我自定义的配置，是topic模式下需要的配置。
 *  rabbit.common.\*是我自定义的配置，是topic和fanout模式下都可以使用的配置。

所以，这里定义了3个queue，2个key（fanout模式下不能使用key），2个key模式匹配（用于匹配topic模式下的key），三个exchange。概念可以会看1.1小节。

### 三、RabbitMQ的初始化配置 ###

RabbitMQ需要将所有的queue、exchange、binding声明为bean。同时可以定义一个MessageConverter，用于消息的转换。

RabbitConfiguration :

package com.cff.springbootwork.rabbitmq;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.amqp.rabbit.annotation.EnableRabbit;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @EnableRabbit
    @Configuration
    public class RabbitConfiguration { 
    	// 队列 分类配置
    	@Value("${rabbit.direct.queue}")
    	private String directQueueName;
    
    	@Value("${rabbit.common.queue}")
    	private String commonQueueName;
    
    	@Value("${rabbit.common.queue2}")
    	private String commonQueueName2;
    
    	// key 分类配置
    	@Value("${rabbit.direct.key}")
    	private String rabbitDirectKey;
    
    	@Value("${rabbit.common.queue.pattern}")
    	private String rabbitCommonKeyPattern;
    
    	@Value("${rabbit.common.queue2.pattern}")
    	private String rabbitCommonKeyPattern2;
    
    	// exchange 分类配置
    	@Value("${rabbit.direct.exchange}")
    	private String rabbitDirectExchange;
    
    	@Value("${rabbit.fanout.exchange}")
    	private String rabbitFanoutExchange;
    
    	@Value("${rabbit.topic.exchange}")
    	private String rabbitTopicExchange;
    
    	// 队列声明
    	@Bean(name = "directQueue")
    	public Queue directQueue() { 
    		return new Queue(directQueueName);
    	}
    
    	@Bean(name = "commonQueueName")
    	public Queue commonQueueName() { 
    		return new Queue(commonQueueName);
    	}
    
    	@Bean(name = "commonQueueName2")
    	public Queue commonQueueName2() { 
    		return new Queue(commonQueueName2);
    	}
    
    	// exchange声明
    	@Bean(name = "directExchange")
    	public DirectExchange directExchange() { 
    		return new DirectExchange(rabbitDirectExchange);
    	}
    
    	@Bean(name = "fanoutExchange")
    	public FanoutExchange fanoutExchange() { 
    		return new FanoutExchange(rabbitFanoutExchange);
    	}
    
    	@Bean(name = "topicExchange")
    	public TopicExchange topicExchange() { 
    		return new TopicExchange(rabbitTopicExchange);
    	}
    
    	// 数据绑定声明
    	@Bean(name = "directBinding")
    	public Binding directBinding() { 
    		return BindingBuilder.bind(directQueue()).to(directExchange()).with(rabbitDirectKey);
    	}
    
    	@Bean(name = "fanoutBinding")
    	public Binding fanoutBinding() { 
    		return BindingBuilder.bind(commonQueueName()).to(fanoutExchange());
    	}
    
    	@Bean(name = "fanoutBinding2")
    	public Binding fanoutBinding2() { 
    		return BindingBuilder.bind(commonQueueName2()).to(fanoutExchange());
    	}
    
    	@Bean(name = "topicBinding1")
    	public Binding topicBinding1() { 
    		return BindingBuilder.bind(commonQueueName()).to(topicExchange()).with(rabbitCommonKeyPattern);
    	}
    
    	@Bean(name = "topicBinding2")
    	public Binding topicBinding2() { 
    		return BindingBuilder.bind(commonQueueName2()).to(topicExchange()).with(rabbitCommonKeyPattern2);
    	}
    
    	@Bean(name = "topicBinding3")
    	public Binding topicBinding3() { 
    		return BindingBuilder.bind(directQueue()).to(topicExchange()).with(rabbitDirectKey);
    	}
    
    	// 数据转换声明
    	@Bean
    	public MessageConverter myMessageConverter() { 
    		return new Jackson2JsonMessageConverter();
    	}
    
    	@Bean
    	public SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
    			ConnectionFactory connectionFactory) { 
    		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    		configurer.configure(factory, connectionFactory);
    		factory.setMessageConverter(myMessageConverter());
    		return factory;
    	}
    }

这里面：

*  Queue是声明队列，上面声明了三个队列。
 *  DirectExchange是Direct模式下的Exchange；
 *  FanoutExchange是Fanout模式下的Exchange；
 *  TopicExchange是Topic模式下的Exchange；
 *  Binding，这里定义了6个绑定，一个绑定到DirectExchange；两个绑定到FanoutExchange；三个绑定到TopicExchange；
 *  声明了一个MessageConverter，用于将实体转换为json，或者将json转换为实体。过程是spring控制自动转换的。
 *  声明一个SimpleRabbitListenerContainerFactory，将MessageConverter传递给SimpleRabbitListenerContainerFactory。

### 四、配置发送方（生产者） ###

下面介绍下三种模式是怎么发送数据的。

#### 4.1 Direct模式 ####

PomitDirectMessageService ：

package com.cff.springbootwork.rabbitmq.service;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    
    import com.cff.springbootwork.rabbitmq.dto.ParamReq;
    
    @Service
    public class PomitDirectMessageService { 
    	@Value("${rabbit.direct.key}")
    	private String rabbitDirectKey;	
    	@Value("${rabbit.direct.exchange}")
    	private String rabbitDirectExchange;
    	
    	@Autowired
    	AmqpTemplate amqpTemplate;
    	
    	public void sendMessage(ParamReq paramReq){ 
    		amqpTemplate.convertAndSend(rabbitDirectExchange,rabbitDirectKey,paramReq);
    	}
    }

这里的amqpTemplate是Spring自动生成的bean。消息的发送指定了rabbitDirectExchange、指定了rabbitDirectKey。

#### 4.2 Fanout模式 ####

PomitFanoutMessageService ：

这里的amqpTemplate是Spring自动生成的bean。消息的发送指定了rabbitFanoutExchange，并没有routingKey.

#### 4.3 Topic模式 ####

PomitTopicMessageService ：

package com.cff.springbootwork.rabbitmq.service;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    
    import com.cff.springbootwork.rabbitmq.dto.ParamReq;
    
    @Service
    public class PomitTopicMessageService { 
    	@Value("${rabbit.topic.exchange}")
    	private String rabbitTopicExchange;
    	
    	@Value("${rabbit.direct.key}")
    	private String rabbitDirectKey;
    	
    	@Value("${rabbit.topic.key}")
    	private String rabbitTopicKey;
    	
    	@Autowired
    	AmqpTemplate amqpTemplateTopic;
    	
    	public void sendMessage(ParamReq paramReq){ 
    		amqpTemplateTopic.convertAndSend(rabbitTopicExchange, rabbitDirectKey, paramReq);
    	}
    	
    	public void sendMessage2(ParamReq paramReq){ 
    		amqpTemplateTopic.convertAndSend(rabbitTopicExchange, rabbitTopicKey, paramReq);
    	}
    }

这里的amqpTemplate是Spring自动生成的bean。消息的发送指定了rabbitTopicExchange、指定了routingKey,这里发送了两个key，是为了分别测试指定key和模糊匹配key的接收情况。

### 五、消息接收方（消费者） ###

我么可以使用@RabbitListener来定义接收方，每一个queue定义一个@RabbitListener即可。因为queue是和模式无关的。参数可以直接传递实体过来，因为Spring根据我们上面定义的MessageConverter会自动将数据转换为实体。

PomitMessageListener:

package com.cff.springbootwork.rabbitmq.listener;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import com.cff.springbootwork.rabbitmq.dto.ParamReq;
    
    @Component
    public class PomitMessageListener{ 
    
    	@RabbitListener(queues = "${rabbit.direct.queue}")
    	public void handleMessage(ParamReq paramReq) { 
    		
    		System.out.println(paramReq);
    	}
    	@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);
    	}
    }

### 六、过程中用到的实体及测试类 ###

ParamReq ：

RabbitWeb:

详细完整的实体及测试用例，可以访问[品茗IT-博客《SpringBoot入门建站全系列（十八）整合RabbitMQ（AMQP类消息队列）》][IT-]进行查看

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

[Spring组件化构建][Spring]

[SpringBoot组件化构建][SpringBoot]

[SpringCloud服务化构建][SpringCloud]

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

[IT-]: https://www.pomit.cn/p/2008148377356801
[IT]: https://www.pomit.cn
[Spring]: https://www.pomit.cn/java/spring/spring.html
[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
[SpringBoot 1]: https://www.pomit.cn/p/169774771175680
[IT 1]: /images/20211129/dc2610d52bc1441c9f9c2240a6af76f9.png