RocketMQ 2（管理工具、在maven中测试、消费模式、顺序消息、延时消息、批量消息）

港控/mmm° 2023-10-08 12:42 43阅读 0赞

#### 文章目录 ####

*   *  1、集群管理工具
     *   *  1.1、mqadmin （指令）
         *  1.2、rocketmq-console （可视化）
     *  2、在maven项目中测试
     *   *  2.1、测试发送消息
         *   *  2.1.1、发送同步消息
             *  2.1.2、发送异步消息
             *  2.1.3、发送单向消息
         *  2.2、测试接收消息
         *   *  2.2.1、广播模式
             *  2.2.2、负载均衡模式
         *  2.3、顺序消息
         *   *  2.3.1、代码实现
             *  2.3.2、代码总结
         *  2.4、延时消息
         *   *  2.4.1、代码实现
         *  2.5、批量消息
         *   *  2.5.1、消息分割

### 1、集群管理工具 ###

#### 1.1、mqadmin （指令） ####

mqadmin是RocketMQ为我们提供的集群管理工具。  
使用方法：

// 进入rocketmq的bin目录
    cd /usr/local/rocketmq/bin
    // 查看帮助文档
    ./mqadmin clusterList -h
    // 查看nameserver1基本信息
    ./mqadmin clusterList -n nameserver1:9876

mqadmin的指令还有很多，可以百度一下  
如果觉得指令工具太麻烦，就用可视化工具rocketmq-console

#### 1.2、rocketmq-console （可视化） ####

[下载][Link 1]：https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console  
它是一个Springboot的maven工程，需要把它打成一个jar包运行。  
打包前需要修改一下配置文件：  
![在这里插入图片描述][20210204103029900.png]

> 这里是在windows下打包了再上传到linux服务器

打包(需要安装了maven才能打包):

// 在rocketmq-console项目根目录下执行
    mvn clean package -Dmaven.test.skip=true

打包完成把项目根目录下的jar文件发送到服务器。  
运行jar：

java -jar rocketmq-console-ng-1.0.1.jar

运行成功，到浏览器访问192.168.1.11:8080  
![在这里插入图片描述][20210204105802395.png]

### 2、在maven项目中测试 ###

新建一个maven项目（这里我直接创建的springboot）  
在pom.ml里添加rocketmq-client的依赖：

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

消息生产者步骤：

1、创建消息生产者producer，并制定生产者组名
    2、指定nameserver地址
    3、启动producer
    4、创建消息对象，指定Topic、tag、消息体
    5、发送消息
    6、关闭producer

消息消费者步骤：

1、创建消费者Consumer，制定消费者组名
    2、指定nameserver地址
    3、订阅主题Topic和Tag
    4、设置回调函数、处理消息
    5、启动消费者consumer

#### 2.1、测试发送消息 ####

##### 2.1.1、发送同步消息 #####

同步消息：消息发送完毕后会等待在那里，直到接收到回传的消息才继续往下执行。  
新建类SyncProducer：

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.创建生产者对象，并指定组名“group1”
            DefaultMQProducer p = new DefaultMQProducer("group1");
            // 2.指定nameserver地址
            p.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            // 3.启动生产者
            p.start();
            // 4.创建消息对象，指定Topic\Tag\消息体
            for (int i = 0; i < 10; i++) {
        
                Message msg = new Message("base","Tag1",("hi rocketmq "+i).getBytes());
                // 5。发送消息
                SendResult sr = p.send(msg);
                String msgId = sr.getMsgId();                      // 获取消息id
                SendStatus msgStatus = sr.getSendStatus();         // 获取消息状态
                int queueId = sr.getMessageQueue().getQueueId();   // 获取queueId
    
                System.out.println("消息id:"+msgId+"。 消息状态："+msgStatus+"。 队列id"+queueId);
                TimeUnit.SECONDS.sleep(1);                      // 让线程停1秒
            }
            // 6.关闭生产者
            p.shutdown();
    
        }
    }

运行它的main方法：  
![在这里插入图片描述][20210204143816961.png]

##### 2.1.2、发送异步消息 #####

