spring boot整合RabbitMQ

作者：AllinToyou | 2024-04-17 06:47:41

踩

spring boot整合rabbitmq

文章目录

前言

一、环境准备

引入依赖生产者和消费都引入这个依赖

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

在生产者application.properties里面配置


#配置rabbitmq
spring.rabbitmq.host=192.168.91.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
#默认虚拟主机
spring.rabbitmq.virtual-host=/
 
#开启生产者通过信道到交换机confirm机制
spring.rabbitmq.publisher-confirm-type=correlated
 
#开启生产者通过交换机到队列
spring.rabbitmq.publisher-returns=true

消费者的配置文件


#配置rabbitmq
spring.rabbitmq.host=192.168.91.133
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtual-host=/
 
#消费端
server.port=8081
#开启消费者端的简单模式的手动确认消息机制
spring.rabbitmq.listener.simple.acknowledge-mode=manual

二、使用步骤

在生产者的spring boot项目中创建配置类


package com.rcg.springbootrabbitmqproducer.config;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Binding;
 
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * @author ：1452863884@qq.com rcg
 * @date ：Created in 2023/3/8 10:15
 * @description：rabbitmq消息中间件配置类
 * @modified By：
 * @version:1.0
 */
@Configuration
public class MyRabbitMqConfig {
 
 
    private String queue01="my_queue01";
    private String queue02="my_queue02";
    private String myExchange="my_top_exchange";
 
 
 
    @Bean(value = "queue01")
    public Queue queue(){
        Queue build = QueueBuilder.durable(queue01).build();
        return build;
    }
    @Bean(value = "queue02")
    public Queue queue02(){
        Queue build = QueueBuilder.durable(queue02).build();
        return build;
    }
 
    @Bean(value = "exchange01")
    public Exchange exchange(){
        Exchange exchange = ExchangeBuilder.topicExchange(myExchange).build();
        return exchange;
    }
 
    @Bean
    public Binding binding(Exchange exchange01,Queue queue01){
        Binding binding = BindingBuilder.bind(queue01)
                .to(exchange01)
                .with("*.orange.*").noargs();
        return binding;
    }
    @Bean
    public Binding binding02(Exchange exchange01,Queue queue02){
        Binding binding = BindingBuilder
                .bind(queue02).to(exchange01).with("*.*.rabbit").noargs();
        return binding;
    }
    @Bean
    public Binding binding03(Exchange exchange01,Queue queue02){
        Binding binding = BindingBuilder.bind(queue02)
                .to(exchange01).with("*.*.lazy.#").noargs();
        return binding;
    }
 
 
}

在测试文件中测试


package com.rcg.springbootrabbitmqproducer;
 
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
 
import javax.annotation.Resource;
import java.util.UUID;
 
