RabbitMQ系列（20）--RabbitMQ备份交换机

news2024/11/20 1:21:12

前言：上一篇文章我们提到当交换机确认消息失败或者交换机发送消息到队列失败，都可以通过回调方法让生产者重新发送消息，除此之外另一种方法就是通过备份交换机的方式保证消息的不丢失，当生产者无法把消息投递给交换机，就通过交换机把消息发送到备份交换机，再让备份交换机通过自己的路由以及自己的队列发送消息给消费者，从而把发送失败的消息保存下来。

1、架构图如下

2、修改ConfirmConfig类的代码

代码如下：

package com.ken.springbootrqbbitmq.config;

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 EXCHANGE_NAME = "confirm_exchange";

    //队列
    public static final String QUEUE_NAME = "confirm_queue";

    //routingkey
    public static final String ROUTING_KEY = "confirm";

    //备份交换机
    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";

    /**
     * 声明交换机
     * durable(true)表示队列持久化
     * withArgument用于写入参数，这里的参数alternate-exchange表示备份交换机
     * @return
     */
    @Bean("confirmExchange")
    public DirectExchange confirmExchange() {
        return ExchangeBuilder.directExchange(EXCHANGE_NAME).durable(true)
                .withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
    }

    //声明队列
    @Bean("confirmQueue")
    public Queue confirmQueue() {
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    //绑定交换机和队列
    @Bean
    public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
                                        @Qualifier("confirmExchange") DirectExchange confirmExchange) {
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(ROUTING_KEY);
    }

    //声明备份交换机
    @Bean("backupExchange")
    public FanoutExchange backupExchange() {
        return new FanoutExchange(BACKUP_EXCHANGE_NAME);
    }

    //声明备份队列
    @Bean("backupQueue")
    public Queue backupQueue() {
        return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
    }

    //声明报警队列
    @Bean("warningQueue")
    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 warningQueue,
                                                    @Qualifier("backupExchange") FanoutExchange backupExchange) {
        return BindingBuilder.bind(warningQueue).to(backupExchange);
    }


}

3、在consumer包里新建一个名为WarningConsumer的类用于编写充当消费者消费消息的代码

代码如下：

package com.ken.springbootrqbbitmq.consumer;

import com.ken.springbootrqbbitmq.config.ConfirmConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 报警消费者
 */
@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);
    }
}

4、因为我们修改了交换机的代码，所以我们需要删除旧的confirm_exchange交换机，然后再启动项目重新生成confirm_exchange交换机