SpringBoot整合使用RabbitMQ

不念不忘少年蓝@ 2022-05-19 00:44 229阅读 0赞

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。消息中间件在互联网公司的使用中越来越多，消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。

# 1、RabbitMQ介绍 #

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

# 2、springboot集成RabbitMQ #

springboot集成RabbitMQ非常简单，如果只是简单的使用配置非常少，springboot提供了spring-boot-starter-amqp项目对消息各种支持。

## 2.1 引入依赖 ##

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

## 2.2 配置文件 ##

在application.yml配置文件中加入RabbitMQ配置信息

server:
      port: 10001
    
    spring:
      application:
        name: rabbitmq-hello
    
      rabbitmq:
          host: localhost
          port: 5672
          username: guest
          password: guest

## 2.3 编写RabbitConfig类 ##

RabbitConfig类里面设置很多个*EXCHANGE,QUEUE,ROUTINGKEY，是为了接下来的不同使用场景，*交换机，队列，通过路由关键字进行绑定。

package com.rabbit.mq;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    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.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.beans.factory.config.ConfigurableBeanFactory;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Scope;
    
    /**
     * 编写RabbitConfig类，类里面设置很多个EXCHANGE,QUEUE,ROUTINGKEY，是为了接下来的不同使用场景。
     *
     * Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
     * Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。
     * Queue:消息的载体,每个消息都会被投到一个或多个队列。
     * Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
     * Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
     * vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
     * Producer:消息生产者,就是投递消息的程序.
     * Consumer:消息消费者,就是接受消息的程序.
     * Channel:消息通道,在客户端的每个连接里,可建立多个channel.
     */
    @Configuration
    public class RabbitConfig {
    
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        @Value("${spring.rabbitmq.host}")
        private String host;
    
        @Value("${spring.rabbitmq.port}")
        private int port;
    
        @Value("${spring.rabbitmq.username}")
        private String username;
    
        @Value("${spring.rabbitmq.password}")
        private String password;
    
    
        public static final String EXCHANGE_A = "my-mq-exchange_A";
        public static final String EXCHANGE_B = "my-mq-exchange_B";
        public static final String EXCHANGE_C = "my-mq-exchange_C";
    
    
        public static final String QUEUE_A = "QUEUE_A";
        public static final String QUEUE_B = "QUEUE_B";
        public static final String QUEUE_C = "QUEUE_C";
    
        public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";
        public static final String ROUTINGKEY_B = "spring-boot-routingKey_B";
        public static final String ROUTINGKEY_C = "spring-boot-routingKey_C";
    
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost("/");
            connectionFactory.setPublisherConfirms(true);
            return connectionFactory;
        }
    
        @Bean
        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
        //必须是prototype类型
        public RabbitTemplate rabbitTemplate() {
            RabbitTemplate template = new RabbitTemplate(connectionFactory());
            return template;
        }
    
        /**
         * 针对消费者配置
         * 1. 设置交换机类型
         * 2. 将队列绑定到交换机
         *
         * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
         * HeadersExchange ：通过添加属性key-value匹配
         * DirectExchange:按照routingkey分发到指定队列
         * TopicExchange:多关键字匹配
         */
        @Bean
        public DirectExchange defaultExchange() {
            return new DirectExchange(EXCHANGE_A);
        }
    
        /**
         * 获取队列A
         * @return
         */
        @Bean
        public Queue queueA() {
            return new Queue(QUEUE_A, true); //队列持久
        }
    
        @Bean
        public Binding binding() {
    
            return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);
        }
    
        @Bean
        public Queue queueB() {
            return new Queue(QUEUE_B, true); //队列持久
        }
        @Bean
        public Binding bindingB(){
            return BindingBuilder.bind(queueB()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_B);
        }
    }

## 2.4 生产者 ##

package com.rabbit.mq;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.UUID;
    
    /**
     * 消息发送
     */
    @Component
    public class MsgProducer implements RabbitTemplate.ConfirmCallback {
    
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        //由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自动注入
        private RabbitTemplate rabbitTemplate;
        /**
         * 构造方法注入rabbitTemplate
         */
        @Autowired
        public MsgProducer(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果为单例的话，那回调就是最后设置的内容
        }
    
        public void sendMsg(String content) {
            CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
            //把消息放入ROUTINGKEY_A对应的队列当中去，对应的是队列A
            rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, content, correlationId);
        }
        /**
         * 回调
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            logger.info(" 回调id:" + correlationData);
            if (ack) {
                logger.info("消息成功消费");
            } else {
                logger.info("消息消费失败:" + cause);
            }
        }
    }

## 2.5 消费者 ##

### 2.5.1 类直接定义接收队列 ###

一个生产者可以多个消费者，多个消费者之间的分发是负载均衡的。

package com.rabbit.mq;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 消息接收
     */
    
    @Component
    @RabbitListener(queues = "QUEUE_A")
    public class MsgReceiver {
    
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        @RabbitHandler
        public void process(String content) {
            logger.info("接收处理队列A当中的消息： " + content);
        }
    
    }

### 2.5.2 在RabbitMQConfig类里面增加bean ###

@Bean
        public SimpleMessageListenerContainer messageContainer() {
            //加载处理消息A的队列
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
            //设置接收多个队列里面的消息，这里设置接收队列A
            //假如想一个消费者处理多个队列里面的信息可以如下设置：
            //container.setQueues(queueA(),queueB(),queueC());
            container.setQueues(queueA());
            container.setExposeListenerChannel(true);
            //设置最大的并发的消费者数量
            container.setMaxConcurrentConsumers(10);
            //最小的并发消费者的数量
            container.setConcurrentConsumers(1);
            //设置确认模式手工确认
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            container.setMessageListener(new ChannelAwareMessageListener() {
                @Override
                public void onMessage(Message message, Channel channel) throws Exception {
                    /**通过basic.qos方法设置prefetch_count=1，这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message，
                     换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它 */
                    channel.basicQos(1);
                    byte[] body = message.getBody();
                    logger.info("接收处理队列A当中的消息:" + new String(body));
                    /**为了保证永远不会丢失消息，RabbitMQ支持消息应答机制。
                     当消费者接收到消息并完成任务后会往RabbitMQ服务器发送一条确认的命令，然后RabbitMQ才会将消息删除。*/
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                }
            });
            return container;
        }

# 3. 测试 #

package com.rabbit.mq;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class HelloController {
    
        @Autowired
        private MsgProducer msgProducer;
    
        @RequestMapping("/send/{name}")
        public void helloworld(@PathVariable String name) {
            msgProducer.sendMsg(name);
        }
    }