RocketMQ简单入门

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
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.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.RPCHook;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class TestRocketMq1 {
    public static void main(String[] args) throws Exception {
        testSyncSendMessage();
//        testAsyncSendMessage()
//        testSyncSendDelayMessage();
//        testConcurrentlyConsumerMessage();
        TimeUnit.SECONDS.sleep(3);
        testOrderlyConsumerMessage();
    }

    /**
     * 并发消费(非顺序)
     *
     * @throws Exception
     */
    private static void testConcurrentlyConsumerMessage() throws Exception {
        DefaultMQPushConsumer consumer = createConsumer();
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                    System.out.println("content：" + content + "," + msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    /**
     * 顺序消费 先发送的消息先消费
     *
     * @throws Exception
     */
    private static void testOrderlyConsumerMessage() throws Exception {
        DefaultMQPushConsumer consumer = createConsumer();
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    String content = new String(msg.getBody(), StandardCharsets.UTF_8);
                    System.out.println("content：" + content + "," + msg);
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }

    /**
     * 发送同步消息
     *
     * @throws Exception
     */
    private static void testSyncSendMessage() throws Exception {
        MQProducer producer = createProducer();
        producer.start();
        //第一个参数是topic，这是主题
        //第二个参数是tags，这是可选参数，用于消费端过滤消息
        //第三个参数是keys，这也是可选参数，如果有多个，用空格隔开。RocketMQ可以根据这些key快速检索到消息，相当于
        //消息的索引，可以设置为消息的唯一编号（主  键）。
        Message msg = new Message("test_topic", "TagA", "6666", "RocketMQ Test message".getBytes());
        //SendResult是发送结果的封装，包括消息状态，消息id，选择的队列等等，只要不抛异常，就代表发送成功
        SendResult sendResult = producer.send(msg);
        System.out.println("sync sendResult: " + sendResult);
        producer.shutdown();
    }

    /**
     * 发送同步延迟消息
     *
     * @throws Exception
     */
    private static void testSyncSendDelayMessage() throws Exception {
        MQProducer producer = createProducer();
        producer.start();
        //第一个参数是topic，这是主题
        //第二个参数是tags，这是可选参数，用于消费端过滤消息
        //第三个参数是keys，这也是可选参数，如果有多个，用空格隔开。RocketMQ可以根据这些key快速检索到消息，相当于
        //消息的索引，可以设置为消息的唯一编号（主  键）。
        Message msg = new Message("test_topic", "TagA", "6666", "RocketMQ Test message".getBytes());
        //delayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        //设置消息延迟级别等于0时，则该消息为非延迟消息。
        //设置消息延迟级别大于等于1并且小于等于18时，消息延迟特定时间，如：设置消息延迟级别等于1，则延迟1s；设置消息延迟级别等于2，则延迟5s，以此类推。
        //设置消息延迟级别大于18时，则该消息延迟级别为18，如：设置消息延迟级别等于20，则延迟2h。
        //设置了此延迟级别就是延迟消息了
        msg.setDelayTimeLevel(2);
        //SendResult是发送结果的封装，包括消息状态，消息id，选择的队列等等，只要不抛异常，就代表发送成功
        SendResult sendResult = producer.send(msg);
        System.out.println("sync sendResult: " + sendResult);
        producer.shutdown();
    }

    /**
     * 发送异步消息
     *
     * @throws Exception
     */
    private static void testAsyncSendMessage() throws Exception {
        MQProducer producer = createProducer();
        producer.start();
        //第一个参数是topic，这是主题
        //第二个参数是tags，这是可选参数，用于消费端过滤消息
        //第三个参数是keys，这也是可选参数，如果有多个，用空格隔开。RocketMQ可以根据这些key快速检索到消息，相当于
        //消息的索引，可以设置为消息的唯一编号（主  键）。
        Message msg = new Message("test_topic", "TagA", "6666", "RocketMQ Async message".getBytes());
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("async sendResult: " + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("async sendResult: " + e);
            }
        });
        //必须等回调结果之后才能shutdown
        TimeUnit.SECONDS.sleep(5);
        producer.shutdown();
    }

    private static MQProducer createProducer() {

        DefaultMQProducer producer = new DefaultMQProducer("test_producer_group", createRpcHook());
        //生产者需用通过NameServer获取所有broker的路由信息，多个用分号隔开，这个跟Redis哨兵一样
        producer.setNamesrvAddr("42.192.20.119:9876");
        return producer;
    }

    private static DefaultMQPushConsumer createConsumer() throws Exception {
        //消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(createRpcHook());
        consumer.setConsumerGroup("test_consumer_group");
        //消费者从NameServer拿到topic的queue所在的Broker地址，多个用分号隔开
        consumer.setNamesrvAddr("42.192.20.119:9876");
        //设置Consumer第一次启动是从队列头部开始消费
        //如果非第一次启动，那么按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //subscribe订阅的第一个参数就是topic,第二个参数为生产者发送时候的tags，*代表匹配所有消息，
        //想要接收具体消息时用||隔开，如"TagA||TagB||TagD"
        consumer.subscribe("test_topic", "TagA");
        //Consumer可以用两种模式启动，广播（Broadcast）和集群（Cluster），广播模式下，一条消息会发送给所有Consumer，
        //集群模式下消息只会发送给一个Consumer
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //批量消费,每次拉取10条
        consumer.setConsumeMessageBatchMaxSize(10);
        return consumer;
    }

    private static RPCHook createRpcHook() {
        //账号密码
        String accessKey = "my_admin";
        String secretKey = "szz123456";
        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
    }
}