說明

上一篇文章裡，我們瞭解瞭如何保證訊息被可靠投遞到RabbitMQ的交換機中，但還有一些不完美的地方，試想一下，如果向RabbitMQ伺服器傳送一條訊息，伺服器確實也接收到了這條訊息，於是給你返回了ACK確認訊息，但伺服器拿到這條訊息一看，找不到路由它的佇列，於是就把它丟進了垃圾桶，emmm，我猜應該屬於可回收垃圾。

如何讓訊息可靠投遞到佇列

如果你對上面的描述還不是很清楚，那我再用程式碼來說明一次。

在僅開啟了生產者確認機制的情況下，交換機接收到訊息後，會直接給訊息生產者傳送確認訊息，如果發現該訊息不可路由，那麼訊息會被直接丟棄，此時，生產者是不知道訊息被丟棄這個事件的。

我們將上一篇中的交換機型別改為DirectExchange，這樣就只有當訊息的 RoutingKey 和佇列繫結時設定的 Bindingkey （這裡即“key”）一致時，才會真正將該訊息進行路由。

public static final String BUSINESS_EXCHANGE_NAME = "rabbitmq.tx.demo.simple.business.exchange";
public static final String BUSINESS_QUEUEA_NAME = "rabbitmq.tx.demo.simple.business.queue";

// 宣告業務 Exchange
@Bean("businessExchange")
public DirectExchange businessExchange(){
    return new DirectExchange(BUSINESS_EXCHANGE_NAME);
}

// 宣告業務佇列
@Bean("businessQueue")
public Queue businessQueue(){
    return QueueBuilder.durable(BUSINESS_QUEUEA_NAME).build();
}

// 宣告業務佇列繫結關係
@Bean
public Binding businessBinding(@Qualifier("businessQueue") Queue queue,
                               @Qualifier("businessExchange") DirectExchange exchange){
    return BindingBuilder.bind(queue).to(exchange).with("key");
}
 

對訊息生產者也稍作修改：

@Autowired
private RabbitTemplate rabbitTemplate;

@PostConstruct
private void init() {
    //        rabbitTemplate.setChannelTransacted(true);
    rabbitTemplate.setConfirmCallback(this);
}

public void sendCustomMsg(String exchange, String msg) {
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

    log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

    rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

    correlationData = new CorrelationData(UUID.randomUUID().toString());

    log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

    rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
}

@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
    String id = correlationData != null ? correlationData.getId() : "";
    if (b) {
        log.info("訊息確認成功, id:{}", id);
    } else {
        log.error("訊息未成功投遞, id:{}, cause:{}", id, s);
    }
}
 

然後我們呼叫該方法，傳送兩條訊息測試一下：

訊息id:ba6bf502-9381-4220-8dc9-313d6a289a4e, msg:1
訊息id:f0040a41-dc02-4e45-b8af-e3cfa8a118b2, msg:1
訊息確認成功, id:ba6bf502-9381-4220-8dc9-313d6a289a4e
訊息確認成功, id:f0040a41-dc02-4e45-b8af-e3cfa8a118b2
收到業務訊息：1

可以看到，傳送了兩條訊息，第一條訊息的 RoutingKey 為 “key”，第二條訊息的 RoutingKey 為 “key2”，兩條訊息都成功被交換機接收，也收到了交換機的確認回撥，但消費者只收到了一條訊息，因為第二條訊息的 RoutingKey 與佇列的 BindingKey 不一致，也沒有其它佇列能接收這個訊息，所有第二條訊息被直接丟棄了。

那麼，如何讓訊息被路由到佇列後再返回ACK呢？或者無法被路由的訊息幫我想辦法處理一下？最起碼通知我一聲，我好自己處理啊。

別慌別慌，RabbitMQ裡有兩個機制剛好可以解決我們上面的疑問：

1、mandatory 引數 2、備份交換機

mandatory 引數

設定 mandatory 引數可以在當訊息傳遞過程中不可達目的地時將訊息返回給生產者。

