Spring boot集成RabbitMQ

「爱情、让人受尽委屈。」 2022-07-13 14:18 178阅读 0赞

#### RabbitMQ简介 ####

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统  
MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

AMQP就是一个协议，是一个高级抽象层消息通信协议。

虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。也就是说AMQP是异步通讯的一个协议。

#### RabbitMQ使用场景 ####

在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。不过大多数不仅仅是无需即时返回，甚至是执行是否成功都无所谓。如果需要即时返回则可以使用[Dubbo][],Spring boot与Dubbo集成可以去看[Spring boot 集成Dubbox][Spring boot _Dubbox]

#### RabbitMQ依赖 ####

RabbitMQ并不是直接一个简单的jar包（Jar包只是提供一个基本的与RabbitMQ本身通讯的一些功能），和Dubbo相同，RabbitMQ也需要其他软件来运行，以下是RabbitMQ运行所需要的软件

##### 1、Erlang #####

由于RabbitMQ软件本身是基于Erlang开发的，所以想要运行RabbitMQ必须要先按照Erlang

[Erlang官网][Erlang]

[Erlang下载地址][Erlang 1]

##### RabbitMQ #####

RabbitMQ才是实现消息队列的核心  
[RabbitMQ官网][RabbitMQ]

[RabbitMQ下载][RabbitMQ 1]

###### 配置RabbitMQ ######

安装完成后，需要完成一些配置才能使用RabbitMQ，可以直接用cmd到RabbitMQ的安装目录下的sbin目录通过命令配置，也可以直接在开始菜单中直接找到RabbitMQ Command Prompt (sbin dir)运行直接到达RabbitMQ的安装目录的sbin，为了方便，我们先启用管理插件,执行命令

rabbitmq-plugins.bat enable rabbitmq_management

即可，==注意，这是在Windows下面，如果是Linux则没有bat后缀==  
然后我们添加一个用户，因为在外网环境没有用户的情况下是不能连接成功的，执行添加用户命令

rabbitmqctl.bat add_user springboot password

springboot是用户名，password是密码  
然后为了方便演示，我们给springboot赋予管理员权限，方便登录管理页面

rabbitmqctl.bat set_user_tags springboot administrator

给账号赋予虚拟主机权限

rabbitmqctl.bat set_permissions -p / springboot  .* .* .*

然后启动RabbitMQ服务  
访问[RabbitMQ管理页面http://localhost:15672][RabbitMQ_http_localhost_15672]即可看见登录页面，如果没有创建用户则可以用guest，guest登录，如果有创建用户则用创建的用户登录

#### 创建Springboot项目 ####

因为创建spring boot项目在前面的文章已经说过很多次了，所以这里就不多说了，如果不会可以去看其他[关于spring boot的博客][spring boot]

#### 添加RabbitMQ相关依赖 ####

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

每错，就是点配置，不过这样可能有点不理解，我还是把全部配置贴出来吧

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
    
      <groupId>wang.raye.rabbitmq</groupId>
      <artifactId>demo1</artifactId>
      <version>0.0.1-SNAPSHOT</version>
      <packaging>jar</packaging>
    
      <name>demo1</name>
      <url>http://maven.apache.org</url>
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
      </properties>
    <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.4.0.RELEASE</version>
        </parent>
    
        <dependencies>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>3.8.1</version>
                <scope>test</scope>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
    
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
    
        </dependencies>
    
    </project>

因为没有做其他操作，所以目前项目主要是依赖2个模块，一个Sprig boot，一个RabbitMQ

#### 添加配置类 ####

package wang.raye.rabbitmq.demo1;
    
    
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /** * rabbitmq 的配置类 * * @author Raye * @since 2016年10月12日10:57:44 */
    @Configuration  
    public class RabbitMQConfig { 
        /** 消息交换机的名字*/
        public static final String EXCHANGE = "my-mq-exchange";
        /** 队列key1*/
        public static final String ROUTINGKEY1 = "queue_one_key1";
        /** 队列key2*/
        public static final String ROUTINGKEY2 = "queue_one_key2";
    
        /** * 配置链接信息 * @return */
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory("127.0.0.1",5672);
    
            connectionFactory.setUsername("springboot");
            connectionFactory.setPassword("password");
            connectionFactory.setVirtualHost("/");
            connectionFactory.setPublisherConfirms(true); // 必须要设置
            return connectionFactory;
        }
    
        /** * 配置消息交换机 * 针对消费者配置 FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念 HeadersExchange ：通过添加属性key-value匹配 DirectExchange:按照routingkey分发到指定队列 TopicExchange:多关键字匹配 */  
        @Bean  
        public DirectExchange defaultExchange() {  
            return new DirectExchange(EXCHANGE, true, false);
        } 
        /** * 配置消息队列1 * 针对消费者配置 * @return */
        @Bean  
        public Queue queue() {  
           return new Queue("queue_one", true); //队列持久 
    
        }
        /** * 将消息队列1与交换机绑定 * 针对消费者配置 * @return */
        @Bean  
        public Binding binding() {  
            return BindingBuilder.bind(queue()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY1);  
        } 
    
        /** * 配置消息队列2 * 针对消费者配置 * @return */
        @Bean  
        public Queue queue1() {  
           return new Queue("queue_one1", true); //队列持久 
    
        }
        /** * 将消息队列2与交换机绑定 * 针对消费者配置 * @return */
        @Bean  
        public Binding binding1() {  
            return BindingBuilder.bind(queue1()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY2);  
        } 
        /** * 接受消息的监听，这个监听会接受消息队列1的消息 * 针对消费者配置 * @return */
        @Bean  
        public SimpleMessageListenerContainer messageContainer() {  
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());  
            container.setQueues(queue());  
            container.setExposeListenerChannel(true);  
            container.setMaxConcurrentConsumers(1);  
            container.setConcurrentConsumers(1);  
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认 
            container.setMessageListener(new ChannelAwareMessageListener() {
    
                public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
                    byte[] body = message.getBody();  
                    System.out.println("收到消息 : " + new String(body));  
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费 
    
                }  
    
            });  
            return container;  
        }  
        /** * 接受消息的监听，这个监听会接受消息队列1的消息 * 针对消费者配置 * @return */
        @Bean  
        public SimpleMessageListenerContainer messageContainer2() {  
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());  
            container.setQueues(queue1());  
            container.setExposeListenerChannel(true);  
            container.setMaxConcurrentConsumers(1);  
            container.setConcurrentConsumers(1);  
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认 
            container.setMessageListener(new ChannelAwareMessageListener() {
    
                public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
                    byte[] body = message.getBody();  
                    System.out.println("queue1 收到消息 : " + new String(body));  
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费 
    
                }  
    
            });  
            return container;  
        }  
    }