异步消息和同步相比，异步不会等待消息的回馈，如果想要接收消息回馈，可以用一个回调函数接收，如下。  
新建类AsyncProducer：

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.创建生产者对象，并指定组名“group1”
            DefaultMQProducer p = new DefaultMQProducer("group1");
            // 2.指定nameserver地址
            p.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            // 3.启动生产者
            p.start();
            // 4.创建消息对象，指定Topic\Tag\消息体
            for (int i = 0; i < 10; i++) {
        
                Message msg = new Message("base","Tag2",("Async hi rocketmq "+i).getBytes());
                // 5。发送消息
                p.send(msg, new SendCallback() {
        
                    // 发送成功的回调函数
                    @Override
                    public void onSuccess(SendResult sendResult) {
        
                        System.out.println("回调结果:"+sendResult);
                    }
                    // 发送失败的回调函数
                    @Override
                    public void onException(Throwable e) {
        
                        System.out.println("回调结果:"+e);
                    }
                });
    
                TimeUnit.SECONDS.sleep(1);                      // 让线程停1秒
            }
            // 6.关闭生产者
            p.shutdown();
        }
    }

运行它的main方法：  
![在这里插入图片描述][20210204145133453.png]

##### 2.1.3、发送单向消息 #####

一般不需要消息回馈的，就用这种方式。  
新建类OneWayProducer：

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 {
        
            // 1.创建生产者对象，并指定组名“group1”
            DefaultMQProducer p = new DefaultMQProducer("group1");
            // 2.指定nameserver地址
    //        p.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            p.setNamesrvAddr("127.0.0.1:9876");
            // 3.启动生产者
            p.start();
            // 4.创建消息对象，指定Topic\Tag\消息体
            for (int i = 0; i < 10; i++) {
        
                Message msg = new Message("base","Tag3",("oneway hi rocketmq "+i).getBytes());
                // 5。发送消息
                p.sendOneway(msg);
    
                TimeUnit.SECONDS.sleep(1);                      // 让线程停1秒
            }
            // 6.关闭生产者
            p.shutdown();
        }
    }

运行它的main方法：  
![在这里插入图片描述][20210204150103382.png]

#### 2.2、测试接收消息 ####

新建类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 java.util.List;
    
    /*
          接收消息
     */
    public class Consumer {
        
        public static void main(String[] args) throws Exception {
        
            // 1、创建消费者Consumer，制定消费者组名
            DefaultMQPushConsumer c = new DefaultMQPushConsumer("group1");
            // 2、指定nameserver地址
            c.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            // c.setNamesrvAddr("127.0.0.1:9876");
            // 3、订阅主题Topic和Tag
            c.subscribe("base","Tag2");
            // 4、设置回调函数、处理消息 (监听消息)
            c.registerMessageListener(new MessageListenerConcurrently() {
        
                // 接收消息内容
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        
                    // msgs里面传过来的是byte数组，这里我们转换一下
                    for (MessageExt msg:msgs) {
        
                        System.out.println( new String(msg.getBody()) );
                    }
                    // 返回消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 5、启动消费者consumer
            c.start();
        }
    }

运行它的main方法(如果没有收到消息，可以尝试用生产者再发一次)：  
![在这里插入图片描述][20210204154806430.png]  
消费多个Tag：  
![在这里插入图片描述][2021020416154241.png]  
消费所有Tag：  
![在这里插入图片描述][20210204161630307.png]

> 如上没有设置消费模式，默认是负载均衡模式

消费者对消息进行消费时，有两种模式：广播模式和负载均衡模式

##### 2.2.1、广播模式 #####

![在这里插入图片描述][20210204155726436.png]  
代码实现：（其他代码不变，只需要单独设置一个消费模式为广播模式）  
![在这里插入图片描述][20210204162048916.png]

##### 2.2.2、负载均衡模式 #####

![在这里插入图片描述][2021020416095454.png]  
代码中不设置消费模式，默认就是负载均衡模式

#### 2.3、顺序消息 ####

> 想看与普通消息的区别：直接点击 2.3.2、代码总结

rocketmq默认消息是通过多线程、多队列发送的，如下  
![在这里插入图片描述][20210204164116529.png]  
如上图，会导致一个人发送的消息，是分散获取，没有序列。接下来我们采用一个队列一个线程，去接收一组消息的方式。  
![在这里插入图片描述][20210204164547949.png]  
接下来用代码实现顺序消息

##### 2.3.1、代码实现 #####

现在模拟一个订单的顺序流程：创建、付款、推送、完成。订单号相同的消息进入一个队列中。

先创建一个实体类OrderStep，用来保存订单好和描述。（实现了get、set、toString）

import java.util.ArrayList;
    import java.util.List;
    
    public class OrderStep {
        
        private long orderId;
        private String desc;
    
        @Override
        public String toString() {
        
            return "OrderStep{" +
                    "orderId=" + orderId +
                    ", desc='" + 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;
        }
    
        // 默认构建三个订单 1001、1002、1002
        // 1001 : 创建、付款、推送、完成
        // 1002 : 创建、付款
        // 1003 ：创建、付款
        public static List<OrderStep> buildOrders(){
        
            List<OrderStep> orderList = new ArrayList<>();
            OrderStep o = new OrderStep();
            o.setOrderId(1001L);
            o.setDesc("创建");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1002L);
            o.setDesc("创建");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1001L);
            o.setDesc("付款");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1003L);
            o.setDesc("创建");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1002L);
            o.setDesc("付款");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1003L);
            o.setDesc("付款");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1002L);
            o.setDesc("完成");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1001L);
            o.setDesc("推送");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1003L);
            o.setDesc("完成");
            orderList.add(o);
    
            o = new OrderStep();
            o.setOrderId(1001L);
            o.setDesc("完成");
            orderList.add(o);
    
            return orderList;
        }
    }

