正文 

当前位置： 首页 > IT > java教程

rabbitmq消息队列笔记(代码片段)

知道什么是码怪吗？ 知道什么是码怪吗？     2022-11-29     142

关键词：

目录

发布确认高级

模拟消息丢失

消息回退

回退测试

备份交换机 

备份交换机测试

优先队列

实现优先队列

队列设置优先级

消息设置优先级

发布确认高级

在生产环境中由于一些不明原因，导致 RabbitMQ 重启，在 RabbitMQ 重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。如何才能进行 RabbitMQ 的消息可靠投递呢？特别是在比较极端的情况，RabbitMQ 集群不可用时，无法投递的消息该如何处理呢？

模拟消息丢失

目录结构

 

配置类新增代码

spring.rabbitmq.publisher-confirm-type = correlated

confirmConfig类代码

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

// 配置类：发布确认（高级）
@Configuration
public class ConfirmConfig 

    //交换机
    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
    //队列
    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
    //RoutingKey
    public static final String CONFIRM_ROUTING_KEY = "key1";

    //声明交换机
    @Bean
    public DirectExchange confirmExchange() 
        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
    

    @Bean
    public Queue confirmQueue() 
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    
    
    //绑定
    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange) 
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);

MyCallBack类代码

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback 


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() 
        rabbitTemplate.setConfirmCallback(this);
    

    /*
     * 交换机确认回调方法,发消息后，交换机接收到了就回调
     *   1.1 correlationData：保存回调消息的ID及相关信息
     *   1.2 b:交换机收到消息，为true
     *   1.3 s:失败原因，成功为null
     *
     * 发消息，交换机接受失败，也回调
     *   2.1 correlationData：保存回调消息的ID及相关信息
     *   2.2 b:交换机没收到消息，为false
     *   2.3 s:失败的原因
     * */

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) 
        String id = correlationData != null ? correlationData.getId() : "";
        if (b) 
            log.info("交换机已经收到id为：的信息", id);
         else 
            log.info("交换机未收到id为：的信息，失败的原因：", id, s);

Consumer类代码

@Slf4j
@Component
public class Consumer 

    @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
    public void receiveConfirmMessage(Message message) 
        String s = new String(message.getBody());
        log.info("接收到的队列confirm.queue消息：", s);

ProducerController类代码 

// 开始发消息 测试确认
@RestController
@Slf4j
@RequestMapping("/confirm")
public class ProducerController 

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //发消息
    @GetMapping("/sendMessage/message")
    public void sendMessage(@PathVariable String message) 
        CorrelationData correlationData1 = new CorrelationData("1");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
                , ConfirmConfig.CONFIRM_ROUTING_KEY
                , message + "key1", correlationData1);
        log.info("发送消息内容：", message + " key1");

        CorrelationData correlationData2 = new CorrelationData("2");
        rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
                , ConfirmConfig.CONFIRM_ROUTING_KEY + "2"
                , message + "key2", correlationData2);
        log.info("发送消息内容：", message + " key2");

测试结果

localhost:8080/confirm/sendMessage/123

发送了两条消息给交换机，但是只有一个消息被接收，另一个消息丢失。 

消息回退

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者，让生产者自己处理消息。

回退测试

配置类新增代码

spring.rabbitmq.publisher-returns = true

MycallBack类内容更新

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback 

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() 
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) 
        String id = correlationData != null ? correlationData.getId() : "";
        if (b) 
            log.info("交换机已经收到id为：的信息", id);
         else 
            log.info("交换机未收到id为：的信息，失败的原因：", id, s);
        
    

    //可以在当消息传递过程中不可达目的的时将消息返回给生产者
    //只有不可达目的地的时候才可回退
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) 
        log.error("消息，被交换机退回，退回原因：，路由key:", new String(returnedMessage.getMessage().getBody())
                , returnedMessage.getExchange()
                , returnedMessage.getReplyText()
                , returnedMessage.getRoutingKey());

测试结果

发送两条消息，不可路由的消息将会被退回。

备份交换机 

备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时， 就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由 备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。

备份交换机测试

更新ConfirmConfig类代码

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

