安装可视化界面

• RocketMQ有一个对其扩展的开源项目rocketmq-externals，该项目中有一个子模块叫rocketmq-console，是一个管理控制台的项目。下载整个扩展项目并编译打包子项目：

git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console 
#修改 rocketmq-console 中配置 namesrv 集群地址：
rocketmq.config.namesrvAddr=192.168.10.102:9876;192.168.10.103:9876
#跳过测试打包
mvn clean package -Dmaven.test.skip=true

• 将生成的rocketmq-console-ng-2.0.0.jar随便放在两台机器中的一台，然后启动rocketmq-console：java -jar rocketmq-console-ng-2.0.0.jar，浏览器输入网址：http://192.168.10.102:8080进行访问。

安装成功

• 新建 SpringBoot 项目，导入MQ客户端依赖：

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

• 生产者发送同步消息：

package com.zzw.rmq.base.producer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.TimeUnit;
/**
 * 发送同步消息
 */
public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //1、创建消息生产者 producer ，并指定生产者组名 group1
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定 NameServer 的地址
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        //设置生产者发送消息的超时时间长一点
        producer.setSendMsgTimeout(60000);
        //3、启动 producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("test-base-mq", "Tag1", ("Hello World" + i).getBytes());
            //5、发送消息
            SendResult result = producer.send(msg);
            //发送状态
            SendStatus status = result.getSendStatus();
            //消息id
            String msgId = result.getMsgId();
            //接收消息的队列id
            int queueId = result.getMessageQueue().getQueueId();
            System.out.println("发送状态：" + status + "，消息id：" + msgId + "，接收消息的队列id：" + queueId);
            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        //6、关闭生产者 producer
        producer.shutdown();
    }
}
/*
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655AD480000，接收消息的队列id：1
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655B1AD0001，接收消息的队列id：2
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655B59B0002，接收消息的队列id：3
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655B9880003，接收消息的队列id：0
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655BD7E0004，接收消息的队列id：1
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655C16B0005，接收消息的队列id：2
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655C5580006，接收消息的队列id：3
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655C9450007，接收消息的队列id：0
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655CD300008，接收消息的队列id：1
发送状态：SEND_OK，消息id：C0A87C173EF418B4AAC24655D11D0009，接收消息的队列id：2
*/

• 生产者发送异步消息：通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。

package com.zzw.rmq.base.producer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.TimeUnit;
/**
 * 发送异步消息
 */
public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        //1、创建消息生产者 producer，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定 NameServer 地址
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        //设置生产者发送消息的超时时间长一点
        producer.setSendMsgTimeout(60000);
        //3、启动producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("test-base-mq", "Tag2", ("Hello World" + 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();
    }
}
/*
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC24650766E0000, offsetMsgId=C0A80A6600002A9F00000000000004D7, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-a, queueId=3], queueOffset=2]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC24650766E0002, offsetMsgId=C0A80A6600002A9F0000000000000426, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-a, queueId=3], queueOffset=1]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC24650766E0003, offsetMsgId=C0A80A6700002A9F0000000000000375, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-b, queueId=3], queueOffset=1]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC24650766E0001, offsetMsgId=C0A80A6700002A9F0000000000000426, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-b, queueId=1], queueOffset=1]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC24650767F0004, offsetMsgId=C0A80A6700002A9F00000000000004D7, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-b, queueId=2], queueOffset=2]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC2465078AE0005, offsetMsgId=C0A80A6700002A9F0000000000000588, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-b, queueId=3], queueOffset=2]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC246507C960006, offsetMsgId=C0A80A6700002A9F0000000000000639, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-b, queueId=0], queueOffset=1]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC2465080800007, offsetMsgId=C0A80A6600002A9F0000000000000588, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-a, queueId=0], queueOffset=2]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC2465084690008, offsetMsgId=C0A80A6700002A9F00000000000006EA, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-b, queueId=3], queueOffset=3]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C173D6418B4AAC2465088530009, offsetMsgId=C0A80A6600002A9F0000000000000639, messageQueue=MessageQueue [topic=test-base-mq, brokerName=broker-a, queueId=0], queueOffset=3]
*/

