RocketMQ<一>API使用

本文轉載自查看原文 2021-06-23 16:01 149 消息中間件

消息隊列的種類有很多，這里以rocketmq為例講一下rocketmq的簡單使用，安裝步驟這里不詳細講述，如果有對消息隊列不太了解的可以先看一下上一篇文章《消息隊列一》。

一、前提

1、需要添加響應的pom依賴，這里不需要整合spring boot，直接使用 rocketmq-client ，【注意】這里pom文件的版本必須要和你安裝的版本一致，否者啟動會報錯。

<dependencies>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.3.0</version>
        </dependency>
</dependencies>

2、這里提示一下新手可能會出現的問題

No route info of this topic：在啟動broker時加上 autoCreateTopicEnable=true，如果是用broker.conf啟動在配置文件中添加，默認自動創建沒有的topic。
Producer沒有正確連接到Name Server：查看是否為防火牆原因關閉防火牆（生產環境不建議），開放rocketmq的相應端口，或者加入相應的可以訪問rocketmq的ip。
connect to <ip:port> failed：rocketmq默認開啟了vip通道 producer.setVipChannelEnabled(false); consumer.setVipChannelEnabled(false);
Send [3] times, still failed, cost [575]ms, Topic: Name, BrokersSent: [broker-a, broker-a, null] ：如果出現開始發送消息成功，消費者接收消息后，再發送消息的話，出現上訴異常，則有很大概率是機器磁盤空間不足。

二、簡單的API使用

1、發送同步消息

這個消息適用於對消息丟失忍受力比較低的,對性能要求沒那么高的

生產者：

/**
 * @author D-L
 * @version 1.0.0
 * @ClassName SyncProducer.java
 * @Description 發送同步消息
 * @createTime 2021-06-22 23:51:00
 */
public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.啟動producer
        producer.start();
        for (int i = 0; i < 5; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數一：消息主題Topic
             * 參數二：消息Tag 用來過濾消息使用
             * 參數三：消息key 用來過濾消息使用
             * 參數四：消息體（內容）
             */
            Message msg = new Message("base", "Tag2", "key2",("同步消息" + i).getBytes());
            //5.發送消息
            SendResult result = producer.send(msg);
            //發送狀態
            SendStatus status = result.getSendStatus();
            System.out.println("發送結果:" + result);
            //線程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //6.關閉生產者producer
        producer.shutdown();
    }
}

消費者：

/**
 * @author D-L
 * @version 1.0.0
 * @ClassName Consumer.java
 * @Description 消息的接受者
 * @createTime 2021-06-22 23:51:00
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group2");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("base", "*");
        //設定消費模式：負載均衡:CLUSTERING  ||  廣播模式:BROADCASTING
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動-----------------------------");
    }
}

2、發送異步消息

生產者：發送異步消息時會有兩個回調，成功和失敗后可以在對應的回調里對發送的消息進行相應的處理。

public class AsyncProducer {

    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.啟動producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            Message msg = new Message("async", "Tag2","key2", ("異步消息" + i).getBytes());
            //5.發送異步消息
            producer.send(msg, new SendCallback() {
                /**
                 * 發送成功回調函數
                 * @param sendResult
                 */
                public void onSuccess(SendResult sendResult) {
                    System.out.println("發送結果：" + sendResult);
                }
                /**
                 * 發送失敗回調函數
                 * @param e
                 */
                public void onException(Throwable e) {
                    System.out.println("發送異常：" + e);
                }
            });
            //線程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //6.關閉生產者producer
        producer.shutdown();
    }
}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group2");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("async", "*");
        //設定消費模式：負載均衡:CLUSTERING  ||  廣播模式:BROADCASTING
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動-----------------------------");
    }
}

3、發送單向消息

發送單向消息，因為是單向消息，所有沒有返回值，也不會阻塞線程，單向發送消息這種方式主要用在不特別關心發送結果的場景，例如日志發送。

生產者：

public class OneWayProducer {
    public static void main(String[] args) throws Exception, MQBrokerException {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //消息發送失敗重試次數
        producer.setRetryTimesWhenSendFailed(3);
        //關閉vip通道，阿里，騰訊上部署時建議關閉，不然可能發送失敗
        //在producer和consumer連接時連接的是vip端口，導致發送、啟動失敗；建議在MQProducerAutoConfiguration 和
        // MQConsumerAutoConfiguration中對producer和consumer進行配置參數讀取並設置VipChannelEnabled的，以便通過配置即可對低版本服務端的支持
        producer.setVipChannelEnabled(false);
        //3.啟動producer
        producer.start();
        for (int i = 0; i < 3; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            Message msg = new Message("OneWay", "Tag2", ("單向消息" + i).getBytes());
            //5.發送單向消息
            producer.sendOneway(msg);
            //線程睡1秒
            TimeUnit.SECONDS.sleep(5);
        }
        //6.關閉生產者producer
        producer.shutdown();
    }
}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group2");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("OneWay", "*");
        //設定消費模式：負載均衡:CLUSTERING  ||  廣播模式:BROADCASTING
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動-----------------------------");
    }
}

