rabbitMq部署(三) SpringBoot整合rabbitMQ收发数据

╰半夏微凉° 2022-02-05 09:45 277阅读 0赞

因要求需要部署一套rabbitmq，以前没接触过，所以在网上找到了方法:

以下是提供思路的大佬的原文：

1.Centos7 离线安装RabbitMQ,并配置集群\_\_\_擎正义之旗

https://blog.csdn.net/Alger\_magic/article/details/82868267

2.Caused by: com.rabbitmq.client.ShutdownSignalException: connection error; protocol method:解决方法\_\_\_czczcz\_

https://blog.csdn.net/czczcz\_/article/details/83379158

3.Spring Boot整合RabbitMQ详细教程\_\_\_梦里梦不出梦里梦的梦

https://blog.csdn.net/qq\_38455201/article/details/80308771

为了自己不弄晕，准备分三章来写，当前记录怎么springBoot整合rabbitmq收发数据

**目录**

1.添加依赖

2.编写Configure类

3.编写producer 和 consumer类

4.测试

--------------------

# 1.添加依赖 #

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

![20190505151623435.png][]

等待依赖注入后，去application.properties上添加rabbitmq的基本参数（可以只要前6个参数）

#对于rabbitmq的支持
    spring.rabbitmq.host=10.30.75.28
    spring.rabbitmq.port=5671
    spring.rabbitmq.username=trans
    spring.rabbitmq.password=trans
    spring.rabbitmq.virtual-host=/
    spring.rabbitmq.connection-timeout=15000
    #conrim模式
    spring.rabbitmq.publisher-confirms=true
    #returns机制
    spring.rabbitmq.publisher-returns=true
    #与returns配合的次属性
    spring.rabbitmq.template.mandatory=true
    #整合rabbitmq消费者
    spring.rabbitmq.listener.concurrency=5
    spring.rabbitmq.listener.max-concurrency=10
    #签收模式  手动签收模式
    spring.rabbitmq.listener.acknowledge-mode=manual
    #限流 - 每次只消费一条
    #spring.rabbitmq.listener.prefetch=1

# 2.编写Configure类 #

编写一个configure类 去配置rabbitmq 里面设置***EXCHANGE(交换机),QUEUE(队列),******ROUTINGKEY（路由关键字）***

/**
     * Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
     * Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。
     * Queue:消息的载体,每个消息都会被投到一个或多个队列。
     * Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
     * Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
     * vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
     * Producer:消息生产者,就是投递消息的程序.
     * Consumer:消息消费者,就是接受消息的程序.
     * Channel:消息通道,在客户端的每个连接里,可建立多个channel.
     */
    
    
    @Configuration
    @Slf4j
    public class RabbitMqConfigure {
    
        @Value("${spring.rabbitmq.host}") //ip
        private String host;
    
        @Value("${spring.rabbitmq.port}") //port
        private int port;
    
        @Value("${spring.rabbitmq.username}") //用户
        private String username;
    
        @Value("${spring.rabbitmq.password}") //密码
        private String password;
    
        public static final String EXCHANGE_GPS = "exchange_gps"; //交换机名称1
        public static final String EXCHANGE_COMMAND = "exchange_command"; //交换机名称2
    
        public static final String QUEUE_GPS = "queue_gps"; //队列名称1
        public static final String QUEUE_COMMAND = "queue_command"; //队列名称2
    
        public static final String ROUTINGKEY_GPS = "routingKey_gps"; //关键字1
        public static final String ROUTINGKEY_COMMAND = "routingKey_command"; //关键字2
    
    
        //创建连接工厂
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost("/");
            //回调函数 消息确认
            connectionFactory.setPublisherConfirms(true);
            //回调函数 消息返回
            connectionFactory.setPublisherReturns(true);
            return connectionFactory;
        }
    
        //设置的多例的 发送模版 必须是多例
        @Bean
        @Scope("prototype")
        public RabbitTemplate rabbitTemplate() {
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
            return rabbitTemplate;
        }
    
    }

# 3.编写producer 和 consumer类 #

先创建producer的类吧，这个主要是用于发送数据，交换机绑定队列，通过唯一的路由key来给通道发送数据