新建类Producer，用于发送顺序消息

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 p = new DefaultMQProducer("group1");
            p.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            p.start();
    
            List<OrderStep> orderSteps = OrderStep.buildOrders();
            for (int i=0 ; i< orderSteps.size() ; i++) {
        
                byte[] body = (orderSteps.get(i)+"").getBytes();
                Message msg = new Message("orderTopic", "order", "num:" + i, body);
                // 发送顺序消息,参数1：消息对象，参数2：消息队列的选择器，参数3：业务标识
                SendResult sendResult = p.send(msg, new MessageQueueSelector() {
        
                    // 这里的Object o实际上就是订单id
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
        
                        long orderId = (long) o;
                        // 将orderId取模。index = 订单id除以消息队列数量的余数
                        // 这一步相当于在为 ”订单id“ 选择 进哪个队列
                        long index = orderId % list.size();
                        return list.get((int) index);
                    }
                }, orderSteps.get(i).getOrderId());
    
                System.out.println("发送结果："+sendResult);
            }
    
        }
    }

新建类Consumer，用于消费消息

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.*;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    public class Consumer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQPushConsumer c = new DefaultMQPushConsumer("group1");
            c.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            c.subscribe("orderTopic","*");
            // 注册监听器
            c.registerMessageListener(new MessageListenerOrderly() {
        
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        
                    for (MessageExt messageExt : list) {
        
                    	System.out.println("线程名称："+Thread.currentThread().getName());
                        System.out.println("消息内容："+new String(messageExt.getBody()));
    
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            c.start();
        }
    }

接下来启动消费者、发送者看看效果。  
![在这里插入图片描述][20210205113627613.png]  
我们发现同一个订单的消息就被按照顺序进行消费了

##### 2.3.2、代码总结 #####

总结一下顺序消息写的代码，和普通消息不一样的地方：  
生产者：  
![在这里插入图片描述][20210205114054101.png]  
消费者：  
![在这里插入图片描述][20210205114437251.png]

#### 2.4、延时消息 ####

比如电商里，用户提交了订单，但是未付款。这时候就可以设置一个延迟消息，一小时后检查订单是否付款，未付款就释放订单。

rocketmq预设的延迟时间有这些：  
![在这里插入图片描述][20210205120448593.png]

##### 2.4.1、代码实现 #####

消费者（相比原来的代码完全没有变化）

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.*;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    
    // 延时消息，消费者
    public class Consumer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQPushConsumer c = new DefaultMQPushConsumer("group1");
            c.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            c.subscribe("base","*");
            // 注册监听器
            c.registerMessageListener(new MessageListenerConcurrently() {
        
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                                ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        
                    for (MessageExt messageExt : list) {
        
                        // 打印延迟的时间
                        System.out.println("消息id："+messageExt.getMsgId()+ "。"+(System.currentTimeMillis()-messageExt.getStoreTimestamp())+"ms 后收到消息");
    
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            } );
            c.start();
        }
    
    }

生产者（相对原来的代码，只多了一句 msg.setDelayTimeLevel(2);）

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;
    
    // 延迟5秒发送消息
    public class Producer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer p = new DefaultMQProducer("group1");
            p.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            p.start();
    
            for (int i = 0; i < 10; i++) {
        
                Message msg = new Message("base","Tag1",("hi rocketmq "+i).getBytes());
                // 延迟五秒发送
                msg.setDelayTimeLevel(2);
                SendResult sr = p.send(msg);
                String msgId = sr.getMsgId();                      // 获取消息id
                SendStatus msgStatus = sr.getSendStatus();         // 获取消息状态
                int queueId = sr.getMessageQueue().getQueueId();   // 获取queueId
    
                System.out.println("消息id:"+msgId+"。 消息状态："+msgStatus+"。 队列id"+queueId);
                TimeUnit.SECONDS.sleep(1);                      // 让线程停1秒
    
            }
            p.shutdown();
        }
    }