• 生产者单向发送消息：主要用在不特别关心发送结果的场景，例如日志发送。

package com.zzw.rmq.base.producer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.TimeUnit;
public class OneWayProducer {
    public static void main(String[] args) throws Exception, MQBrokerException {
        //1、创建消息生产者 producer，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        //2、指定 NameServer 地址
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        //设置生产者发送消息的超时时间长一点
        producer.setSendMsgTimeout(60000);
        //3、启动 producer
        producer.start();
        for (int i = 0; i < 3; i++) {
            //4、创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("base", "Tag3", ("Hello World，单向消息" + i).getBytes());
            //5、发送单向消息
            producer.sendOneway(msg);
            //线程睡1秒
            TimeUnit.SECONDS.sleep(5);
        }
        //6、关闭生产者producer
        producer.shutdown();
    }
}

• 消费者消费消息的两种模式：负载均衡和广播模式，默认是负载均衡模式。

package com.zzw.rmq.base.consumer;
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.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws Exception {
        //1、创建消费者 Consumer，并指定消费者组名（push消费模式）
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        //2、指定 NameServer 地址
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        //3、订阅主题Topic和Tag， * 表示所有 tag
        consumer.subscribe("test-base-mq", "*");
        /**
         *  设定消费模式：负载均衡（默认） | 广播模式
         *      负载均衡：多个消费者共同消费队列消息，每个消费者处理的消息不同
         *      广播模式：不同的消费者接收到的消息是一样的
         */
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //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();
    }
}

• 顺序消息：消息有序指可以按照消息的发送顺序来消费（FIFO）。RocketMQ可以严格地保证消息有序，分为分区有序或全局有序。
• 在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue（分区队列）；而消费消息时从多个queue上拉取消息，这种情况不能保证消费顺序。但若控制发送的顺序消息只依次发送到同一个queue中，消费时只从这个queue上依次拉取，则保证了消费顺序。当发送和消费参与的queue只有一个，即为全局有序；若多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。
• 下面用订单进行分区有序的示例。一个订单的顺序流程是：创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时同一个OrderId获取到的肯定是同一个队列。
• 订单构建者：

package com.zzw.rmq.order;
import java.util.ArrayList;
import java.util.List;
/**
 * 订单构建者
 */