當把 mandotory 引數設定為 true 時，如果交換機無法將訊息進行路由時，會將該訊息返回給生產者，而如果該引數設定為false，如果發現訊息無法進行路由，則直接丟棄。

那麼如何設定這個引數呢？在傳送訊息的時候，只需要在初始化方法新增一行程式碼即可：

rabbitTemplate.setMandatory(true);

開啟之後我們再重新執行前面的程式碼：

訊息id:19729f33-15c4-4c1b-8d48-044c301e2a8e, msg:1
訊息id:4aea5c57-3e71-4a7b-8a00-1595d2b568eb, msg:1
訊息確認成功, id:19729f33-15c4-4c1b-8d48-044c301e2a8e
Returned message but no callback available
訊息確認成功, id:4aea5c57-3e71-4a7b-8a00-1595d2b568eb
收到業務訊息：1

我們看到中間多了一行提示 Returned message but no callback available 這是什麼意思呢？

我們上面提到，設定 mandatory 引數後，如果訊息無法被路由，則會返回給生產者，是通過回撥的方式進行的，所以，生產者需要設定相應的回撥函式才能接受該訊息。

為了進行回撥，我們需要實現一個介面 RabbitTemplate.ReturnCallback。

@Slf4j
@Component
public class BusinessMsgProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    private void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    public void sendCustomMsg(String exchange, String msg) {
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

        correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (b) {
            log.info("訊息確認成功, id:{}", id);
        } else {
            log.error("訊息未成功投遞, id:{}, cause:{}", id, s);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("訊息被伺服器退回。msg:{}, replyCode:{}. replyText:{}, exchange:{}, routingKey :{}",
                new String(message.getBody()), replyCode, replyText, exchange, routingKey);
    }
}

然後我們再來重新執行一次：

訊息id:2e5c336a-883a-474e-b40e-b6e3499088ef, msg:1
訊息id:85c771cb-c88f-47dd-adea-f0da57138423, msg:1
訊息確認成功, id:2e5c336a-883a-474e-b40e-b6e3499088ef
訊息無法被路由，被伺服器退回。msg:1, replyCode:312. replyText:NO_ROUTE, exchange:rabbitmq.tx.demo.simple.business.exchange, routingKey :key2
訊息確認成功, id:85c771cb-c88f-47dd-adea-f0da57138423
收到業務訊息：1

可以看到，我們接收到了被退回的訊息，並帶上了訊息被退回的原因：NO_ROUTE。但是要注意的是， mandatory 引數僅僅是在當訊息無法被路由的時候，讓生產者可以感知到這一點，只要開啟了生產者確認機制，無論是否設定了 mandatory 引數，都會在交換機接收到訊息時進行訊息確認回撥，而且通常訊息的退回回撥會在訊息的確認回撥之前。

備份交換機

有了 mandatory 引數，我們獲得了對無法投遞訊息的感知能力，有機會在生產者的訊息無法被投遞時發現並處理。但有時候，我們並不知道該如何處理這些無法路由的訊息，最多打個日誌，然後觸發報警，再來手動處理。而通過日誌來處理這些無法路由的訊息是很不優雅的做法，特別是當生產者所在的服務有多臺機器的時候，手動複製日誌會更加麻煩而且容易出錯。

而且設定 mandatory 引數會增加生產者的複雜性，需要新增處理這些被退回的訊息的邏輯。如果既不想丟失訊息，又不想增加生產者的複雜性，該怎麼做呢？

前面在設定死信佇列的文章中，我們提到，可以為佇列設定死信交換機來儲存那些處理失敗的訊息，可是這些不可路由訊息根本沒有機會進入到佇列，因此無法使用死信佇列來儲存訊息。

不要慌，在 RabbitMQ 中，有一種備份交換機的機制存在，可以很好的應對這個問題。