#### 2.5、批量消息 ####

之前的代码都是一条一条发送的消息，现在来批量发送，注意消息总大小不应超出4MB。

消费者和之前的代码一样。

生产者代码里面就多了把“消息”，用ArrsyList装起来。

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.ArrayList;
    import java.util.concurrent.TimeUnit;
    
    // 批量消息,生产者
    public class Producer {
        
        public static void main(String[] args) throws Exception {
        
            DefaultMQProducer p = new DefaultMQProducer("group1");
            p.setNamesrvAddr("192.168.1.11:9876;192.168.1.12:9876");
            p.start();
    
            ArrayList<Message> msgs = new ArrayList<>();
    
            Message msg1 = new Message("ms","Tag1",("hi rocketmq "+1).getBytes());
            Message msg2 = new Message("ms","Tag1",("hi rocketmq "+2).getBytes());
            Message msg3 = new Message("ms","Tag1",("hi rocketmq "+3).getBytes());
    
            msgs.add(msg1);
            msgs.add(msg2);
            msgs.add(msg3);
    
            SendResult sr = p.send(msgs);
            String msgId = sr.getMsgId();                      // 获取消息id
            SendStatus msgStatus = sr.getSendStatus();         // 获取消息状态
            int queueId = sr.getMessageQueue().getQueueId();   // 获取queueId
            System.out.println("消息id:"+msgId+"。 消息状态："+msgStatus+"。 队列id"+queueId);
            TimeUnit.SECONDS.sleep(1);                      // 让线程停1秒
    
            p.shutdown();
        }
    }

##### 2.5.1、消息分割 #####

如果消息大于4MB，需要进行消息分割

新建类ListSplitter

import org.apache.rocketmq.common.message.Message;
    
    import java.util.Iterator;
    import java.util.List;
    import java.util.Map;
    
    // 消息大于4MB时，用此类分割
    public class ListSplitter implements Iterator<List<Message>> {
        
        private final int SIZE_LIMIT = 1024*1024*4;
        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字节
                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;
        }
    }

这个类的使用示例：

ArrayList<Message> msgs = new ArrayList<>();
    
            Message msg1 = new Message("ms","Tag1",("hi rocketmq "+1).getBytes());
            Message msg2 = new Message("ms","Tag1",("hi rocketmq "+2).getBytes());
            Message msg3 = new Message("ms","Tag1",("hi rocketmq "+3).getBytes());
    
            msgs.add(msg1);
            msgs.add(msg2);
            msgs.add(msg3);
    
            ListSplitter splitter = new ListSplitter(msgs);
            while (splitter.hasNext()){
        
                try{
        
                    List<Message> listItem = splitter.next();
                    // producer来自DefaultMQProducer producer = new DefaultMQProducer("group1");
                    producer.send(listItem);
                }catch (Exception e){
        
                    e.printStackTrace();
                    // 处理异常
                }
            }

[Link 1]: https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console
[20210204103029900.png]: https://img-blog.csdnimg.cn/20210204103029900.png
[20210204105802395.png]: https://img-blog.csdnimg.cn/20210204105802395.png
[20210204143816961.png]: https://img-blog.csdnimg.cn/20210204143816961.png?
[20210204145133453.png]: https://img-blog.csdnimg.cn/20210204145133453.png
[20210204150103382.png]: https://img-blog.csdnimg.cn/20210204150103382.png
[20210204154806430.png]: https://img-blog.csdnimg.cn/20210204154806430.png
[2021020416154241.png]: https://img-blog.csdnimg.cn/2021020416154241.png
[20210204161630307.png]: https://img-blog.csdnimg.cn/20210204161630307.png
[20210204155726436.png]: https://img-blog.csdnimg.cn/20210204155726436.png
[20210204162048916.png]: https://img-blog.csdnimg.cn/20210204162048916.png
[2021020416095454.png]: https://img-blog.csdnimg.cn/2021020416095454.png
[20210204164116529.png]: https://img-blog.csdnimg.cn/20210204164116529.png
[20210204164547949.png]: https://img-blog.csdnimg.cn/20210204164547949.png
[20210205113627613.png]: https://img-blog.csdnimg.cn/20210205113627613.png
[20210205114054101.png]: https://img-blog.csdnimg.cn/20210205114054101.png
[20210205114437251.png]: https://img-blog.csdnimg.cn/20210205114437251.png
[20210205120448593.png]: https://img-blog.csdnimg.cn/20210205120448593.png