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;
    }
}