什麼是備份交換機呢？備份交換機可以理解為 RabbitMQ 中交換機的“備胎”，當我們為某一個交換機宣告一個對應的備份交換機時，就是為它建立一個備胎，當交換機接收到一條不可路由訊息時，將會將這條訊息轉發到備份交換機中，由備份交換機來進行轉發和處理，通常備份交換機的型別為 Fanout ，這樣就能把所有訊息都投遞到與其繫結的佇列中，然後我們在備份交換機下繫結一個佇列，這樣所有那些原交換機無法被路由的訊息，就會都進入這個隊列了。當然，我們還可以建立一個報警佇列，用獨立的消費者來進行監測和報警。

聽的不太明白？沒關係，看個圖就知道是怎麼回事了。

（emmm，調整了一下配色，感覺還是很醜- - 。急需一個UI來拯救我。）

接下來，我們就來設定一下備份交換機：

@Configuration
public class RabbitMQConfig {

    public static final String BUSINESS_EXCHANGE_NAME = "rabbitmq.backup.test.exchange";
    public static final String BUSINESS_QUEUE_NAME = "rabbitmq.backup.test.queue";
    public static final String BUSINESS_BACKUP_EXCHANGE_NAME = "rabbitmq.backup.test.backup-exchange";
    public static final String BUSINESS_BACKUP_QUEUE_NAME = "rabbitmq.backup.test.backup-queue";
    public static final String BUSINESS_BACKUP_WARNING_QUEUE_NAME = "rabbitmq.backup.test.backup-warning-queue";

    // 宣告業務 Exchange
    @Bean("businessExchange")
    public DirectExchange businessExchange(){
        ExchangeBuilder exchangeBuilder = ExchangeBuilder.directExchange(BUSINESS_EXCHANGE_NAME)
                .durable(true)
                .withArgument("alternate-exchange", BUSINESS_BACKUP_EXCHANGE_NAME);

        return (DirectExchange)exchangeBuilder.build();
    }

    // 宣告備份 Exchange
    @Bean("backupExchange")
    public FanoutExchange backupExchange(){
        ExchangeBuilder exchangeBuilder = ExchangeBuilder.fanoutExchange(BUSINESS_BACKUP_EXCHANGE_NAME)
                .durable(true);
        return (FanoutExchange)exchangeBuilder.build();
    }

    // 宣告業務佇列
    @Bean("businessQueue")
    public Queue businessQueue(){
        return QueueBuilder.durable(BUSINESS_QUEUE_NAME).build();
    }

