rocketmq基本使用

pom依赖

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

一、同步发送消息

       public static void main(String[] args) throws Exception {
        // 创建一个producer，参数为Producer group ID
        DefaultMQProducer producer = new DefaultMQProducer("money");
        // 指定nameServer地址
        producer.setNamesrvAddr("192.168.200.132:9876");
        // 设置当发送失败时重试发送的次数，默认为2次
        producer.setRetryTimesWhenSendFailed(3);
        // 设置发送超时时效为5s，默认为3s
        producer.setSendMsgTimeout(5000);
        // 开启生产者
        producer.start();

        // 生产并发送100条消息
        for (int i = 0; i < 100; i++) {
            byte[] body = ("money" + i).getBytes();
            Message message = new Message("moneyTopic", "moneyTag", body);
            // 为消息指定key
            message.setKeys("moneyKey" + i);
            // 发送消息
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult);
        }
        // 关闭producer
        producer.shutdown();
    }

二、异步发送消息

    public static void main(String[] args) throws Exception {
        // 创建一个producer，参数为Producer group ID
        DefaultMQProducer producer = new DefaultMQProducer("money");
        // 指定nameServer地址
        producer.setNamesrvAddr("192.168.200.132:9876");
        // 指定异步发送失败后不进行重试发送
        producer.setRetryTimesWhenSendAsyncFailed(0);
        // 指定新创建的Topic数量为2，默认为4
        producer.setDefaultTopicQueueNums(2);
        // 开启生产者
        producer.start();

        // 生产并发送100条消息
        for (int i = 0; i < 100; i++) {
            byte[] body = ("money" + i).getBytes();
            Message message = new Message("moneyAsynTopicA", "moneyTag", body);
            // 为消息指定key
            message.setKeys("moneyKey" + i);
            // 发送消息，指定回调
            producer.send(message, new SendCallback() {
                // 当producer接收到MQ发送来的ACK后就会触发该回调方法的执行
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);

                }

                @Override
                public void onException(Throwable throwable) {
                    throwable.printStackTrace();
                }
            });
        }
        // 由于采取的异步，还没发送完就关了，所以修改休眠一会
        TimeUnit.SECONDS.sleep(5);
        // 关闭producer
        producer.shutdown();
    }

三、consumer消费消息

    public static void main(String[] args) throws Exception {
        // 定义一个pull消息
//        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("consumerGroup");
        // 定义一个push消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
        // 指定nameServer地址
        consumer.setNamesrvAddr("192.168.200.132:9876");
        // 指定从第一条消息开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 指定消费topic与tag
        consumer.subscribe("moneyTopic", "*");

        // 指定采用“广播模式”，进行消费，默认为“集群模式”
        consumer.setMessageModel(MessageModel.BROADCASTING);

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            // 一但broker中有了其订阅的消息就会触发该方法的执行
            // 其返回值为当前consumer消费的状态
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 逐条消息消息
                for (MessageExt msg : msgs) {
                    System.out.println(msg);
                }
                // 返回消息状态，消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 开启消费者消费
        consumer.start();
    }

四、顺序消息

    public static void main(String[] args) throws Exception {
        // 创建一个producer，参数为Producer group ID
        DefaultMQProducer producer = new DefaultMQProducer("money");
        // 指定nameServer地址
        producer.setNamesrvAddr("192.168.200.132:9876");
        // 开启生产者
        producer.start();
        // 生产并发送100条消息
        for (int i = 0; i < 100; i++) {
            Integer orderId = i;
            byte[] body = ("money" + i).getBytes();
            Message message = new Message("moneyTopic", "moneyTag", body);
            // 将orderId作为消息key
            message.setKeys(orderId.toString());
            // send()的第三个参数值会传递给选择器的select()的第三个参数
            // 该send为同步发送
            SendResult sendResult = producer.send(message, new MessageQueueSelector() {
                // 具体的选择算法在该方法中定义
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message message, Object arg) {
                    // 以下是使用消息key作为 选择的选择算法
                    String keys = message.getKeys();
                    Integer id = Integer.valueOf(keys);

                    // 以下是使用arg作为选择key的选择算法
                    // Integer id = (Integer) arg;

                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            }, orderId);
            System.out.println(sendResult);
        }
        producer.shutdown();
    }

五、延迟消费

public class DelayProducer extends Exception {

    public static void main(String[] args) throws Exception {
        // 创建一个producer，参数为Producer group ID
        DefaultMQProducer producer = new DefaultMQProducer("money");
        // 指定nameServer地址
        producer.setNamesrvAddr("192.168.200.132:9876");
        // 开启生产者
        producer.start();

        // 生产并发送100条消息
        for (int i = 0; i < 10; i++) {
            byte[] body = ("money" + i).getBytes();
            Message message = new Message("delayTopic", "delayTag", body);
            // 指定消息延迟等级为3级，即延迟10s
            message.setDelayTimeLevel(3);
            SendResult sendResult = producer.send(message);
            // 输出消息被发送的时间
            System.out.print(new SimpleDateFormat("mm:ss ").format(new Date()));
            System.out.println(sendResult);
        }
        producer.shutdown();
    }
}

六、发送事务消息

public class TransactionProducer {

    public static void main(String[] args) throws Exception {
        // 创建一个producer，参数为Producer group ID
        TransactionMQProducer producer = new TransactionMQProducer("money");
        // 指定nameServer地址
        producer.setNamesrvAddr("192.168.200.132:9876");

        /**
         * 定义一个线程池
         * corePoolSize:线程池中核心线程数量
         * maximumPoolSize:线程池中最多线程数
         * keepAliveTime:当线程池中线程数量大于核心线程数量时，多余空闲线程的存活时长
         * unit:时间单位
         * workQueue:临时存放任务的队列，其参数就是队列的长度
         * threadFactory:线程工厂
         */
        ThreadPoolExecutor executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("client-transation-msg-check-thread");
                        return thread;
                    }
                });

        // 为生产者指定一个线程池
        producer.setExecutorService(executorService);
        // 为生产者添加事务监听器
        producer.setTransactionListener(new ICBCTransactionListener());

        producer.start();

        String[] tags = {"firstTag", "secondTag", "thirdTag"};
        for (int i = 0; i < 3; i++) {
            byte[] body = ("moeny" + i).getBytes();
            Message message = new Message("transactionTopic", tags[i], body);
            // 发送事务消息
            // 第二个参数用于指定在本地事务时要使用的业务参数
            SendResult sendResult = producer.sendMessageInTransaction(message, null);
            System.out.println("发送结果为：" + sendResult);
        }
    }
}


class ICBCTransactionListener implements TransactionListener {
    // 回调操作方法
    // 消息预提交成功就会触发该方法的执行，用于完成本地事务
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg,
                                                         Object arg) {
        System.out.println("预提交消息成功：" + msg);
        // 假设接收到TAGA的消息就表示扣款操作成功，TAGB的消息表示扣款失败，
        // TAGC表示扣款结果不清楚，需要执行消息回查
        if (StringUtils.equals("firstTag", msg.getTags())) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else if (StringUtils.equals("secondTag", msg.getTags())) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        } else if (StringUtils.equals("thirdTag", msg.getTags())) {
            return LocalTransactionState.UNKNOW;
        }
        return LocalTransactionState.UNKNOW;
    }

    // 消息回查方法
    // 引发消息回查的原因最常见的有两个：
    // 1)回调操作返回UNKNWON
    // 2)TC没有接收到TM的最终全局事务确认指令
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("执行消息回查" + msg.getTags());
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}