4、批量發送消息

生產者：

public class Producer {
    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        producer.setVipChannelEnabled(false);
        producer.setRetryTimesWhenSendFailed(3);
        //3.啟動producer
        producer.start();
        List<Message> msgs = new ArrayList<Message>();
        //4.創建消息對象，指定主題Topic、Tag和消息體
        /**
         * 參數一：消息主題Topic
         * 參數二：消息Tag
         * 參數三：消息內容
         */
        Message msg1 = new Message("BatchTopic", "Tag1", ("Hello World" + 1).getBytes());
        Message msg2 = new Message("BatchTopic", "Tag1", ("Hello World" + 2).getBytes());
        Message msg3 = new Message("BatchTopic", "Tag1", ("Hello World" + 3).getBytes());
        msgs.add(msg1);
        msgs.add(msg2);
        msgs.add(msg3);
        //5.發送消息
        SendResult result = producer.send(msgs);
        //發送狀態
        SendStatus status = result.getSendStatus();
        System.out.println("發送結果:" + result);
        //線程睡1秒
        TimeUnit.SECONDS.sleep(1);
        //6.關閉生產者producer
        producer.shutdown();
    }
}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("BatchTopic", "*");

        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動");
    }
}

5、發送延遲消息

等待一定時間后投遞消息，默認延遲等級 : 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，傳入1代表1s, 2代表5s, 以此類推

生產者：

public class Producer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        producer.setRetryTimesWhenSendFailed(3);
        producer.setVipChannelEnabled(false);
        //3.啟動producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數一：消息主題Topic
             * 參數二：消息Tag
             * 參數三：消息內容
             * 默認延遲等級 : 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h， 傳入1代表1s, 2代表5s, 以此類推
             */
            Message msg = new Message("DelayTopic", "Tag1", ("延遲消息發送" + i).getBytes());
            //設定延遲時間
            msg.setDelayTimeLevel(3);
            //5.發送消息
            SendResult result = producer.send(msg);
            //發送狀態
            SendStatus status = result.getSendStatus();
            System.out.println("發送結果:" + result);
            //線程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //6.關閉生產者producer
        producer.shutdown();
    }
}

消費者：

public class Consumer {

    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        consumer.setVipChannelEnabled(false);
        //3.訂閱主題Topic和Tag
        consumer.subscribe("DelayTopic", "*");

        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("消息ID：【" + msg.getMsgId() + "】,延遲時間：" + (System.currentTimeMillis() - msg.getStoreTimestamp()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動");
    }
}

6、消息過濾

消息過濾通過兩種：tag過濾、sql過濾

<1.>tag過濾

生產者：

public class Producer {
    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.啟動producer
        producer.start();

        for (int i = 0; i < 3; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數一：消息主題Topic
             * 參數二：消息Tag
             * 參數三：消息內容
             */
            Message msg = new Message("FilterTagTopic", "Tag1", ("Hello World" + i).getBytes());
            //5.發送消息
            SendResult result = producer.send(msg);
            //發送狀態
            SendStatus status = result.getSendStatus();

            System.out.println("發送結果:" + result);

            //線程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //6.關閉生產者producer
        producer.shutdown();
    }

}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("FilterTagTopic", "Tag1 || Tag2 ||Tag3");

        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動");
    }
}

<2.>sql過濾【注意】這里要想通過sql過濾必須要在 broker.conf中添加： enablePropertyFilter=true

生產者：

public class Producer {
    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.啟動producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數一：消息主題Topic
             * 參數二：消息Tag
             * 參數三：消息內容
             */
            Message msg = new Message("FilterSQLTopic", "Tag1", ("消息過濾sql" + i).getBytes());
            msg.putUserProperty("i", String.valueOf(i));
            //5.發送消息
            SendResult result = producer.send(msg);
            //發送狀態
            SendStatus status = result.getSendStatus();
            System.out.println("發送結果:" + result);
            //線程睡1秒
            TimeUnit.SECONDS.sleep(2);
        }
        //6.關閉生產者producer
        producer.shutdown();
    }

}

消費者：

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("FilterSQLTopic", MessageSelector.bySql("i>5"));
        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("消費者啟動");
    }
}

7、發送順序消息

順序消息可以通過 new MessageQueueSelector() 消息選擇器放入同一個隊列。【注意】這里的順序只能保證局部的順序，而不是全局的順序性。