// 配置类：发布确认（高级）
@Configuration
public class ConfirmConfig 

    //交换机
    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
    //队列
    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
    //RoutingKey
    public static final String CONFIRM_ROUTING_KEY = "key1";
    //备份交换机
    public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
    //备份队列
    public static final String BACKUP_QUEUE_NAME = "backup_queue";
    //报警队列
    public static final String WARNING_QUEUE_NAME = "warning_queue";

    //声明交换机
    @Bean
    public DirectExchange confirmExchange() 
        return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
                .withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME).build();
    

    //声明队列
    @Bean
    public Queue confirmQueue() 
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    

    //绑定
    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange) 
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    

    //备份交换机
    @Bean
    public FanoutExchange backupExchange() 
        return new FanoutExchange(BACKUP_EXCHANGE_NAME);
    

    //备份队列
    @Bean
    public Queue backupQueue() 
        return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
    

    //报警队列
    @Bean
    public Queue warningQueue() 
        return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
    

    //备份交换机与备份队列绑定
    @Bean
    public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
                                                    @Qualifier("backupExchange") FanoutExchange backupExchange) 
        return BindingBuilder.bind(backupQueue).to(backupExchange);
    

    //备份交换机与报警队列绑定
    @Bean
    public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue backupQueue,
                                                     @Qualifier("backupExchange") FanoutExchange backupExchange) 
        return BindingBuilder.bind(backupQueue).to(backupExchange);

新增WarningConsumer类

// 报警消费者
@Component
@Slf4j
public class WarningConsumer 

    //接受报警消息
    @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
    public void receiveWarningMsg(Message message) 
        String msg = new String(message.getBody());
        log.error("报警发现不可路由消息：", msg);

测试结果

不可路由的消息被报警队列接收到了。输出的是报警发现不可路由消息，并不是消息回退。

mandatory参数与备份交换机可以一起使用时，如果两者同时开启，备份交换机优先级高。

优先队列

带有优先级的队列，处理消息时根据优先级排序，优先级高的先处理。类似于工厂或者商店的大订单先处理一个道理。

实现优先队列

队列实现优先级需要做的事情有如下事情：（1）队列需要设置为优先级队列，（2）消息需要设置消息的优先级，消费者需要等待消息已经发送到队列中才去消费，这样才有机会对消息进行排序。

队列设置优先级

Map<String, Object> params = new HashMap();
params.put("x-max-priority", 10);
channel.queueDeclare("hello", true, false, false, params);

消息设置优先级

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();

生产者

public class Producer 
    // 队列名称
    public static final String QUEUE_NAME = "hello";

    // 发消息
    public static void main(String[] args) throws IOException, TimeoutException 
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 工厂IP连接RabbitMQ的队列
        factory.setHost("192.168.193.147");
        // 用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123");
        factory.setPort(5672);
        // 创建连接
        Connection connection = factory.newConnection();
        // 获取信道
        Channel channel = connection.createChannel();
        Map<String, Object> arguments = new HashMap<>();
        //官方允许是0-255之间，此处设置10，允许优化级范围为0-10，不要设置过大，浪费CPU与内存
        arguments.put("x-max-priority", 10);
        channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
        // 发消息
        for (int i = 0; i < 10; i++) 
            String message = "info" + i;
            if (i == 5) //当i等于5时设置这一条消息优先级为5
                AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();
                channel.basicPublish("", QUEUE_NAME, properties, message.getBytes(StandardCharsets.UTF_8));
             else 
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            
        
        System.out.println("消息发送完毕！");

消费者

public class Consumer 
    // 队列名称
    public static final String QUEUE_NAME = "hello";

    // 接受消息
    public static void main(String[] args) throws IOException, TimeoutException, TimeoutException 
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.193.147");
        factory.setUsername("admin");
        factory.setPassword("123");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明 接受消息
        DeliverCallback deliverCallback = (consumerTag, message) -> 
            System.out.println(new String(message.getBody()));
        ;
        // 声明 取消消息
        CancelCallback cancelCallback = consumer -> 
            System.out.println("消息消费被中断");
        ;
        System.out.println("等待接收消息……");
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

测试结果

如果我们先启动消费者，那么出现的结果将会按照顺序显示，因为没有出现消息的堆积，来一个就处理一个。

然而当我们先启动生产者，产生消息在信道当中堆积，那么处理时就会先按照优先级排序再进行处理。

rabbitmq消息队列笔记(代码片段)

...迟队列优化 插件实现延迟队列安装插件测试插件交换机RabbitMQ消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。生产者 查看详情

rabbitmq消息队列笔记(代码片段)

...发布确认高级在生产环境中由于一些不明原因，导致RabbitMQ 重启，在RabbitMQ重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。如何才能进行RabbitMQ的消息可靠投递呢？特别是在比较极端 查看详情

rabbitmq消息队列笔记(代码片段)

...发布确认高级在生产环境中由于一些不明原因，导致RabbitMQ 重启，在RabbitMQ重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。如何才能进行RabbitMQ的消息可靠投递呢？特别是在比较极端 查看详情

rabbitmq学习笔记(代码片段)

目录一、引⾔二、RabbitMQ介绍三、RabbitMQ安装四、RabbitMQ架构1.官⽅的简单架构图2.RabbitMQ的完整架构图3.查看图形化界⾯并创建⼀个VirtualHost五、RabbitMQ的队列模式1.RabbitMQ的通讯⽅式2.HelloWorld模式-简单队列模式3.work队列模式：... 查看详情

