圖文并茂講解RocketMQ消息類別

更新時間：2022年12月27日 15:35:42 作者：一個雙子座的Java攻城獅

這篇文章主要介紹了圖文并茂講解RocketMQ消息類別，RocketMQ對于消息提供了很多用法，包括：同步消息、異步消息、單向發(fā)送、順序消息、延時消息、批量消息、過濾消息、事務(wù)消息等

1、同步消息

即時性較強，重要的消息，且必須有回執(zhí)的消息，例如短信，通知（轉(zhuǎn)賬成功）

生產(chǎn)者：

public class Producer {
    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer=new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.23.127:9876");
        producer.start();
        for (int i = 1; i <= 5; i++) {
            Message msg = new Message("topic2",("同步消息：hello rocketmq "+i).getBytes("UTF-8"));
            //同步消息發(fā)送
            SendResult result = producer.send(msg);
            System.out.println("返回結(jié)果："+result);
        }
        producer.shutdown();
    }
}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.23.127:9876");
        consumer.subscribe("topic2","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    //System.out.println("收到消息："+msg);
                    System.out.println("消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功處理, mq 收到這個 標(biāo)記后相同的消息講不會再次發(fā)給消費者
            }
        });
        consumer.start();// 開啟多線程 監(jiān)控消息,持續(xù)運行
        System.out.println("接收消息服務(wù)已運行");
    }
}

測試：

2、異步消息

即時性較弱，但需要有回執(zhí)的消息，例如訂單中的某些信息

生產(chǎn)者：

public class Producer {
    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer=new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.23.127:9876");
        producer.start();
        for (int i = 1; i <= 5; i++) {
            //異步消息發(fā)送
            Message msg = new Message("topic2",("異步消息：hello rocketmq "+i).getBytes("UTF-8"));
            producer.send(msg, new SendCallback() {
                //表示成功返回結(jié)果
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);
                }
                //表示發(fā)送消息失敗
                @Override
                public void onException(Throwable throwable) {
                    System.out.println(throwable);
                }
            });
        }
        //添加一個休眠操作，確保異步消息返回后能夠輸出
        TimeUnit.SECONDS.sleep(10);
        producer.shutdown();
    }
}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.23.127:9876");
        consumer.subscribe("topic2","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    //System.out.println("收到消息："+msg);
                    System.out.println("消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功處理, mq 收到這個 標(biāo)記后相同的消息講不會再次發(fā)給消費者
            }
        });
        consumer.start();// 開啟多線程 監(jiān)控消息,持續(xù)運行
        System.out.println("接收消息服務(wù)已運行");
    }
}

測試：

3、單向消息

不需要有回執(zhí)的消息，例如日志類消息

生產(chǎn)者：

public class Producer {
    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer=new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.23.127:9876");
        producer.start();
        for (int i = 1; i <= 5; i++) {
            //單向消息
            Message msg = new Message("topic2",("單向消息：hello rocketmq "+i).getBytes("UTF-8"));
            producer.sendOneway(msg);
        }
        //添加一個休眠操作，確保異步消息返回后能夠輸出
        TimeUnit.SECONDS.sleep(10);
        producer.shutdown();
    }
}

消費者代碼同上

測試：

總結(jié) 同步消息

SendResult result = producer.send(msg);

異步消息（回調(diào)處理結(jié)果必須在生產(chǎn)者進(jìn)程結(jié)束前執(zhí)行，否則回調(diào)無法正確執(zhí)行）

		producer.send(msg, new SendCallback() {
                //表示成功返回結(jié)果
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);
                }
                //表示發(fā)送消息失敗
                @Override
                public void onException(Throwable throwable) {
                    System.out.println(throwable);
                }
            });

單向消息