全面解剖 消息中间件 RocketMQ-（4）

一、RocketMQ 顺序消息分析

1、消息有序：指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ 可以严格的保证消息有序，可以分为分区有序或者全局有序。

2、顺序消费的原理解析

在默认的情况下消息发送会采取 Round Robin 轮询方式把消息发送到不同的 queue (分区队列)，而消费消息的时候从多个 queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个 queue 中，消费的时候只从这个 queue 上依次拉取，则就保证了顺序。当发送和消费参与的 queue 只有一个，则是全局有序;如果多个 queue 参与，则为分区有序，即相对每个 queue，消息都是有序的。

3、下面用订单进行分区有序的示例。一个订单的顺序流程是:创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中，消费时，同一个 Orderld 获取到的肯定是同一个队列。

二、RocketMQ 顺序消息发送者

1、在工程 rocketmq_demo （模块）中，创建 订单构建 实体类 OrderStep.java

/***  D:\java-test\idea2019\rocketmq_demo\src\main\java\djh\it\mq\rocketmq\order\OrderStep.java**  2024-6-2 创建 订单构建 实体类 OrderStep.java*/package djh.it.mq.rocketmq.order;import java.util.ArrayList;
import java.util.List;public class OrderStep {private long orderId;  //订单 idprivate 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;}@Overridepublic 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;}
}

2、在工程 rocketmq_demo （模块）中，创建 顺序消息发送 类 Producer.java

/***   rocketmq_demo\src\main\java\djh\it\mq\rocketmq\order\Producer.java**   2024-6-2  创建 顺序消息发送 类 Producer.java*/
package djh.it.mq.rocketmq.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 {//1.创建消息生产者 producer，并制定生产者组名DefaultMQProducer producer = new DefaultMQProducer("group1");//2.指定 Nameserver 地址（集群配置）//producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//2.指定 Nameserver 地址（非集群配置）producer.setNamesrvAddr("172.18.30.110:9876");//3.启动 producerproducer.start();//构建消息集合List<OrderStep> orderSteps = OrderStep.buildOrders();//发送消息for(int i=0; i<orderSteps.size(); i++){String body = orderSteps.get(i)+"";//4.创建消息对象，指定主题 Topic、Tag 和消息体//参数一：消息对象, 参数二：消息队列选择器, 参数三：选择队列的业务标识（订单id）Message message = new Message("OrderTopic", "Order", "i"+i, body.getBytes());//5.发送 异步 消息SendResult sendResult = producer.send(message, new MessageQueueSelector(){/**** @param mqs ：队列集合* @param msg ：消息对象				 ** @param arg ：业务标识的参数* @return*/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);}//6.关闭生产者 producer。producer.shutdown();  	   }
}

三、RocketMQ 顺序消息消费者

1、在工程 rocketmq_demo （模块）中，创建 顺序消息消费 类 Consumer.java

/***   rocketmq_demo\src\main\java\djh\it\mq\rocketmq\order\Consumer.java**   2024-6-2  创建 顺序消息消费 类 Consumer.java 。*/
package djh.it.mq.rocketmq.order;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
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，制定消费者组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2.指定 Nameserver 地址（集群配置）//consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//2.指定 Nameserver 地址（非集群配置）consumer.setNamesrvAddr("172.18.30.110:9876");//3.订阅主题 Topic 和 Tagconsumer.subscribe("OrderTopic", "*");  //接收所有消息。//4.注册消息监听器consumer.registerMessageListener(new MessageListenerOrderly() {public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {for(MessageExt msg : msgs){System.out.println("线程名称：【"+Thread.currentThread().getName() + "】 消费消息：" + new String(msg.getBody()));  //转换为字符串消息}return ConsumeOrderlyStatus.SUCCESS;}});//5.启动消费者 consumer。consumer.start();System.out.println("消费消息启动了");}
}

2、先启动 顺序消息发送 类 Producer.java，再启动 顺序消息消费 类 Consumer.java 进行测试。

四、RocketMQ 延迟消息

1、RocketMQ 延迟消息

比如电商里，提交了一个订单就可以发送一个延时消息，1h 后去检查这个订单的状态，
如果还是未付款就取消订单释放库存。

2、RocketMQ 延迟消息 使用限制

//org/apache/rocketmg/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”.

现在 RocketMg 并不支持任意时间的延时，需要设置几个固定的延时等级，从 1s 到 2h 分别对应着等级 1 到 18。

3、在工程 rocketmq_demo （模块）中，创建 延迟消息 发送 类 Producer.java