@SpringBootTest
class SpringbootRabbitmqProducerApplicationTests {
 
 
    @Resource
    private RabbitTemplate rabbitTemplate;
 
 
    /**
     * 测试死信队列+ttl
     * 就是给正常队列设置一些参数
     * pt_exchange
     * 比如参数长度 超过该长度的信息, 就把信息弄到死信队列里面
     */
    @Test
    public void testHeadQueue(){
 
 
        for (int i = 0; i < 20; i++) {
            //设置id号
            MessagePostProcessor messagePostProcessor=new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
 
                    return message;
                }
            };
            rabbitTemplate.convertAndSend("pt_exchange","","今天天气不错",messagePostProcessor);
 
        }
 
 
 
 
    }
 
 
 
 
    /**
     * 测试消费者手动确认消息被消费
     */
    @Test
    public void testHandAck(){
 
 
        //设置id号
        MessagePostProcessor messagePostProcessor=new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                //设置消息的过期时间
                //如果对列也设置了过期时间那么就是谁的过期时间短就先过期谁的
                message.getMessageProperties().setExpiration("5000");
                return message;
            }
        };
        rabbitTemplate.convertAndSend("simple_exchange","","今天天气不错",messagePostProcessor);
 
    }
 
 
 
 
    /**
     * 测试交换机到队列的返回机制
     */
    @Test
    public void testReturn(){
 
        //只有在发生失败时才会触发
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                System.out.println("消费者从交换机到队列失败");
            }
        });
 
        //设置id号
        MessagePostProcessor messagePostProcessor=new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                return message;
            }
        };
 
 
        //测试失败 就是来个没有的规则就转发不到队列
        rabbitTemplate.convertAndSend("my_top_exchange","xx.lazy.ads","今天天气真冷",messagePostProcessor);
 
 
 
 
    }
 
 
 
 
    /**
     * 测试信道到交换机确认机制
     */
    @Test
    public void testConfirm(){
 
        //当发生消息到交换机无论成功还是失败--都会触发该方法
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if (b==false){
                    //弥补机制--重新发送--取消订单--业务
                    System.out.println("重新发送信息");
                }else{
                    System.out.println("消息发送成功");
                }
            }
        });
 
        //设置id号
        MessagePostProcessor messagePostProcessor=new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                return message;
            }
        };
        //测试成功
        //rabbitTemplate.convertAndSend("my_top_exchange","xx.*.lazy.ads","今天天气真冷",messagePostProcessor);
        //测试失败 就是发送到一个不存在的交换机 来模拟故障
        rabbitTemplate.convertAndSend("my_top_exchange1","xx.*.lazy.ads","今天天气真冷",messagePostProcessor);
 
 
    }
 
 
 
 
 
    /**
     * 测试主题模式
     */
    @Test
    public void testTopExchange(){
        //设置id号
        MessagePostProcessor messagePostProcessor=new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                return message;
            }
        };
 
 
        rabbitTemplate.convertAndSend("my_top_exchange","xx.*.lazy.ads","今天天气真冷",messagePostProcessor);
 
    }
 
 
    @Test
    void contextLoads() {
 
        rabbitTemplate.convertAndSend("my_exchange01","info","今天天气不错");
 
    }
 
}

消费者的代码


package com.rcg.springbootrabbitmqcluseer;
 
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
import java.io.IOException;
 
/**
 * @author ：1452863884@qq.com rcg
 * @date ：Created in 2023/3/8 13:51
 * @description：监听消息队列消费
 * @modified By：
 * @version:1.0
 */
@Component
public class MyListener {
 
 
    @RabbitListener(queues = {"simple_queue"})
    public void getMsg(Message message, Channel channel){
        byte[] body = message.getBody();
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
 
        try {
            //int a=10/0;
            //手动确认 批量
            channel.basicAck(deliveryTag,true);
 
        }catch (Exception e){
            //重发
            try {
                //多次重发
                channel.basicNack(deliveryTag,true,true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
 
        }
 
 
 
        //得到消息的id
        String messageId = message.getMessageProperties().getMessageId();
        System.out.println(messageId+"队列信息= " + new String(body));
 
    }
 
 
}

2.1 RabbitMQ高级特性

2.1.1 消息的可靠性传递

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
confirm 确认模式
return 退回模式

rabbitmq 整个消息投递的路径为：
producer--->rabbitmq broker--->exchange--->queue--->consumer
消息从 producer 到 exchange 则会返回一个 confirmCallback 。
消息从 exchange-->queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递

设置ConnectionFactory的publisher-confirm-type: correlated开启确认模式。

使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

设置ConnectionFactory的publisher-returns="true" 开启退回模式。

使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后执行回调函数returnedMessage。

在RabbitMQ中也提供了事务机制，但是性能较差，此处不做讲解。
使用channel下列方法，完成事务控制：
txSelect(), 用于将当前channel设置成transaction模式
txCommit()，用于提交事务
txRollback(),用于回滚事务

2.1.2 Consumer Ack

ack指Acknowledge，确认。表示消费端收到消息后的确认方式。
有三种确认方式：
自动确认：acknowledge="none"
手动确认：acknowledge="manual"
根据异常情况确认：acknowledge="auto"，（这种方式使用麻烦，并且不常用，不作讲解）

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息队列中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

在rabbit:listener-container标签中设置acknowledge属性，设置ack方式 none：自动确认，manual：手动确认

如果在消费端没有出现异常，则调用channel.basicAck(deliveryTag,false);方法确认签收消息

如果出现异常，则在catch中调用 basicNack或 basicReject，拒绝消息，让MQ重新发送消息。