SpringAMQP 消息適配器 - MessageListenerAdapter

MessageListenerAdapter 即消息監聽適配器

這一節，咱們先寫一下代碼，再總結 MessageListenerAdapter 的使用

代碼示例：

代碼地址：  https://github.com/hmilyos/rabbitmqdemo.git  rabbitmq-api 項目下
複製代碼

1.簡單使用默認方法

修改上一節 SpringAMQP 消息容器 - SimpleMessageListenerContainer 的 RabbitMQConfig 的 messageContainer 方法
複製代碼

@Bean   //connectionFactory 也是要和最上面方法名保持一致
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(queue001(), queue002(), queue003());    //監聽的隊列
        container.setConcurrentConsumers(1);    //當前的消費者數量
        container.setMaxConcurrentConsumers(5); //  最大的消費者數量
        container.setDefaultRequeueRejected(false); //是否重回隊列
        container.setAcknowledgeMode(AcknowledgeMode.AUTO); //簽收模式
        container.setExposeListenerChannel(true);
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {    //消費端的標籤策略
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });

//        1.1 適配器方式. 默認是有本身的方法名字的：handleMessage
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
    	container.setMessageListener(adapter);
        return container;
     }
複製代碼

public class MessageDelegate {

    private static final Logger log = LoggerFactory.getLogger(MessageDelegate.class);

    //這個handleMessage方法名要根據org.springframework.amqp.rabbit.listener.adapter包下的
    //      MessageListenerAdapter.ORIGINAL_DEFAULT_LISTENER_METHOD的默認值來肯定
    public void handleMessage(byte[] messageBody) {
        log.info("默認方法, 消息內容:" + new String(messageBody));
    }
}
複製代碼

這個 handleMessage 方法名要根據 org.springframework.amqp.rabbit.listener.adapter 包下的 MessageListenerAdapter.ORIGINAL_DEFAULT_LISTENER_METHOD 的默認值來肯定，源碼以下

運行以前的測試用例 testSendMessage ，handleMessage 方法進行消息的消費

2.採用本身指定一個方法的名字

將上面的 messageContainer 修改爲以下的

@Bean   //connectionFactory 也是要和最上面方法名保持一致
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(queue001(), queue002(), queue003());    //監聽的隊列
        container.setConcurrentConsumers(1);    //當前的消費者數量
        container.setMaxConcurrentConsumers(5); //  最大的消費者數量
        container.setDefaultRequeueRejected(false); //是否重回隊列
        container.setAcknowledgeMode(AcknowledgeMode.AUTO); //簽收模式
        container.setExposeListenerChannel(true);
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {    //消費端的標籤策略
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });
                //1.2 適配器方式. 能夠本身指定一個方法的名字: consumeMessage
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");
    	container.setMessageListener(adapter);
        return container;
     }
複製代碼

MessageDelegate 裏面的消費方法改爲 consumeMessage

public void consumeMessage(byte[] messageBody) {
        log.info("字節數組方法, 消息內容:" + new String(messageBody));
    }
複製代碼

繼續運行 testSendMessage， 查看到消費

3. 添加一個轉換器，從字節數組轉換爲 String

//1.3 適配器方式.也能夠添加一個轉換器: 從字節數組轉換爲String
public class TextMessageConverter implements MessageConverter {

    @Override
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        return new Message(object.toString().getBytes(), messageProperties);
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        String contentType = message.getMessageProperties().getContentType();
        if(null != contentType && contentType.contains("text")) {
            return new String(message.getBody());
        }
        return message.getBody();
    }

}
複製代碼

toMessage 就是 Java 對象轉換爲 Message，fromMessage 就是 Message 轉爲 Java 對象

將上面的 messageContainer 修改爲以下的

@Bean   //connectionFactory 也是要和最上面方法名保持一致
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(queue001(), queue002(), queue003());    //監聽的隊列
        container.setConcurrentConsumers(1);    //當前的消費者數量
        container.setMaxConcurrentConsumers(5); //  最大的消費者數量
        container.setDefaultRequeueRejected(false); //是否重回隊列
        container.setAcknowledgeMode(AcknowledgeMode.AUTO); //簽收模式
        container.setExposeListenerChannel(true);
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {    //消費端的標籤策略
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });
              //1.3 適配器方式.也能夠添加一個轉換器: 從字節數組轉換爲String
    	MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
    	adapter.setDefaultListenerMethod("consumeMessage");
    	adapter.setMessageConverter(new TextMessageConverter());
    	container.setMessageListener(adapter);
        return container;
     }
複製代碼

關鍵點，這裏使用的再也不是字節數組了！！

//1.3 適配器方式.也能夠添加一個轉換器: 從字節數組轉換爲String
    public void consumeMessage(String messageBody) {
        log.info("字符串方法, 消息內容:" + messageBody);
    }
複製代碼

寫個單元測試用例，注意 contentType 要包含 text ！！

@Test
    public void testSendMessage4Text() throws Exception {
        //1 建立消息
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        Message message = new Message("mq 消息1234".getBytes(), messageProperties);

        rabbitTemplate.send("topic001", "spring.abc", message);
    }
複製代碼

運行單元測試

4. 隊列名稱 和 方法名稱 也能夠進行一一的匹配

將上面的 messageContainer 修改爲以下的

@Bean   //connectionFactory 也是要和最上面方法名保持一致
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(queue001(), queue002(), queue003());    //監聽的隊列
        container.setConcurrentConsumers(1);    //當前的消費者數量
        container.setMaxConcurrentConsumers(5); //  最大的消費者數量
        container.setDefaultRequeueRejected(false); //是否重回隊列
        container.setAcknowledgeMode(AcknowledgeMode.AUTO); //簽收模式
        container.setExposeListenerChannel(true);
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {    //消費端的標籤策略
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });
        //        2 適配器方式: 咱們的隊列名稱 和 方法名稱 也能夠進行一一的匹配
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
    	adapter.setMessageConverter(new TextMessageConverter());
    	Map<String, String> queueOrTagToMethodName = new HashMap<>();
    	queueOrTagToMethodName.put("queue001", "method1");
    	queueOrTagToMethodName.put("queue002", "method2");
    	adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
    	container.setMessageListener(adapter);
        return container;
     }
複製代碼

//        2 適配器方式: 咱們的隊列名稱 和 方法名稱 也能夠進行一一的匹配
    public void method1(String messageBody) {
        log.info("method1 收到消息內容:" + new String(messageBody));
    }
    public void method2(String messageBody) {
        log.info("method2 收到消息內容:" + new String(messageBody));
    }
複製代碼

看一下以前創建的綁定關係

修改一下單元測試用例

@Test
    public void testSendMessage4Text() throws Exception {
        //1 建立消息
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        Message message = new Message("mq 消息1234".getBytes(), messageProperties);

        rabbitTemplate.send("topic001", "spring.abc", message);
        rabbitTemplate.send("topic002", "rabbit.abc", message);
    }
複製代碼

運行測試， 查看到兩個隊列的消費

綜上，經過上面 MessageListenerAdapter 的使用代碼，咱們能夠看出它有以下核心屬性

• defaultListenerMethod 默認監聽方法名稱:用於設置監聽方法的名稱

• delegate 委派對象: 實際真實的委派對象，用於處理消息

• queueOrTagMethodName 隊列標識於方法名稱組成的集合。

• 能夠一一進行隊列於方法名稱的匹配。

• 隊列和方法名稱綁定，即指定隊列裏的消息會被綁定的方法所接受處理。