/***   rocketmq_demo\src\main\java\djh\it\mq\rocketmq\delay\Producer.java**   2024-6-2  创建 延迟消息 发送 类 Producer.java*/
package djh.it.mq.rocketmq.delay;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 {//1.创建消息生产者 producer，并制定生产者组名DefaultMQProducer producer = new DefaultMQProducer("group1");//2.指定 Nameserver 地址（集群配置）//producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//2.指定 Nameserver 地址（非集群配置）producer.setNamesrvAddr("172.18.30.110:9876");//3.启动 producerproducer.start();//发送消息for(int i=0; i<10; i++){//4.创建消息对象，指定主题 Topic、Tag 和消息体//参数一：消息主题 Topic, 参数二：消息 Tag, 参数三：消息内容Message msg = new Message("DelayTopic", "Tag1", ("Hello World"+i).getBytes());//设定延迟发送 时间为 5 秒（目前 rocketmq 支持的延迟等级："1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h".）msg.setDelayTimeLevel(2);//5.发送消息SendResult result = producer.send(msg);System.out.println("发送结果："+result);TimeUnit.SECONDS.sleep(1);  //线程睡1秒}//6.关闭生产者 producer。producer.shutdown();}
}

4、在工程 rocketmq_demo （模块）中，创建 延迟消息 消费 类 Consumer.java 。

/***   rocketmq_demo\src\main\java\djh\it\mq\rocketmq\delay\Consumer.java**   2024-6-2  创建 延迟消息 消费 类 Consumer.java 。*/
package djh.it.mq.rocketmq.delay;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
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，制定消费者组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2.指定 Nameserver 地址（集群配置）//consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//2.指定 Nameserver 地址（非集群配置）consumer.setNamesrvAddr("172.18.30.110:9876");//3.订阅主题 Topic 和 Tag//consumer.subscribe("base", "Tag1");  //接收同步消息//consumer.subscribe("base", "Tag2");  //接收异步消息前，可以让先发送异步消息。//consumer.subscribe("base", "Tag1 | Tag2");  //接收同步消息 和 异步消息consumer.subscribe("DelayTopic", "*");  //接收所有消息。//添加消费模式//consumer.setMessageModel(MessageModel.CLUSTERING);  //默认是负载均衡模式消费consumer.setMessageModel(MessageModel.BROADCASTING);  //广播模式消费//4.设置回调函数，处理消息consumer.registerMessageListener(new MessageListenerConcurrently(){//接受消息内容public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){//System.out.println(msgs); //接收到的消息是未转换的字节码for(MessageExt msg : msgs){System.out.println("消息ID：【" + msg.getMsgId()+"】,延迟时间："+(System.currentTimeMillis()-msg.getStoreTimestamp()));  //转换为字符串消息}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//5.启动消费者 consumer。consumer.start();System.out.println("消费者启动");}
}

5、先启动 延迟消息 消费 类 Consumer.java 再启动 延迟消息 发送 类 Producer.java 进行测试。

五、RocketMQ 批量消息发送

1、批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的 topic，相同的 waitstoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过無 4MB。如果消息的总长度可能大于4MB时，这时候最好把消息进行分割。

2、在工程 rocketmq_demo （模块）中，创建 批量消息发送 发送 类 Producer.java

/***   rocketmq_demo\src\main\java\djh\it\mq\rocketmq\batch\Producer.java**   2024-6-2  创建 批量消息发送的 发送 类 Producer.java*/
package djh.it.mq.rocketmq.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 {//1.创建消息生产者 producer，并制定生产者组名DefaultMQProducer producer = new DefaultMQProducer("group1");//2.指定 Nameserver 地址（集群配置）//producer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//2.指定 Nameserver 地址（非集群配置）producer.setNamesrvAddr("172.18.30.110:9876");//3.启动 producerproducer.start();//创建一个集合List<Message> msgs = new ArrayList<Message>();//发送消息//4.创建消息对象，指定主题 Topic、Tag 和消息体//参数一：消息主题 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);//5.发送消息SendResult result = producer.send(msgs);System.out.println("发送结果："+result);TimeUnit.SECONDS.sleep(1);  //线程睡1秒//6.关闭生产者 producer。producer.shutdown();}
}

3、在工程 rocketmq_demo （模块）中，创建 批量消息发送 消费 类 Consumer.java 。

/***   rocketmq_demo\src\main\java\djh\it\mq\rocketmq\batch\Consumer.java**   2024-6-2  创建 批量消息发送 消费 类 Consumer.java 。*/
package djh.it.mq.rocketmq.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 org.apache.rocketmq.common.protocol.heartbeat.MessageModel;import java.util.List;public class Consumer {public static void main(String[] args) throws Exception {//1.创建消费者 Consumer，制定消费者组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2.指定 Nameserver 地址（集群配置）//consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//2.指定 Nameserver 地址（非集群配置）consumer.setNamesrvAddr("172.18.30.110:9876");//3.订阅主题 Topic 和 Tagconsumer.subscribe("BatchTopic", "*");  //接收所有消息。//4.设置回调函数，处理消息consumer.registerMessageListener(new MessageListenerConcurrently(){//接受消息内容public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){//System.out.println(msgs); //接收到的消息是未转换的字节码for(MessageExt msg : msgs){System.out.println("consumeThread=" + Thread.currentThread().getName()+", "+new String(msg.getBody()));  //转换为字符串消息}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//5.启动消费者 consumer。consumer.start();System.out.println("消费者启动");}
}

4、先启动 批量消息发送 消费 类 Consumer.java 再启动 批量消息发送 发送 类 Producer.java 进行测试。