public class OrderStep {
    private long orderId;
    private String desc;
    public long getOrderId() {
        return orderId;
    }
    public void setOrderId(long orderId) {
        this.orderId = orderId;
    }
    public String getDesc() {
        return desc;
    }
    public void setDesc(String desc) {
        this.desc = desc;
    }
    @Override
    public String toString() {
        return "OrderStep{" +
                "orderId=" + orderId +
                ", desc='" + desc + '\'' +
                '}';
    }
    public static List<OrderStep> buildOrders() {
        //  1039L：创建  付款  推送  完成
        //  1065L：创建  付款
        //  7235L：创建  付款
        List<OrderStep> orderList = new ArrayList<OrderStep>();

        OrderStep orderDemo = new OrderStep();
        orderDemo.setOrderId(1039L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new OrderStep();
        orderDemo.setOrderId(1065L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

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

        orderDemo = new OrderStep();
        orderDemo.setOrderId(7235L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

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

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

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

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

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

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

• 生产者发送顺序消息：

package com.zzw.rmq.order;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import java.util.List;
public class Producer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        //设置生产者发送消息的超时时间长一点
        producer.setSendMsgTimeout(60000);
        producer.start();
        //构建消息集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();
        //发送消息
        for (int i = 0; i < orderSteps.size(); i++) {
            String body = orderSteps.get(i) + "";
            /**
             * 参数一：消息对象
             * 参数二：消息队列的选择器
             * 参数三：选择队列的业务标识（订单ID）
             */
            Message message = new Message("OrderTopic", "Order", "i" + i, body.getBytes());
            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();
    }
}

• 消费者顺序消费消息：

package com.zzw.rmq.order;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 若非第一次启动，则按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("OrderTopic", "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            //同一个消息队列只会被同一个线程进行处理
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("线程名称：【" + Thread.currentThread().getName() + "】：" + new String(msg.getBody()) + "，消息队列id：" + msg.getQueueId());
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("消费者启动");
    }
}

/*
消费者启动
线程名称：【ConsumeMessageThread_1】：OrderStep{orderId=1065, desc='创建'}，消息队列id：1
线程名称：【ConsumeMessageThread_2】：OrderStep{orderId=7235, desc='创建'}，消息队列id：3
线程名称：【ConsumeMessageThread_1】：OrderStep{orderId=1065, desc='付款'}，消息队列id：1
线程名称：【ConsumeMessageThread_2】：OrderStep{orderId=7235, desc='付款'}，消息队列id：3
线程名称：【ConsumeMessageThread_1】：OrderStep{orderId=1065, desc='完成'}，消息队列id：1
线程名称：【ConsumeMessageThread_2】：OrderStep{orderId=7235, desc='完成'}，消息队列id：3
线程名称：【ConsumeMessageThread_3】：OrderStep{orderId=1039, desc='创建'}，消息队列id：3
线程名称：【ConsumeMessageThread_3】：OrderStep{orderId=1039, desc='付款'}，消息队列id：3
线程名称：【ConsumeMessageThread_3】：OrderStep{orderId=1039, desc='推送'}，消息队列id：3
线程名称：【ConsumeMessageThread_3】：OrderStep{orderId=1039, desc='完成'}，消息队列id：3
*/

• 延迟消息：比如在电商里提交了一个订单就发送一个延时消息，1h后去检查这个订单的状态，若还是未付款则取消订单释放库存。
• 使用限制：RocketMq并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18：

// org/apache/rocketmq/store/config/MessageStoreConfig.java
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

• 生产者发送延迟消息：

package com.zzw.rmq.delay;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.concurrent.TimeUnit;
public class Producer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        producer.setSendMsgTimeout(60000);
        producer.start();
        for (int i = 0; i < 10; i++) {
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("DelayTopic", "Tag1", ("Hello World" + i).getBytes());
            //多了这行代码：设定延迟等级为1，即这个消息将在1s后发送
            msg.setDelayTimeLevel(1);
            SendResult result = producer.send(msg);
            System.out.println("发送结果：" + result);
            //线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
    }
}

• 消费者消费延迟消息：

package com.zzw.rmq.delay;
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.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        consumer.subscribe("DelayTopic", "*");
        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()) + "ms later");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("消费者启动");
    }
}
/*
消费者启动
消息ID：【C0A87C17250818B4AAC24A4FE0650001】，延迟时间：588ms later
消息ID：【C0A87C17250818B4AAC24A4FDBFD0000】，延迟时间：1698ms later
消息ID：【C0A87C17250818B4AAC24A4FE4BD0002】，延迟时间：108ms later
消息ID：【C0A87C17250818B4AAC24A4FE91C0003】，延迟时间：101ms later
消息ID：【C0A87C17250818B4AAC24A4FED0D0004】，延迟时间：7ms later
消息ID：【C0A87C17250818B4AAC24A4FF1630005】，延迟时间：5ms later
消息ID：【C0A87C17250818B4AAC24A4FF5550006】，延迟时间：5ms later
消息ID：【C0A87C17250818B4AAC24A4FF9430007】，延迟时间：7ms later
消息ID：【C0A87C17250818B4AAC24A4FFD3A0008】，延迟时间：105ms later
消息ID：【C0A87C17250818B4AAC24A5001290009】，延迟时间：1ms later
*/

