RabbitMQ消息路由失敗的處理方案(回調(diào)與備份交換機(jī)AE)

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=futao
spring.rabbitmq.password=123456789
spring.rabbitmq.virtual-host=/tech-sharing

# 當(dāng)exchange無(wú)法找到任何一個(gè)合適的queue時(shí)，將消息return給生產(chǎn)者
spring.rabbitmq.template.mandatory=false
# 必須設(shè)置為true，否則消息消息路由失敗也無(wú)法觸發(fā)Return回調(diào)
spring.rabbitmq.publisher-returns=false

• 注意: 使用備份交換機(jī)模式，mandatory將無(wú)效，即就算mandatory設(shè)置為false，路由失敗的消息同樣會(huì)被投遞到綁定的備份交換機(jī)。
• 正常業(yè)務(wù)交換機(jī)(不綁定隊(duì)列，使得消息一定會(huì)路由失敗)

/**
 * 業(yè)務(wù)交換機(jī)
 *
 * @return */ @Bean public Exchange noMatchQueueExchange() { return ExchangeBuilder
            .topicExchange(NoMatchQueueAlternateExchange.EXCHANGE_NAME)
            .durable(true) // 綁定備份交換機(jī) .alternate(X_ALTERNATE)
            .build();
}

• 備份交換機(jī)/隊(duì)列/綁定

/**
 * 備份隊(duì)列
 *
 * @return */ @Bean public Queue alternateQueue() { return QueueBuilder
            .durable("Q_ALTERNATE")
            .build();
} /**
 * 備份交換機(jī)
 *
 * @return */ @Bean public Exchange alternateExchange() { return ExchangeBuilder
            .fanoutExchange(X_ALTERNATE)
            .durable(true)
            .build();
} /**
 * 備份綁定
 *
 * @param alternateExchange
 * @param alternateQueue
 * @return */ @Bean public Binding alternateBinding(Exchange alternateExchange, Queue alternateQueue) { return BindingBuilder
            .bind(alternateQueue)
            .to(alternateExchange)
            .with("")
            .noargs();
}

• 消息投遞

/**
 * 正常業(yè)務(wù)交換機(jī)
 */ public static final String EXCHANGE_NAME = "X_NO_MATCH_QUEUE_ALTERNATE"; @Autowired private RabbitTemplate rabbitTemplate; /**
 * 發(fā)送消息
 */ @PostConstruct public void send() {
    log.info("發(fā)送消息");
    Order order = new Order(1, BigDecimal.TEN, OrderStatusEnum.UN_PAY.getStatus());
    Message message = MessageBuilder
            .withBody(JSON.toJSONString(order).getBytes(StandardCharsets.UTF_8))
            .setContentEncoding(StandardCharsets.UTF_8.displayName())
            .setContentType(MessageProperties.CONTENT_TYPE_JSON)
            .build();
    rabbitTemplate.convertAndSend(EXCHANGE_NAME, "", message);
}

• 結(jié)果是消息被路由到備份交換機(jī)的備份隊(duì)列

• 且： 如果你同時(shí)使用了兩種方式，即（mandatory為true+Listener監(jiān)聽(tīng)）和（備份交換機(jī)AlternateExchange），消息將只會(huì)路由到備份交換機(jī)，不會(huì)Return回生產(chǎn)者。

@Slf4j public class AeTest { /**
     * 獲取Channel
     */ private static final Channel CHANNEL = MqChannelUtils.getChannel(); /**
     * 備份交換機(jī)
     */ private static final String X_AE = "X_AE"; /**
     * 備份交換機(jī)綁定的隊(duì)列
     */ private static final String Q_AE = "Q_AE"; /**
     * 正常業(yè)務(wù)的交換機(jī)
     */ private static final String X_1 = "X_1"; public static void main(String[] args) throws IOException { // 定義備份交換機(jī)-其實(shí)也是一個(gè)正常的交換機(jī) CHANNEL.exchangeDeclare(X_AE, BuiltinExchangeType.FANOUT, true); // 定義備份隊(duì)列 CHANNEL.queueDeclare(Q_AE, true, false, false, null); // 綁定備份 CHANNEL.queueBind(Q_AE, X_AE, "");

        HashMap arguments = new HashMap<>(); // 綁定的備份交換機(jī) arguments.put("alternate-exchange", X_AE); // 定義交換機(jī) CHANNEL.exchangeDeclare(X_1, BuiltinExchangeType.TOPIC, false, false, arguments); // 添加監(jiān)聽(tīng)器，看看是否還會(huì)return消息 CHANNEL.addReturnListener(new ReturnCallback() { @Override public void handle(Return returnMessage) {
                log.error("消息被退回{}", returnMessage);
            }
        }); // 嘗試向交換機(jī)發(fā)送消息（無(wú)法路由）- mandatory參數(shù)無(wú)效 CHANNEL.basicPublish(X_1, "", false, false, new AMQP.BasicProperties(), "阿依古麗".getBytes(StandardCharsets.UTF_8));
    }
}

• 兩個(gè)交換機(jī)，正常的交換機(jī)X_1和備份交換機(jī)X_AE