注意，为了更好的展示如何配置，我配置了2个消息队列，而本类除了链接配置哪里，其他都是针对消息消费者的，当然不管消息消费者和消息生产者都需要配置链接信息，而为了方便，所以本项目的消息消费者和生产者都在本项目，一般实际项目中不会在同一项目，由于注释很详细，我就不多说了

#### 发送消息 ####

为了方便发送消息，所以我直接写了一个Controller，通过访问接口的形式来调用发送消息的方法，话不多说，上代码

package wang.raye.rabbitmq.demo1;
    
    import java.util.UUID;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /** * 测试RabbitMQ发送消息的Controller * @author Raye * */
    @RestController
    public class SendController implements RabbitTemplate.ConfirmCallback{ 
        private RabbitTemplate rabbitTemplate;
        /** * 配置发送消息的rabbitTemplate，因为是构造方法，所以不用注解Spring也会自动注入（应该是新版本的特性） * @param rabbitTemplate */
        public SendController(RabbitTemplate rabbitTemplate){
            this.rabbitTemplate = rabbitTemplate;
            //设置消费回调
            this.rabbitTemplate.setConfirmCallback(this);
        }
        /** * 向消息队列1中发送消息 * @param msg * @return */
        @RequestMapping("send1")
        public String send1(String msg){
            String uuid = UUID.randomUUID().toString();
            CorrelationData correlationId = new CorrelationData(uuid);
            rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTINGKEY1, msg,
                    correlationId);
            return null;
        }
        /** * 向消息队列2中发送消息 * @param msg * @return */
        @RequestMapping("send2")
        public String send2(String msg){
            String uuid = UUID.randomUUID().toString();
            CorrelationData correlationId = new CorrelationData(uuid);
            rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTINGKEY2, msg,
                    correlationId);
            return null;
        }
        /** * 消息的回调，主要是实现RabbitTemplate.ConfirmCallback接口 * 注意，消息回调只能代表成功消息发送到RabbitMQ服务器，不能代表消息被成功处理和接受 */
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println(" 回调id:" + correlationData);
            if (ack) {
                System.out.println("消息成功消费");
            } else {
                System.out.println("消息消费失败:" + cause+"\n重新发送");
    
            }
        }
    }

==需要注意的是消息回调只能代表消息成功发送到RabbitMQ服务器==

然后我们启动项目，访问[http://localhost:8082/send1?msg=aaaa][http_localhost_8082_send1_msg_aaaa]  
会发现控制台输出了

收到消息 : aaaa
     回调id:CorrelationData [id=37e6e913-835a-4eca-98d1-807325c5900f]
    消息成功消费

当然回调id可能不同，如果我们访问[http://localhost:8082/send2?msg=bbbb][http_localhost_8082_send2_msg_bbbb] 则输出

queue1 收到消息 : bbbb
     回调id:CorrelationData [id=0cec7500-3117-4aa2-9ea5-4790879812d4]
    消息成功消费

#### 最后说两句 ####

因为本文主要是说明如何从零到springboot集成RabbitMQ，所以对于RabbitMQ的很多信息和用法没有说明，如果对RabbitMQ本身不太熟悉的可以去看看其他关于RabbitMQ的文章,附上本文[demo][]

[Dubbo]: https://github.com/alibaba/dubbo
[Spring boot _Dubbox]: http://raye.wang/2016/11/23/spring-boot-ji-cheng-dubbox/
[Erlang]: http://www.erlang.org/
[Erlang 1]: http://www.erlang.org/downloads
[RabbitMQ]: http://www.rabbitmq.com/
[RabbitMQ 1]: http://www.rabbitmq.com/download.html
[RabbitMQ_http_localhost_15672]: http://localhost:15672
[spring boot]: http://raye.wang/tag/springboot/
[http_localhost_8082_send1_msg_aaaa]: http://localhost:8082/send1?msg=aaaa
[http_localhost_8082_send2_msg_bbbb]: http://localhost:8082/send2?msg=bbbb
[demo]: https://github.com/RayeWang/rabbitmqdemo