    // 宣告業務佇列繫結關係
    @Bean
    public Binding businessBinding(@Qualifier("businessQueue") Queue queue,
                                    @Qualifier("businessExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("key");
    }

    // 宣告備份佇列
    @Bean("backupQueue")
    public Queue backupQueue(){
        return QueueBuilder.durable(BUSINESS_BACKUP_QUEUE_NAME).build();
    }

    // 宣告報警佇列
    @Bean("warningQueue")
    public Queue warningQueue(){
        return QueueBuilder.durable(BUSINESS_BACKUP_WARNING_QUEUE_NAME).build();
    }

    // 宣告備份佇列繫結關係
    @Bean
    public Binding backupBinding(@Qualifier("backupQueue") Queue queue,
                                   @Qualifier("backupExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

    // 宣告備份報警佇列繫結關係
    @Bean
    public Binding backupWarningBinding(@Qualifier("warningQueue") Queue queue,
                                 @Qualifier("backupExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

}

這裡我們使用 ExchangeBuilder 來建立交換機，併為其設定備份交換機：

 .withArgument("alternate-exchange", BUSINESS_BACKUP_EXCHANGE_NAME);

為業務交換機綁定了一個佇列，為備份交換機綁定了兩個佇列，一個用來儲存不可投遞訊息，待之後人工處理，一個專門用來做報警用途。

接下來，分別為業務交換機和備份交換機建立消費者：

@Slf4j
@Component
public class BusinessMsgConsumer {

    @RabbitListener(queues = BUSINESS_QUEUE_NAME)
    public void receiveMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("收到業務訊息：{}", msg);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    }
}

@Slf4j
@Component
public class BusinessWaringConsumer {

    @RabbitListener(queues = BUSINESS_BACKUP_WARNING_QUEUE_NAME)
    public void receiveMsg(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.error("發現不可路由訊息：{}", msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

接下來我們分別傳送一條可路由訊息和不可路由訊息：

@Slf4j
@Component
public class BusinessMsgProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendCustomMsg(String exchange, String msg) {


        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());


        log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

        correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
    }
}

訊息如下：

訊息id:5c3a33c9-0764-4d1f-bf6a-a00d771dccb4, msg:1
訊息id:42ac8c35-1d0a-4413-a1df-c26a85435354, msg:1
收到業務訊息：1
發現不可路由訊息：1

這裡僅僅使用 error 日誌配合日誌系統進行報警，如果是敏感資料，可以使用郵件、釘釘、簡訊、電話等報警方式來提高時效性。

那麼問題來了，mandatory 引數與備份交換機可以一起使用嗎？設定 mandatory 引數會讓交換機將不可路由訊息退回給生產者，而備份交換機會讓交換機將不可路由訊息轉發給它，那麼如果兩者同時開啟，訊息究竟何去何從？？

emmm，想這麼多幹嘛，試試不就知道了。

修改一下生產者即可：

@Slf4j
@Component
public class BusinessMsgProducer implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    private void init() {
//        rabbitTemplate.setChannelTransacted(true);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    public void sendCustomMsg(String exchange, String msg) {


        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());


        log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key", msg, correlationData);

        correlationData = new CorrelationData(UUID.randomUUID().toString());

        log.info("訊息id:{}, msg:{}", correlationData.getId(), msg);

        rabbitTemplate.convertAndSend(exchange, "key2", msg, correlationData);
    }


    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        String id = correlationData != null ? correlationData.getId() : "";
        if (b) {
            log.info("訊息確認成功, id:{}", id);
        } else {
            log.error("訊息未成功投遞, id:{}, cause:{}", id, s);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("訊息被伺服器退回。msg:{}, replyCode:{}. replyText:{}, exchange:{}, routingKey :{}",
                new String(message.getBody()), replyCode, replyText, exchange, routingKey);
    }
}

再來測試一下：

訊息id:0a3eca1e-d937-418c-a7ce-bfb8ce25fdd4, msg:1
訊息id:d8c9e010-e120-46da-a42e-1ba21026ff06, msg:1
訊息確認成功, id:0a3eca1e-d937-418c-a7ce-bfb8ce25fdd4
訊息確認成功, id:d8c9e010-e120-46da-a42e-1ba21026ff06
發現不可路由訊息：1
收到業務訊息：1

可以看到，兩條訊息都可以收到確認成功回撥，但是不可路由訊息不會被回退給生產者，而是直接轉發給備份交換機。可見備份交換機的處理優先順序更高。

總結

上一篇中，我們介紹了事務機制和生產者確認機制來確保訊息的可靠投遞，相對而言，生產者確認機制更加高效和靈活。本篇中，我們介紹了另外兩種確保生產者的訊息不丟失的機制，即通過 mandatory 引數和備份交換機來處理不可路由訊息。

通過以上幾種機制，我們總算是可以確保訊息被萬無一失的投遞到目的地了。到此，我們的訊息可靠投遞也就告一段落了。訊息可靠投遞是我們使用MQ時無法逃避的話題，一次性搞定它，就不會再為其所困。總的來說，方法總比問題多，但如果你不知道這些方法，那麼當問題來臨時，也許就會不知所措了。

相信通過這幾篇關於 RabbitMQ 文章的學習，對於 RabbitMQ 的理解已經突破天際，那還在等什麼，趕緊把接入 RabbitMQ 的專案好好優化一下吧，相信現在你就不會再被那些不知所云的配置和程式碼所迷惑了。

到此為止，本篇就完美落幕了，希望能給你帶來一些啟發，也歡迎關注我的公眾號進行留言交流。

相關推薦