SpringBoot入门建站全系列（十八）整合RabbitMQ（

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 提供在线支持：

一键快速构建Spring项目工具

一键快速构建SpringBoot项目工具

一键快速构建SpringCloud项目工具

一站式Springboot项目生成

Mysql一键生成Mybatis注解Mapper

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

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入门建站全系列（一）项目建立》。

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 ：

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 PomitFanoutMessageService {
    @Value("${rabbit.fanout.exchange}")
    private String rabbitFanoutExchange;
    @Autowired
    AmqpTemplate amqpTemplate;

    public void sendMessage(ParamReq paramReq) {
        amqpTemplate.convertAndSend(rabbitFanoutExchange, null, paramReq);
    }
}

这里的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类消息队列）》进行查看

快速构建项目

Spring组件化构建

SpringBoot组件化构建

SpringCloud服务化构建

喜欢这篇文章么，喜欢就加入我们一起讨论SpringBoot使用吧！

品茗IT交流群