MessageListenerAdapter 即消息監聽適配器
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> queueOrTagToMethodName = new HashMap<>();
queueOrTagToMethodName.put("queue001", "method1");
queueOrTagToMethodName.put("queue002", "method2");
adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
container.setMessageListener(adapter);
return container;
}
/<string>
// 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 隊列標識於方法名稱組成的集合。
- 可以一一進行隊列於方法名稱的匹配。
- 隊列和方法名稱綁定，即指定隊列裡的消息會被綁定的方法所接受處理。
關注我：私信回覆“555”獲取往期Java高級架構資料、源碼、筆記、視頻Dubbo、Redis、Netty、zookeeper、Spring cloud、分佈式、高併發等架構技術往期架構視頻截圖
分享到:
閱讀更多 來一杯82年的Java 的文章