rabbitmq笔记(代码片段)

RabbitMQ1MQ引言1.1什么是MQMQ(MessageQuene):翻译为消息队列,通过典型的生产者和消费者模型,生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接... 查看详情

rabbitmq笔记(代码片段)

RabbitMQ1MQ引言1.1什么是MQMQ(MessageQuene):翻译为消息队列,通过典型的生产者和消费者模型,生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接... 查看详情

springcloudstream消息驱动设计思想以及整合rabbitmq消息队列案例--学习笔记(代码片段)

一，RabbitMQ的安装和配置并启动参见RabbitMQ的安装和配置https://blog.csdn.net/weixin_43025151/article/details/123186641RabbitMQ启动成功：二，SpringCloudStream消息驱动1，设计思想一个标准的消息队列MQ,如下图：为什么用Cloud 查看详情

rabbitmq学习笔记(代码片段)

视频教程【编程不良人】MQ消息中间件之RabbitMQ以及整合SpringBoot2.x实战教程1.MQ引言1.1什么是MQMQ(MessageQuene)：消息队列，也叫消息中间件。通过典型的生产者和消费者模型，生产者不断向消息队列中生产消息，消费... 查看详情

rabbitmq消息队列笔记(代码片段)

目录RabbitMQ概念四大核心概念核心部分测试"HelloWorld"测试WorkQueues消息应答自动应答手动应答消息应答方法消息自动重新入队手动应答测试持久化队列持久化消息持久化不公平分发预取值消息确认开启发布确认单个确认发布... 查看详情

rabbitmq消息队列笔记(代码片段)

目录RabbitMQ概念四大核心概念核心部分测试"HelloWorld"测试WorkQueues消息应答自动应答手动应答消息应答方法消息自动重新入队手动应答测试持久化队列持久化消息持久化不公平分发预取值消息确认开启发布确认单个确认发布... 查看详情

rabbitmq消息队列笔记(代码片段)

...迟队列优化 插件实现延迟队列安装插件测试插件交换机RabbitMQ消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。生产者只能将消息发送到交换机(exchange)。交换机工作内容非常简单，一方面它接收来... 查看详情

rabbitmq学习笔记四：消息publish/subscrible(代码片段)

...任务意外被杀死，会转发给另外的工作者，参见[Rabbitmq学习笔记三：工作队列workqueue](http://blog.csdn.net/u010416588/article/details/54629334)。一发布、订阅简介 查看详情

rabbitmq学习笔记(代码片段)

rabbitmqAMQP协议安装如何使用AMQP协议生产者消费者队列一个队列存多条消息消息连接发消息和接收消息都是必须先建立连接通道（channel）一个连接有多个通道，消息会在通道里面流转，之后到达消费者交换机（... 查看详情

rabbitmq学习笔记(代码片段)

文章目录1.RabbitMq(消息中间件)1.概念：是基于队列模式实现的异步/同步的传输数据。2.传统的Http请求存在哪些缺点？3.mq的使用场景？4.mq服务器如何保证消息不丢失？5.VirtualHost?Exchange分派我们的消息存放在哪个队... 查看详情

rabbitmq学习笔记(代码片段)

文章目录RabbitMq(消息中间件)1.概念：是基于队列模式实现的异步/同步的传输数据。2.传统的Http请求存在哪些缺点？3.mq的使用场景？4.mq服务器如何保证消息不丢失？5.VirtualHost?Exchange分派我们的消息存放在哪个队列... 查看详情

rabbitmq入门笔记(代码片段)

文章目录第一章安装RabbitMQ1.1安装Erlang1.2安装RabbitMQ1.3RabbitMQ后台管理系统1.3.1概述1.3.2连接1.3.3信道1.3.4交换机1.3.4队列1.3.4admin1.4配置用户与虚拟主机1.4.1配置用户1.4.2新增虚拟主机第二章RabbitMQ配置2.1rabbitmq架构2.2web后台创建消息... 查看详情

rabbitmq实战：理解消息通信(代码片段)

本系列是「RabbitMQ实战：高效部署分布式消息队列」书籍的总结笔记。前段时间总结完了「深入浅出MyBatis」系列，对MyBatis有了更全面和深入的了解，在掘金社区也收到了一些博友的喜欢，很高兴。另外，短暂的陪产假就要结束... 查看详情

rabbitmq学习笔记五：rabbitmq之优先级消息队列

RabbitMQ优先级队列注意点：1、只有当消费者不足，不能及时进行消费的情况下，优先级队列才会生效2、RabbitMQ3.5以后才支持优先级队列代码在博客：RabbitMQ学习笔记三：Java实现RabbitMQ之与Spring集成 最后面有下载地址，只是做... 查看详情