接收消息

消息接收步驟:

1. 創建消息消費者, 指定消費者所屬的組名

2. 指定Nameserver地址

3. 指定消費者訂閱的主題和標簽

4. 設置回調函數，編寫處理消息的方

5. 啟動消息消費者

package com.leon.test;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class RocketMQReceiveMessageTest {//接收消息public static void main(String[] args) throws Exception {//1 創建消費者,并且為其指定消費者組名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");//2 為消費者設置NameServer的地址consumer.setNamesrvAddr("192.168.109.131:9876");//3 指定消費者訂閱的主題和標簽consumer.subscribe("myTopic", "*");//4 設置一個回調函數,并在函數中編寫接收到消息之后的處理方法consumer.registerMessageListener(new MessageListenerConcurrently() {//處理獲取到的消息@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {//消費邏輯System.out.println("Message===>" + list);//返回消費成功狀態return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//5 啟動消費者consumer.start();System.out.println("啟動消費者成功了");} }

?

總結

以上是生活随笔為你收集整理的Java实现消息消费的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。