• 批量消息：批量发送消息能显著提高传递小消息的性能，限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，不能是延时消息，并且这批消息的总大小不能超过4MB。
• 生产者发送批量消息：若每次发送的消息大小不超过4MB，则使用以下模板1：

package com.zzw.rmq.batch;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
public class Producer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        producer.setSendMsgTimeout(60000);
        producer.start();
        List<Message> msgs = new ArrayList<>();
        /**
         * 参数一：消息主题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);
        SendResult result = producer.send(msgs);
        System.out.println("发送结果：" + result);
        TimeUnit.SECONDS.sleep(1);
        producer.shutdown();
    }
}
/*
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C171BD418B4AAC24A5798310000,C0A87C171BD418B4AAC24A5798320001,C0A87C171BD418B4AAC24A5798320002, offsetMsgId=C0A80A6700002A9F0000000000002BCC,C0A80A6700002A9F0000000000002C7B,C0A80A6700002A9F0000000000002D2A, messageQueue=MessageQueue [topic=BatchTopic, brokerName=broker-b, queueId=3], queueOffset=0]
*/

• 生产者发送批量消息：若发送消息的总长度大于4MB，则使用以下模板2：

package com.zzw.rmq.batch;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
public class ProducerOther {
    public static void main(String[] args) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        producer.setSendMsgTimeout(60000);
        producer.start();
        List<Message> messages = new ArrayList<>();
        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<Message> listItem = splitter.next();
                producer.send(listItem);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
class ListSplitter implements Iterator<List<Message>> {
    private final List<Message> messages;
    private int currIndex;
    public ListSplitter(List<Message> messages) {
        this.messages = messages;
    }
    @Override
    public boolean hasNext() {
        return currIndex < messages.size();
    }
    @Override
    public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; // 增加日志的开销20字节
            int SIZE_LIMIT = 1024 * 1024 * 4;
            if (tmpSize > SIZE_LIMIT) {
                //单个消息超过了最大的限制
                //忽略,否则会阻塞分裂的进程
                if (nextIndex - currIndex == 0) {
                    //若下一个子列表没有元素，则添加这个子列表然后退出循环，否则只是退出循环
                    nextIndex++;
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
}

• 消费者消费消息：

package com.zzw.rmq.batch;
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.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        consumer.subscribe("BatchTopic", "*");
        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;
            }
        });
        consumer.start();
        System.out.println("消费者启动");
    }
}
/*
消费者启动
consumeThread=ConsumeMessageThread_2，Hello World2
consumeThread=ConsumeMessageThread_1，Hello World1
consumeThread=ConsumeMessageThread_3，Hello World3
*/

• TAG过滤消息，生产者发送消息：

package com.zzw.rmq.filter.tag;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.TimeUnit;
public class Producer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        producer.setSendMsgTimeout(60000);
        producer.start();
        for (int i = 0; i < 3; i++) {
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("FilterTagTopic", "Tag1", ("Hello World" + i).getBytes());
            SendResult result = producer.send(msg);
            System.out.println("发送结果：" + result);
            TimeUnit.SECONDS.sleep(1);
        }
        producer.shutdown();
    }
}

• 消费者按TAG过滤并消费消息：

package com.zzw.rmq.filter.tag;
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.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        consumer.subscribe("FilterTagTopic", "Tag1 || Tag2");
        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;
            }
        });
        consumer.start();
        System.out.println("消费者启动");
    }
}
/*
消费者启动
consumeThread=ConsumeMessageThread_1，Hello World2
consumeThread=ConsumeMessageThread_2，Hello World1
consumeThread=ConsumeMessageThread_3，Hello World0
consumeThread=ConsumeMessageThread_4，Hello World0
consumeThread=ConsumeMessageThread_5，Hello World1
consumeThread=ConsumeMessageThread_6，Hello World2
*/

