RocketMq的簡單用法

RocketMq用法

引入jar包

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

//生産值代碼塊
public class RocketMqProducer {

    public static void main(String[] args) {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("order_Producer");
            producer.setNamesrvAddr("ip");
            producer.start();

            for (int i = 1; i <= 5; i++) {
                //主題、标簽、值、内容
                Message msg = new Message("TopicOrderTest", "order_1", "KEY" + i, ("消息内容： " + i).getBytes());

                SendResult sendResult = producer.send(msg, (List<MessageQueue> mqs, Message message, Object arg)->  {

                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);

                }, 0);

                System.out.println(sendResult);
            }

            producer.shutdown();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 同步發送消息傳回消息結果
     * @throws Exception
     */
    private void sendResultMessage() throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("test");
        producer.start();
        for (int i = 0; i < 100; i++) {
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }

    /**
     * 異步發送消息
     * @throws MQClientException
     * @throws InterruptedException
     * @throws UnsupportedEncodingException
     */
    private void sendMessage()throws MQClientException, InterruptedException, UnsupportedEncodingException {

            DefaultMQProducer producer = new DefaultMQProducer("Jodie_Daily_test");
            producer.start();
            producer.setRetryTimesWhenSendAsyncFailed(0);

            for (int i = 0; i < 10000000; i++) {
                try {
                    final int index = i;
                    Message msg = new Message("Jodie_topic_1023",
                            "TagA",
                            "OrderID188",
                            "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    //重點在這裡 異步發送回調
                    producer.send(msg, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                        }

                        @Override
                        public void onException(Throwable e) {
                            System.out.printf("%-10d Exception %s %n", index, e);
                            e.printStackTrace();
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            producer.shutdown();
    }


    /**
     * 單向發送日志消息
     */
    private void sendOnewayMessage() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("Test");
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            producer.sendOneway(msg);

        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }



}
           

//消費者代碼塊
public class RocketMqConsumer{
//分組名
String producerGroup = "";
//ip位址
String nameServerAddress = "";
 // 執行個體消息的生産者，然後通過構造方法制定其組
        DefaultMQPushConsumer defaultMqPushConsumer = new DefaultMQPushConsumer();
        defaultMqPushConsumer.setConsumerGroup(producerGroup);
         //設定nameServer的位址 
        defaultMqPushConsumer.setNamesrvAddr(nameServerAddress);
        /**
         * 設定Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費<br>
         * 如果非第一次啟動，那麼按照上次消費的位置繼續消費
         */
        defaultMqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //當分組為指定一個分組時 全組tag為一個值，這樣才能正常接收到消息defaultMqPushConsumer.subscribe("toipc","*");
        //當分組和訂閱為一一對應時，可以訂閱不同的tag，但是當方法重新開機時，會建立一個新的消費者，那麼所有的任務又會重新消費一次
        defaultMqPushConsumer.subscribe("toipc","tag");
        System.out.println("開始接受消息==========================");
        defaultMqPushConsumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                // 設定自動送出
                context.setAutoCommit(true);
                for (MessageExt msg : msgs) {
 
                    Thread t0 = new Thread(() -> {
                       //自己的業務處理代碼
                        System.out.println("receive msg:" + new String(msg.getBody()));
                    });
                    t0.start();
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }

        });
        defaultMqPushConsumer.start();
        return defaultMqPushConsumer;
    }
}