RabbitMQ整合SpringBoot

￡神魔★判官ぃ 2022-10-07 02:56 5阅读 0赞

# RabbitMQ整合SpringBoot #

## 一.RabbitMQ整合SpringBoot（手动ack） ##

### 1.创建SpringBoot工程 ###

### 2.导入依赖 ###

<dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>

### 3.编写配置文件 ###

**application.yml**

>     spring:
>       rabbitmq:
>         host: 192.168.247.128
>         port: 5672
>         username: root
>         password: 123
>         virtual-host: /root
>         listener:
>           simple:
>             acknowledge-mode: manual  #手动ack
>         publisher-confirm-type: simple #开启config机制
>         publisher-returns: true #开启return机制

### 4.声明exchange，queue ###

@Configuration
    public class RabbitMQConfig {
    
        //1. 创建exchange - topic
        @Bean   // 把方法的返回值放入到IOC容器中，方法的名字就是bean的ID
        public TopicExchange getTopicExchange(){
            return new TopicExchange("boot-topic-exchange",true,false);
        }
        //2. 创建queue
        @Bean
        public Queue getQueue(){
            return new Queue("boot-queue",true,false,false,null);
        }
        //3. 绑定在一起
        @Bean
        public Binding getBinding(TopicExchange topicExchange,Queue queue){
            return BindingBuilder.bind(queue).to(topicExchange).with("*.red.*");
        }
    }

### 5.发布消息到RabbitMQ ###

**在test测试方法中**

@SpringBootTest
    class SpringbootRabbitmqApplicationTests {
    
    
        @Autowired
        //    xxxxxxTemplate;
        //    JdbcTemplate; // Spring和JDBC的整合
        //    RedisTemplate; // SPring和Redis
        //    SqlSessionTemplate; // Sprign和MyBatis
        //    HibnerateTemplate;
        // Spring和RabbitMQ整合的bean，可以通过这个bean来操作MQ
        private RabbitTemplate rabbitTemplate;
    
        @Test
        void contextLoads() {
            System.out.println(rabbitTemplate);
            for (int i = 0; i < 1; i++) {
                rabbitTemplate.convertAndSend("boot-topic-exchange", "text", "红色大狼狗" + i);
            }
            System.out.println("消息发送完成。。。。。");
        }
    }

### 6.创建消费者监听消息 ###

>     // 手动ack  
>     channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

@Component
    public class Consumer {
    
    //    @RabbitListener(queues = "boot-queue")
    //    public void getMessage(String message) {
    //        System.out.println("接受到的消息:" + message);
    //    }
    
        @RabbitListener(queues = "boot-queue")  // 指定监听那个队列中的消息
        public void getMessage(String data, Channel channel, Message message) {
            // 因为在消费的过程中可能会出现异常，所以把所有的业务代码都要用try包起来
    
            try {
                // 1、消费这个消息
                System.out.println("监听者获取到的消息是：" + data);
                Thread.sleep(500);
    
    //            int i = 10 / 0; // 模拟消费过程中出现了异常
    
                // 2.获取消息的表示
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                try {
                    // 3.把消息消费成功，告诉MQ这个消息我已经消费了
                    // 第一个参数是应答消息的标识(10)
                    // 第二个参数是否批量应答(如果开启的批量应答，会把小于10的消息全部应答)
                    channel.basicAck(deliveryTag, false);
                    System.out.println("消息【" + deliveryTag + "】应答MQ--》已完成");
    
    
                } catch (IOException e) {
    //                e.printStackTrace();
                }
            } catch (InterruptedException e) {
                System.out.println("消费者在消费的过程中出现了异常。。。");
                // 既然出现异常了，说明消息没有正常消费，应该通知MQ这个消息我无法消费，把这个消息转发给其他的消费者
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
    
                try {
                    // 应答MQ，这个消息我消费不了
                    // 第一个参数是应答消息的标识(10)
                    // 第二个参数是否批量应答(如果开启的批量应答，会把小于10的消息全部应答)
                    // 第三个参数:该消息是否重入队列，压倒队列的头部
                    //          false:该消息就会丢失
                    channel.basicNack(deliveryTag, false, true);
                    System.out.println("消息【" + deliveryTag + "】应答MQ--》未完成");
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
    
        }
    }

## 解决：消息是否到达交换机?消息是否到达队列? ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70][]

>  *  采用Confirm保证消息到达exchange，无法保证消息可以被exchange分发到指定queue。
>  *  exchange是不能持久化消息的，queue是可以持久化消息。
>  *  采用Return机制来监听消息是否从exchange送到了指定的queue中。

### 7.开启Confirm和Return机制 ###

@Component
    public class RabbitMQConfirm implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
    
        @Autowired
       private RabbitTemplate rabbitTemplate;
    
        // init-mtehod<bean init-method>
        @PostConstruct // 在对象实例化完成以后调用
        public void init(){
            // 消息发送出去后就会调用这个类中的confirm();
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnsCallback(this);
        }
    
        // confirm是确认消息是否到达交换机
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String context) {
            if (ack) {
                System.out.println("消息已经发送到了交换机上:" + context);
            }else {
                System.out.println("消息没有发送到了交换机上:" + context);
            }
        }
    
        // 消息没有发送到队列时调用
        @Override
        public void returnedMessage(ReturnedMessage returnedMessage) {
            byte[] body = returnedMessage.getMessage().getBody();
            System.out.println("消息没有发送到队列"+returnedMessage.getRoutingKey()+"-->"+new String(body));
        }
    }

## 解决：消费者在消费消息时出现闪断的问题？ ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70 1][]

如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70 2][]

### 为了解决这个问题我们需要借助Redis，下一篇文章解决。 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70]: /images/20221005/c3c95065efd948b796315efb349d0e40.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70 1]: /images/20221005/d71cd6e6932742f7b31c3a70ab531842.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70 2]: /images/20221005/2eb713ee82d24526af87d5c230801cb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cTE1MTgzNDY4NjQ_size_16_color_FFFFFF_t_70 3]: /images/20221005/7a249fd68f444d7990981e3ffb02d8fb.png