@Component
    @Slf4j
    public class RabbitMqProducer implements ConfirmCallback {  //这里实现了一个回调的接口
    
        //由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自动注入
        private RabbitTemplate rabbitTemplate;
    
        //构造自身
        @Autowired
        public RabbitMqProducer(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            rabbitTemplate.setConfirmCallback(this);
        }
    
        @Override //这个方法是确认消息是否发送成功
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            log.info("回调的的id: " + correlationData);
            if (ack) {
                log.info("rabbitmq消息确认成功");
            } else {
                log.error("rabbitmq消息确认失败", cause);
            }
        }
    
        //发送数据的方法
        public void sendMsg(String content) {
            CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
            rabbitTemplate.convertAndSend(RabbitMqConfigure.EXCHANGE_GPS, RabbitMqConfigure.ROUTINGKEY_GPS, content, correlationId);
    
        }
    }

然后这里需要在configure类中添加绑定关系 将交换机  队列 和 路由关键字进行绑定

@Configuration
    @Slf4j
    public class RabbitMqConfigure {
    
    ..........................................
    ..........................................
    ..........................................
    /**
         * 把交换机，队列，通过路由关键字进行绑定
         * 针对消费者配置
         * 1. 设置交换机类型
         * 2. 将队列绑定到交换机
         * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
         * HeadersExchange ：通过添加属性key-value匹配
         * DirectExchange:按照routingkey分发到指定队列
         * TopicExchange:多关键字匹配
         */
        
        //创建交换机
        @Bean
        public DirectExchange exChangeGps() {
            return new DirectExchange(EXCHANGE_GPS);
        }
    
        @Bean
        public DirectExchange exChangeCommand() {
            return new DirectExchange(EXCHANGE_COMMAND);
        }
    
    
        /*
         * 获取队列  设置了持久化的问题 durable abbitMQ默认将消息存储在内存中，若rabbitMQ宕机，那么所有数据就会丢失，所以在声明队列的时候可以声明将数据持久化 (第二个参数 true 就是设置持久化)
         * */
        @Bean
        public Queue QueueGps() {
            return new Queue(QUEUE_GPS, true);
        }
    
        /*
         * 获取队列  设置了持久化的问题 durable abbitMQ默认将消息存储在内存中，若rabbitMQ宕机，那么所有数据就会丢失，所以在声明队列的时候可以声明将数据持久化
         * */
        @Bean
        public Queue QueueCommand() {
            return new Queue(QUEUE_COMMAND, true);
        }
    
        /* **
         * @Description: 绑定关系 gps一个交换机，一条队列   command一个交换机 一个队列
         * @Param: []
         * @return: org.springframework.amqp.core.Binding
         * @Author:
         * @Date: 2019/4/28
         */
        @Bean
        public Binding bindingGps() {
            return BindingBuilder.bind(QueueGps()).to(exChangeGps()).with(RabbitMqConfigure.ROUTINGKEY_GPS);
        }
    
        @Bean
        public Binding bindingCommand() {
            return BindingBuilder.bind(QueueCommand()).to(exChangeCommand()).with(RabbitMqConfigure.ROUTINGKEY_COMMAND);
        }
    
    }

这个时候编写好了producer，可以编写一个consumer去监听获取数据

package com.hyxt.transferservice.rabbitmq;
    
    import com.hyxt.transferservice.config.RabbitMqConfigure;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Component;
    
    /**
     * @author ： 
     * @create : 2019/4/29 10:00
     * @descrption :rabbitmq消费者
     */
    @Component
    //监听注解  这里只是绑定了gps 的 队列 declare 是将其持久化,  绑定了交换机 还有 路由关键字
    // direct 模式 是 直接交换模式,所以用到了routingkey 还有其他模式
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = RabbitMqConfigure.QUEUE_GPS, declare = "true"), exchange = @Exchange(value = RabbitMqConfigure.EXCHANGE_GPS, durable = "true", type = "direct"), key = RabbitMqConfigure.ROUTINGKEY_GPS))
    public class RabbitMqConsumer {
    
        @RabbitHandler
        public void receiveMqMsg(String message) {
            System.out.println("收到数据...." + message);
        }
    }

还有其他的方法去监听 这里我根据我自己的情况 就暂时选择直接交换， 原文档在最上面。

# 4.测试 #

这里进行了一个测试，我刷10条数据进入rabbitmq，然后通过监听获取到

![20190505155416755.png][]

结果如下:

1.这个是在haproxy监控上看到数据流入与流出

![20190505155808617.png][]

2.在后台日志中看到获取到数据了

![20190505155911311.png][]

[20190505151623435.png]: /images/20220205/5b4ecb967386438782c745cd89bf48ca.png
[20190505155416755.png]: /images/20220205/e4fa34d5cd2b41dfb2918dae1d1a0a68.png
[20190505155808617.png]: /images/20220205/603af7580e544f358daba8ab3e0903cc.png
[20190505155911311.png]: /images/20220205/8d3490a692a641269b4bac6eb57a8c32.png