您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

尚硅谷RabbitMQ视频-023-进阶篇-可靠性-生端确-备交机

发布时间：2026-11-04 19:42:31编辑：雪饮阅读（）

上篇是通过消息确认机制实现可靠性，那么接下来通过建立备份交换机及备份队列方式实现可靠性。

在rabbitmq后台管理系统中建立备份交换机exchange.test.backup，因为进入备份交换机不带路由键的，所以类型必需是这个fanout

同样在rabbitmq后台创建和备份交换机绑定的队列queue.test.backup
然后进入新建的这个queue.test.backup队列中建立与备份交换机的绑定
只需要填写From exchange的值为备份交换机的名称exchange.test.backup

交换机要配置备份交换机，需要通过参数名Alternate exchange来指定，但该参数只能在建立时候指定，所以还得先删除原来的同名交换机，那么上篇中我们的同名交换机叫做exchange.direct.order，那么我们也在rabbitmq的后台中删除重建，重建时候在arguments后面的输入框下方的add的右边的Alternate exchange链接点击下会自动填充alternate-exchange参数名，然后参数值你可以手动填写为我们用来备份用的那个交换机的名称exchange.test.backup
新建之后重新绑定原来的队列，我们找到原来的队列queue.order然后去绑定这个新的交换机From exchange为exchange.test.backup，Routing key也是原来的order

接下来是代码层面
D:\rabbitMqLearning\src\main\resources\application.yml:
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: 123456
    virtual-host: /
    publisher-confirm-type: correlated #交换机的确认
    publisher-returns: true #队列的确认
logging:
  level:
    org.example.springBoot.listener.MyMessageListener: info
    org.example.confirmProducer.listener.MyMessageListener: info
    org.example.alternate.listener.MyMessageListener: info

其中新增的是org.example.alternate.listener.MyMessageListener: info

RabbitMQConsumerMainType.java：

package org.example.alternate;



import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;



@SpringBootApplication

public class RabbitMQConsumerMainType {

    //运行main方法使得消费者监听队列(注意默认会占用本地8080端口，若已经被占用，则可能无法启动)

    public static void main(String[] args) {

        SpringApplication.run(RabbitMQConsumerMainType.class, args);

    }

}

MyMessageListener.java：

package org.example.alternate.listener;



import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.Exchange;

import org.springframework.amqp.rabbit.annotation.Queue;

import org.springframework.amqp.rabbit.annotation.QueueBinding;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;



//没有这个注解，消费者不会监听队列，也不会接收消息

@Component

@Slf4j

public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";

    public static final String ROUTING_KEY = "order";

    public static final String QUEUE_NAME = "queue.order";



    /*

    * 监听队列，不存在则创建

    * */

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = QUEUE_NAME, durable = "true"),

            //这里仅仅指定了交换机名称，没有指定交换机类型，默认是direct交换机类型

            exchange = @Exchange(value = EXCHANGE_DIRECT),

            key = {ROUTING_KEY}

    ))



    // 写法二：仅仅监听队列，不自动创建

//@RabbitListener(queues = {QUEUE_NAME})

    public void processMessage(String dataString, Message message, Channel channel) {

        System.out.println("test confirmProducer listener 消费端接收到了消息：" + dataString);

    }

}

RabbitConfig.java：

package org.example.alternate.config;



import jakarta.annotation.PostConstruct;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.ReturnedMessage;

import org.springframework.amqp.rabbit.connection.CorrelationData;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Configuration;

@Configuration

@Slf4j

public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Autowired

    /*

    这里容易出现这个问题

    无法自动装配。找不到 'RabbitTemplate' 类型的 Bean。

    需要在类级别的注解上增加注解:

    @SpringBootApplication(scanBasePackages = "org.example")

    这里的注解扫描包名要同一个包或者父级包



    这里局部变量 可能报错：无法自动装配。找不到 'RabbitTemplate' 类型的 Bean。，

    这是ide上面的报错，在这里是误报，可以忽略。

    * */

    private RabbitTemplate rabbitTemplate;



    @PostConstruct

    public void initRabbitTemplate() {

        rabbitTemplate.setConfirmCallback(this);

        rabbitTemplate.setReturnsCallback(this);

    }



    @Override

    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        // 消息发送到交换机成功或失败时调用这个方法

        log.info("confirm() 回调函数打印 CorrelationData: " + correlationData);

        //这里ack值为true仅仅表示消息成功发送到交换机

        log.info("confirm() 回调函数打印 ack: " + ack);

        log.info("confirm() 回调函数打印 cause: " + cause);

    }



    @Override

    public void returnedMessage(ReturnedMessage returned) {

        // 发送到队列失败时才调用这个方法(不被调用 = 消息成功路由到队列或者消息成功路由到备用队列)

        log.info("消息主体: " + new String(returned.getMessage().getBody()));

        log.info("应答码: " + returned.getReplyCode());

        log.info("描述: " + returned.getReplyText());

        log.info("消息使用的交换器 exchange : " + returned.getExchange());

        log.info("消息使用的路由键 routing : " + returned.getRoutingKey());

    }

}

RabbitMQTest.java：

package org.example.alternate;



import org.junit.jupiter.api.Test;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;



@SpringBootTest

public class RabbitMQTest {

public static final String EXCHANGE_DIRECT = "exchange.direct.order";

    public static final String ROUTING_KEY = "order";



    @Autowired

    private RabbitTemplate rabbitTemplate;



    @Test

    public void test01SendMessage() {

        //正常发送

        //rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "send test confirmProducer");

        //故意改错交换机然后发送

        //rabbitTemplate.convertAndSend(EXCHANGE_DIRECT+"error", ROUTING_KEY, "send test confirmProducer");

        //故意改错队列(让其匹配不到队列)然后发送，结果就是可能进入到备份队列

        rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY+"error", "send test confirmProducer");

    }

}

然后运行test01SendMessage方法，然后在rabbitmq后台管理中看到备份队列Queue queue.test.backup有数据了

因为这个备份队列暂时没有消费者，所以不会被消费，所以会在备份队列中看到数据，若有消费者则另说。