springboot+rabbitmq實現指定消費者才能消費

程序里有一個應用場景使用到了rabbitmq——當財務確認收到企業的打款金額后，系統會把企業訂單生成用戶付款單。由於訂單記錄數據量大，改為通過mq來異步實現。即財務確認收款操作后，將企業訂單數據放入mq，另一端監聽mq消息隊列，將收到的企業訂單加工轉換成用戶付款單，並做持久化。

 

 

本地開發環境與測試環境共用一套rabbitmq。當項目部署到測試環境后，QA測試過程中，總是“莫名其妙”的發現所保存的用戶付款單數據有問題。

當然，首先要排查程序，檢查Consumer的數據處理的邏輯是否有bug。單元測試后，發現並不存在測試環境的bug。

 

原來，消息隊列被“非正常”消費了！

Q: 什么情況？

A: 幾個伙伴一起參與的項目，大家總是要調試自己的程序的。而如果碰巧本地程序監聽到消息隊列里有消息，那么，消息就被本地程序消費掉了。問題正是出現在這里！————團隊開發，大家並不會及時檢出git上最新的程序版本。如果本地的程序版本不是最新的正確的版本，勢必會出現bug。

 

那么，怎么辦？

每次你改了邏輯，告訴大家獲取最新？

不現實，約定的東西往往不奏效的。

 

如何保證mq隊列里的消息只被測試服務器上的consumer消費，避免本地環境誤消費？ 或者說，如何實現消息的定向消費呢？

只要肯琢磨，辦法總比困難多！百思可得解！

 

我們知道，rabbitmq手動ack模式。這還不夠，因為我們怎么讓consumer來決定是否消費呢？ 所以，我們需要一個標識————producer設定一個標識，consumer如果匹配這個標識，則消費，否則予以reject放回消息隊列。

 

 

通過查看spring-rabbit/spring-amqp的代碼，發現可以在spring-amqp里的MessageProperties上做文章。生產者與消費者每次消息傳輸都會攜帶一個MessageProperties，通常我們是不指定的，走MessageProperties的默認設置值。

 

我的策略：MessageProperties有一個屬性叫AppId。我們程序所部署的測試機器就一台，即消息Producer和消息Consumer在一台機器上。那么，我就可以利用機器的IP來識別消息。只有Producer與Consumer的IP匹配，才消費消息。程序員本機IP與測試服務器IP不一樣，就會拒絕接收消息，會把消息重新放回消息隊列，等待測試服務器的Consumer消費。

話不多說，上代碼吧，

生產者代碼：

package com.sboot.mq;

import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import java.net.InetAddress;
import java.util.UUID;

public class MQProducerTest extends BaseTest {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    public void test() throws Exception {
        for (int i = 1; i <= 5; i++) {
            MessageProperties messageProperties = new MessageProperties();
            String ip = InetAddress.getLocalHost().getHostAddress();
            messageProperties.setAppId(ip);
//            messageProperties.setUserId(String.valueOf(i));
            MessageConverter messageConverter = new SimpleMessageConverter();
            String msg = UUID.randomUUID().toString();
//            System.out.println(msg);
            Message message1 = messageConverter.toMessage(msg, messageProperties);
            rabbitTemplate.send(MessageQueueConstant.USER_SETTLEMENT_EXCHANGE, "UserSettlementRouting", message1);
            System.out.println("入隊完成");
            Thread.sleep(500L);
        }
    }
}

 

消費者手動ACK，要實現ChannelAwareMessageListener接口，感知rabbitmq.client.Channel實例，調用channel的basicAck、basicReject等方法：

package com.sboot.mq;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Component;

import java.net.InetAddress;

@Component
@Profile(value = "dev")
@Slf4j
public class UserSettlementDevConsumer implements ChannelAwareMessageListener {

    @RabbitHandler
    @RabbitListener(queues = MessageQueueConstant.USER_SETTLEMENT_QUEUE, ackMode = "MANUAL")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        Thread.currentThread().setName(UserSettlementDevConsumer.class.getSimpleName() + System.currentTimeMillis());

        long tag = message.getMessageProperties().getDeliveryTag();
        String appId = message.getMessageProperties().getAppId();
        log.info("{}-{}, 消息出隊", tag, appId);
        String receiveMsg = "";
        try {
            //核對標識，決定是否消費消息
            String ip = InetAddress.getLocalHost().getHostAddress();
            if (!ip.equals(appId)) {
                log.info("這不是我需要的消息。放回隊列。{}", receiveMsg);
//                channel.basicNack(tag, false, true);
                channel.basicReject(tag, true);
//                channel.basicRecover(true);
                return;
            }

            MessageConverter messageConverter = new SimpleMessageConverter();
            receiveMsg = String.valueOf(messageConverter.fromMessage(message));
            。。。。在這里消費消息
            log.info("success " + receiveMsg);
            channel.basicAck(tag, false);

        } catch (Exception e) {
            log.error("receive message has an error, ", e);
            channel.basicNack(tag, false, true);
        }
    }

}

 

說明一下依賴的spring-rabbit包的版本，我的是2.2.0.RELEASE。如果是2.1.4版本里，@RabbitListener注解沒有ackMode。

 

 

---

解決本案問題過程中的花絮：

 spring-rabbit-2.1.4.RELEASE

 spring-rabbit-2.2.0.RELEASE

 

@RabbitListener的ackMode的值見枚舉org.springframework.amqp.core.AcknowledgeMode

• NONE-- no acks（自動消費 autoAck）
• MANUAL --Manual acks - user must ack/nack via a channel aware listener.（手動消費，Consumer端必須顯式調用ack或nack）
• AUTO --

設置了手動消費，上文消費端的deliveryTag會是不同的long值。自動消費的deliveryTag是重復的1和2這樣的。並且，自動消費時，如果要使用channel的ack/nack，會報異常：

2020-06-19 22:26:54.586 [AMQP Connection 192.168.40.20:5672] ERROR o.s.a.rabbit.connection.CachingConnectionFactory:1468 - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
2020-06-19 22:26:54.599 [SimpleAsyncTaskExecutor-1] ERROR c.e.z.r.p.modules.mq.UserSettlementAckConsumer:49 - 
org.springframework.amqp.AmqpException: PublisherCallbackChannel is closed
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$CachedChannelInvocationHandler.invoke(CachingConnectionFactory.java:1092)