構建訂單集合：

/**
 * 訂單構建者
 */
public class OrderStep {
    private long orderId;
    private String desc;
    //省略get set 方法
public static List<OrderStep> buildOrders() {
        //  1039L   : 創建    付款 推送 完成
        //  1065L   ： 創建   付款
        //  7235L   ：創建    付款
        List<OrderStep> orderList = new ArrayList<OrderStep>();

        OrderStep orderDemo = new OrderStep();
        orderDemo.setOrderId(1L);
        orderDemo.setDesc("創建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(2L);
        orderDemo.setDesc("創建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(1L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(3L);
        orderDemo.setDesc("創建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(2L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(3L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(2L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(1L);
        orderDemo.setDesc("推送");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(3L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(1L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        return orderList;
    }
}

生產者：

public class Producer {
    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.啟動producer
        producer.start();
        //構建消息集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();
        //發送消息
        for (int i = 0; i < orderSteps.size(); i++) {
            String body = orderSteps.get(i) + "";
            Message message = new Message("OrderTopic", "Order", "i" + i, body.getBytes());
            /**
             * 參數一：消息對象
             * 參數二：消息隊列的選擇器
             * 參數三：選擇隊列的業務標識（訂單ID）
             */
            SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                /**
                 *
                 * @param mqs：隊列集合
                 * @param msg：消息對象
                 * @param arg：業務標識的參數
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    long orderId = (long) arg;
                    long index = orderId % mqs.size();
                    return mqs.get((int) index);
                }
            }, orderSteps.get(i).getOrderId());

            System.out.println("發送結果：" + sendResult);
        }
        producer.shutdown();
    }

}

消費者：

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("OrderTopic", "*");
        //4.注冊消息監聽器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("QueueId :   "+ msg.getQueueId() + "------"+ new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        //5.啟動消費者
        consumer.start();
        System.out.println("消費者啟動");
    }
}

8、事務消息

先發送一個Half Message：預處理消息，當broker收到此類消息后，會存儲到RMQ_SYS_TRANS_HALF_TOPIC的消息消費隊列中，然后根據本地事務邏輯確定半消息的提交（COMMIT_MESSAGE）、回滾（ROLLBACK_MESSAGE），不確定（UNKNOW）等狀態。如果消息為UNKNOW會進行定期回查，如果多次回查還是不能確定，會進行消息的清除。通過事務監聽器中的回調方法，處理對應的本地事務，消息回查，保證數據的一致性。

生產者：

/**
 * 發送同步消息
 */
public class Producer {
    public static void main(String[] args) throws Exception {
        //1.創建消息生產者producer，並制定生產者組名
        TransactionMQProducer producer = new TransactionMQProducer("group1");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //添加事務監聽器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 在該方法中執行本地事務
             * @param msg
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                if (StringUtils.equals("TAGA", msg.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("TAGB", msg.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if (StringUtils.equals("TAGC", msg.getTags())) {
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }
            /**
             * 該方法時MQ進行消息事務狀態回查
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("沒有獲得消息ack  -----  進行消息回查   消息的Tag:" + msg.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        //3.啟動producer
        producer.start();
        String[] tags = {"TAGA", "TAGB", "TAGC"};
        for (int i = 0; i < 3; i++) {
            //4.創建消息對象，指定主題Topic、Tag和消息體
            /**
             * 參數一：消息主題Topic
             * 參數二：消息Tag
             * 參數三：消息內容
             */
            Message msg = new Message("TransactionTopic", tags[i], ("Hello World" + i).getBytes());
            //5.發送消息
            SendResult result = producer.sendMessageInTransaction(msg, null);
            //發送狀態
            SendStatus status = result.getSendStatus();
            System.out.println("發送結果:" + result);
            //線程睡1秒
            TimeUnit.SECONDS.sleep(2);
        }
        //6.關閉生產者producer
        //producer.shutdown();
    }
}

消費者：

/**
 * 消息的接受者
 */
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1.創建消費者Consumer，制定消費者組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("xx.xx.xx.xx:9876");
        //3.訂閱主題Topic和Tag
        consumer.subscribe("TransactionTopic", "*");
        //4.設置回調函數，處理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //接受消息內容
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "," + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動消費者consumer
        consumer.start();
        System.out.println("生產者啟動");
    }
}

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 RocketMQ基本API使用 RocketMQ使用 RocketMQ的基本使用 RocketMQ的使用 RocketMQ的使用 RocketMQ的基本使用 RocketMQ學習筆記（8）----RocketMQ的Producer API簡介 RocketMQ學習筆記（12）----RocketMQ的Consumer API簡介 rocketmq(三 java操作rocket API, rocketmq 冪等性) Windows安裝RocketMQ，RocketMQ Windows安裝和使用