• SQL过滤消息：RocketMQ配置文件默认是不开启属性过滤的，所以测试过程中会出现这样的错误：Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: CODE: 1 DESC: The broker does not support consumer to filter message by SQL92，解决方案如下：
• 修改两台机器中RocketMQ的安装目录的conf/2m-2s-async下的配置文件：
  • 服务器IP（192.168.10.102）：vim /opt/module/rocketmq/conf/2m-2s-sync/broker-a.properties，vim /opt/module/rocketmq/conf/2m-2s-sync/broker-b-s.properties
  • 服务器IP（192.168.10.103）：vim /opt/module/rocketmq/conf/2m-2s-sync/broker-b.properties，vim /opt/module/rocketmq/conf/2m-2s-sync/broker-a-s.properties
  • 在文件末尾追加以下内容：

#是否支持根据属性过滤 若使用基于标准的sql92模式过滤消息，则改参数必须设置为true
enablePropertyFilter=true

• 关闭broker集群：sh bin/mqshutdown broker，重新启动broker集群，重启后使用jps命令查看启动进程的状态：

#启动Broker集群
#启动 Master1 节点（服务器ip：192.168.10.102）：
nohup sh bin/mqbroker -c /opt/module/rocketmq/conf/2m-2s-sync/broker-a.properties &
#启动 Slave2 节点（服务器ip：192.168.10.102）：
nohup sh bin/mqbroker -c /opt/module/rocketmq/conf/2m-2s-sync/broker-b-s.properties &
#启动 Master2 节点（服务器ip：192.168.10.103）：
nohup sh bin/mqbroker -c /opt/module/rocketmq/conf/2m-2s-sync/broker-b.properties &
#启动 Slave1 节点（服务器ip：192.168.10.103）：
nohup sh bin/mqbroker -c /opt/module/rocketmq/conf/2m-2s-sync/broker-a-s.properties &

• 生产者发送消息，通过putUserProperty来设置自定义过滤属性：

package com.zzw.rmq.filter.sql;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.TimeUnit;
public class Producer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        producer.setSendMsgTimeout(60000);
        producer.start();
        for (int i = 0; i < 10; i++) {
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("FilterSQLTopic", "Tag1", ("Hello World" + i).getBytes());
            //设置自定义的属性
            msg.putUserProperty("i", String.valueOf(i));
            SendResult result = producer.send(msg);
            System.out.println("发送结果：" + result);
            TimeUnit.SECONDS.sleep(2);
        }
        producer.shutdown();
    }
}

• RocketMQ只定义了一些基本语法来支持SQL过滤消息：
  • 数值比较：>，>=，<，<=，BETWEEN，=
  • 字符比较：=，<>，IN
  • IS NULL，IS NOT NULL
  • 逻辑符号：AND，OR，NOT
  • 数值，如：123，3.1415
  • 字符，如：'abc'，必须用单引号包裹起来
  • 特殊的常量：NULL
  • 布尔值：TRUE，FALSE
• 只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：public void subscribe(finalString topic, final MessageSelector messageSelector)。
• 消费者消费消息，通过消息选择器来实现消息过滤：

package com.zzw.rmq.filter.sql;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        //参数2：消息选择器
        consumer.subscribe("FilterSQLTopic", MessageSelector.bySql("i>5"));
        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;
            }
        });
        consumer.start();
        System.out.println("消费者启动");
    }
}
/*
消费者启动
consumeThread=ConsumeMessageThread_1，Hello World6
consumeThread=ConsumeMessageThread_2，Hello World7
consumeThread=ConsumeMessageThread_3，Hello World9
consumeThread=ConsumeMessageThread_4，Hello World8
*/

事务消息的解决方案

• 事务消息发送及提交流程分析：
  • 发送half消息；
  • 服务端响应消息写入结果；
  • 根据发送结果执行本地事务（若写入失败，则half消息对业务不可见，本地逻辑不执行）；
  • 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）。
