RabbitMQ之备用交换机

作用：当主交换器无法路由消息时，那么消息将转发到备用交换器，由备用交换机处理。

spring:
  rabbitmq:
    port: 5672
    host: 127.0.0.1
    username: guest
    password: guest

package com.yzm.rabbitmq_07.config;

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

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitConfig {

    // 普通交换机
    public static final String ORDINARY_EXCHANGE = "ordinary-exchange";
    public static final String ORDINARY_QUEUE = "ordinary-queue";
    // 备用交换机
    public static final String ALTERNATE_EXCHANGE = "alternate-exchange";
    public static final String ALTERNATE_QUEUE = "alternate-queue";

    @Bean
    public Queue alQueue() {
        return QueueBuilder.durable(ALTERNATE_QUEUE).build();
    }

    // 备用交换器一般都是设置FANOUT模式
    @Bean
    public FanoutExchange alExchange() {
        return ExchangeBuilder.fanoutExchange(ALTERNATE_EXCHANGE).build();
    }

    @Bean
    public Binding alBinding() {
        return BindingBuilder.bind(alQueue()).to(alExchange());
    }

    //-----------------------------------------------------------------------------------------

    @Bean
    public Queue orQueue() {
        return QueueBuilder.durable(ORDINARY_QUEUE).build();
    }

    // 普通交换机绑定备用交换机
    @Bean
    public DirectExchange orExchange() {
        Map args = new HashMap<>();
        args.put("alternate-exchange", ALTERNATE_EXCHANGE);
        return ExchangeBuilder.directExchange(ORDINARY_EXCHANGE).withArguments(args).build();
    }

    @Bean
    public Binding orBinding() {
        return BindingBuilder.bind(orQueue()).to(orExchange()).with("ordinary.yzm");
    }
}

package com.yzm.rabbitmq_07.sender;

import com.yzm.rabbitmq_07.config.RabbitConfig;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class Sender {

    private final RabbitTemplate rabbitTemplate;

    public Sender(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    @GetMapping("/alternate")
    public void alternate() {
        String[] routs = {"ordinary.yzm", "alternate.yzm","other.yzm"};
        for (int i = 0; i < 10; i++) {
            String routing = routs[i % 3];
            String message = routing + " ! @yzm_" + i;
            rabbitTemplate.convertAndSend(RabbitConfig.ORDINARY_EXCHANGE, routing, message);
            System.out.println(" [ 生产者 ] Sent ==> '" + message + "'");
        }
    }
}

package com.yzm.rabbitmq_07.receiver;

import com.yzm.rabbitmq_07.config.RabbitConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Receiver {

    @RabbitListener(queues = RabbitConfig.ORDINARY_QUEUE)
    public void ordinary(Message message) {
        System.out.println(" [ 消费者@普通号 ] 接收到消息 ==> '" + new String(message.getBody()));
    }

    @RabbitListener(queues = RabbitConfig.ALTERNATE_QUEUE)
    public void alternate(Message message) {
        System.out.println(" [ 消费者@备用号 ] 接收到消息 ==> '" + new String(message.getBody()));
    }
}

启动，运行结果：

主交换机能路由的只有 ordinary.yzm 路由键对应的消息，其他由备用交换机处理

首页
上一篇：延时队列
下一篇：消息回调