• 事务补偿流程分析：
  • 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”；
  • Producer收到回查消息，检查回查消息对应本地事务的状态；
  • 根据本地事务状态，重新Commit或者Rollback。
• 事务消息共有三种状态：提交状态、回滚状态、中间状态。
  • TransactionStatus.CommitTransaction：提交事务，它允许消费者消费此消息。
  • TransactionStatus.RollbackTransaction：回滚事务，它代表该消息将被删除，不允许被消费。
  • TransactionStatus.Unknown：中间状态，它代表需要检查消息队列来确定状态。
• 生产者发送事务消息：

package com.zzw.rmq.transaction;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.concurrent.TimeUnit;
public class Producer {
    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("group5");
        producer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        producer.setSendMsgTimeout(60000);
        //添加事务监听器
        producer.setTransactionListener(new TransactionListener() {
            //在该方法中执行本地事务
            @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进行消息事务状态回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("消息的Tag:" + msg.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;//提交消息后就能被消费
            }
        });
        producer.start();
        String[] tags = {"TAGA", "TAGB", "TAGC"};
        for (int i = 0; i < 3; i++) {
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message msg = new Message("TransactionTopic", tags[i], ("Hello World" + i).getBytes());
            SendResult result = producer.sendMessageInTransaction(msg, null);
            System.out.println("发送结果：" + result);
            TimeUnit.SECONDS.sleep(2);
        }
        //不能关闭，不然回查方法不能被执行
        //producer.shutdown();
    }
}
/*
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C1733B418B4AAC24AA52C540000, offsetMsgId=null, messageQueue=MessageQueue [topic=TransactionTopic, brokerName=broker-b, queueId=1], queueOffset=0]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C1733B418B4AAC24AA534690001, offsetMsgId=null, messageQueue=MessageQueue [topic=TransactionTopic, brokerName=broker-b, queueId=2], queueOffset=1]
发送结果：SendResult [sendStatus=SEND_OK, msgId=C0A87C1733B418B4AAC24AA53C490002, offsetMsgId=null, messageQueue=MessageQueue [topic=TransactionTopic, brokerName=broker-b, queueId=3], queueOffset=2]
消息的Tag:TAGC
*/

• 消费者消费事务消息：

package com.zzw.rmq.transaction;
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.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
public class Consumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
        consumer.setNamesrvAddr("192.168.10.102:9876;192.168.10.103:9876");
        consumer.subscribe("TransactionTopic", "*");
        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;
            }
        });
        consumer.start();
        System.out.println("消费者启动");
    }
}
/*
消费者启动
consumeThread=ConsumeMessageThread_1，Hello World0
consumeThread=ConsumeMessageThread_2，Hello World2
*/

• 实现事务的监听接口：当发送半消息成功时，使用executeLocalTransaction方法来执行本地事务，它将返回前面提到的三个事务状态之一。checkLocalTranscation 方法用于检查本地事务状态，并响应消息队列的检查请求，它也是返回前面提到的三个事务状态之一。
• 事务消息的使用限制：
  • 不支持延时消息和批量消息。
  • 为了避免单个消息被检查多次而导致半队列消息累积，默认将单个消息的检查次数限制为 15 次，用户可以通过 Broker 配置文件中的transactionCheckMax参数来修改此限制。若已经检查某条消息超过 N 次的话（ N = transactionCheckMax ） 则 Broker 将丢弃此消息，同时打印错误日志。用户可以通过重写 AbstractTransactionCheckListener类来修改这个行为。
  • 事务消息将在 Broker 配置文件中的参数transactionMsgTimeout特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS来改变这个限制，该参数优先于transactionMsgTimeout参数。
  • 事务性消息可能不止一次被检查或消费。
  • 提交给用户的目标主题消息可能会失败，目前这种依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，若希望事务消息不丢失、事务完整性得到保证，则建议使用同步的双重写入机制。
  • 事务消息的生产者 ID 不能与其它类型消息的生产者 ID 共享。与其它类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。