rocketmq

🍓代码仓库

https://gitee.com/xuhx615/rocket-mqdemo.git

🍓基本概念

• ⭐生产者(Producer)：消息发布者
• ⭐主题（Topic）：topic用于标识同一类业务类型的消息
• ⭐消息队列（MessageQueue）：传输和存储消息的容器，是消息的最小存储单元
• ⭐消费者（Consumer）：消息订阅者
• ⭐消费者组（ConsumerGroup）：消息订阅者组，多个消费者之间进行负载均衡消费消息
• ⭐nameServer：注册中心
• ⭐Broker：消息中转站，用于接收生产者的消息并持久化，然后发送给对应的topic

🍓下载安装rocketmq

1. ⭐前往官网https://rocketmq.apache.org下载rocketmq安装包和rocketmq图形化界面rocketmq Dashboard
2. ⭐解压rocketmq安装包

   [root@Centos101 rocketmq]# unzip rocketmq-all-5.1.3-bin-release.zip
   

3. ⭐修改namserver启动脚本runserver的JVM内存参数（根据实际服务器资源设置，以下参数为学习时设置的参数）

    [root@Centos101 bin]# vi runserver.sh修改前：JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"修改后：JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
   

4. ⭐启动nameserver

    [root@Centos101 bin]# ./mqnamesrv &
   

5. ⭐查看nameserver启动日志

   [root@Centos101 bin]# tail -100f nohup.out
   

6. ⭐修改broker启动脚本runboker的JVM内存参数

   [root@Centos101 bin]# vi runbroker.sh
   修改前：JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g"
   修改后：JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m"
   

7. ⭐修改broker.conf配置文件

   默认配置
   #集群名称
   brokerClusterName = DefaultCluster
   #broker名称
   brokerName = broker-a
   #当前节点为主节点（主节点为0）
   brokerId = 0
   deleteWhen = 04
   fileReservedTime = 48
   brokerRole = ASYNC_MASTER
   flushDiskType = ASYNC_FLUSH新增以下配置
   #自动创建topic
   autoCreateTopicEnable = true
   #namesrvAddr地址
   namesrvAddr = 192.168.113.101:9876
   

8. ⭐启动broker

   [root@Centos101 bin]# ./mqbroker -c ../conf/broker.conf &
   

9. ⭐验证

   生产者：[root@Centos101 bin]# export NAMESRV_ADDR='192.168.113.101:9876'[root@Centos101 bin]# ./tools.sh org.apache.rocketmq.example.quickstart.Producer
   消费者：[root@Centos101 bin]# export NAMESRV_ADDR='192.168.113.101:9876'[root@Centos101 bin]# ./tools.sh org.apache.rocketmq.example.quickstart.Consumer
   

10. ⭐关闭broker

    [root@Centos101 bin]# sh ./mqshutdown broker
    

11. ⭐关闭nameserver

    [root@Centos101 bin]# sh ./mqshutdown namesrv
    

🍓rocketmq集群安装

1. ⭐主机名配置

   192.168.113.101 Centos101
   192.168.113.102 Centos102
   192.168.113.103 Centos103
   

2. ⭐免密登录

3. ⭐关闭防火墙

4. ⭐配置文件配置：

   • 📌2m-2s-async：2主2从异步刷盘（吞吐量较大，但消息可能会丢失）当生产者发送消息到主节点，主节点会直接给生产返回收到消息，然后异步同步给从节点
   • 📌2m-2s-sync：2主2从同步刷盘（吞吐量会下降，但消息会更安全）当生产者发送消息到主节点，主节点会同步同步给从节点，然后才给生产者返回收到消息
   • 📌2m-noslave：2主无从（单点故障），然后还可以直接配置broker.conf，进行单点环境配置
   • 📌集群搭建架构

        Centos101：部署nameserverCentos102：部署nameserver  broker-a,broker-b-sCentos103：部署nameserver  broker-b,broker-a-s
     

5. ⭐集群启动

   • 📌nameserver服务启动
     分别在三个机器上启动nameserver

     [root@Centos101 bin]# ./mqnamesrv &
     [root@Centos102 bin]# ./mqnamesrv &
     [root@Centos103 bin]# ./mqnamesrv &
     

   • 📌broker服务启动

     在Centos102机器上启动broker（broker-a主节点和broker-b-s从节点）
     [root@Centos102 bin]# ./mqbroker -c ../conf/2m-2s-async/broker-a.properties &
     [root@Centos102 bin]# ./mqbroker -c ../conf/2m-2s-async/broker-b-s.properties &
     在Centos103机器上启动broker（broker-b主节点和broker-a-s从节点）
     [root@Centos103 bin]# ./mqbroker -c ../conf/2m-2s-async/broker-b.properties &
     [root@Centos103 bin]# ./mqbroker -c ../conf/2m-2s-async/broker-a-s.properties &
     

   • 📌验证

     在Centos102上模拟生产者
     [root@Centos102 bin]# export NAMESRV_ADDR='Centos101:9876;Centos102:9876;Centos103:9876'
     [root@Centos102 bin]# ./tools.sh org.apache.rocketmq.example.quickstart.Producer
     在Centos103上模拟消费者
     [root@Centos103 bin]# export NAMESRV_ADDR='Centos101:9876;Centos102:9876;Centos103:9876'
     [root@Centos103 bin]# ./tools.sh org.apache.rocketmq.example.quickstart.Consumer
     

🍓安装rocketmq图形化管理界面：dashboard

修改application.properties
rocketmq.config.namesrvAddr=Centos101:9876;Centos102:9876;Centos103:9876
修改logback.xml日志路径

🍓rocketmq的local模式启动，新增proxy模块（5.0后支持的模块）

引入 Proxy 模块后，Proxy 承担了协议适配、权限管理、消息管理等计算功能，Broker 则更加专注于存储。这样存储和计算相分离，在云原生环境下可以更好地进行资源调度。

[root@Centos102 bin]# ./mqbroker -c ../conf/2m-2s-async/broker-a.properties --enable-proxy &
[root@Centos103 bin]# ./mqbroker -c ../conf/2m-2s-async/broker-b.properties --enable-proxy &

🍓部署模型

🍓消息发送过程

🍓消息存储过程

🍓生产者

生产者分为同步生产者和异步生产者以及单项生产者

• ⭐同步生产者：生产者将消息推送Broker，等待Broker返回推送确认，再推送下一个
  1、可靠性要求高
  2、数据量级少
  3、实时响应

  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;DefaultMQProducer producer = null;try {producer = new DefaultMQProducer("syncProducer");producer.setNamesrvAddr("192.168.113.101:9876");producer.start();for (int i = 0; i < 2; i++) {String body = "Hello zhang " + i;//参数一：主题、参数二：过滤、参数三：消息内容Message message = new Message("rocketmq_syncDemo","tag", body.getBytes("UTF-8"));//同步发送SendResult result = producer.send(message);String msgId = result.getMsgId();SendStatus sendStatus = result.getSendStatus();logger.info("{}消息发送状态为{}", msgId, sendStatus);}} catch (Exception e) {logger.error("生产者发送消息失败！" ,e);} finally {if (producer != null) {producer.shutdown();}}
  

• ⭐异步生产者：生产者将消息推送Broker，不会等待Broker返回推送确认，直接推送下一个，但是会回调方法告诉生产者消息是否发送成功。

  	try {//异步发送producer.send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {String msgId = sendResult.getMsgId();SendStatus sendStatus = sendResult.getSendStatus();logger.info("{}消息发送状态为{}", msgId, sendStatus);}@Overridepublic void onException(Throwable throwable) {logger.error("消息发送失败", throwable);}});} catch (Exception e) {logger.error("生产者发送消息失败！" ,e);} finally {//异步发送不应该关闭，关闭了便无法回调方法}
  

• ⭐单项生产者：生产者将消息推送Broker，不会等待Broker返回推送确认，直接推送下一个。

  	//单向发送producer.sendOneway(message);
  

🍓消费者

消费者分为推模式和拉模式

• ⭐推模式：消费者等待Broker把消息推送过来（被动消费）

  import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  import org.apache.rocketmq.client.consumer.listener.*;
  import org.apache.rocketmq.client.exception.MQClientException;
  import org.apache.rocketmq.common.message.MessageExt;DefaultMQPushConsumer consumer = null;try {consumer = new DefaultMQPushConsumer("group_rocketmq_syncDemo");consumer.setNamesrvAddr("192.168.113.101:9876");//参数一：topic、参数二：过滤（*表示不过滤）consumer.subscribe("rocketmq_syncDemo", "*");//设置消息监听//MessageListenerConcurrently 并发消费监听consumer.setMessageListener(new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {list.forEach(item -> {try {logger.info("消息消费成功！消息ID={}，消息内容：{}", item.getMsgId(), new String(item.getBody(), "UTF-8"));} catch (Exception e) {logger.error("消息消费失败！", e);}});return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});//消费者启动consumer.start();} catch (MQClientException e) {logger.error("消费者消费异常！",e);}
  

• ⭐拉模式：消费者主动去Broker上拉取消息（主动消费）

  import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
  import org.apache.rocketmq.client.consumer.PullResult;
  import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
  import org.apache.rocketmq.client.exception.MQClientException;
  import org.apache.rocketmq.common.message.MessageQueue;try {DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("group_rocketmq_asyncDemo");consumer.setNamesrvAddr("192.168.113.101:9876");Set<String> topicSet = new HashSet<>();topicSet.add("rocketmq_asyncDemo");consumer.setRegisterTopics(topicSet);consumer.start();//主题遍历while (true) {consumer.getRegisterTopics().forEach(item -> {try {Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(item);//消息队列messageQueues.forEach(item2 -> {try {long offset = consumer.getOffsetStore().readOffset(item2, ReadOffsetType.READ_FROM_MEMORY);if (offset < 0) {offset = consumer.getOffsetStore().readOffset(item2, ReadOffsetType.READ_FROM_STORE);}if (offset < 0) {offset = consumer.maxOffset(item2);}if (offset < 0) {offset = 0;}PullResult result = consumer.pull(item2, "*", offset, 32);if (result != null) {switch (result.getPullStatus()) {case FOUND:{result.getMsgFoundList().forEach(item3 -> {try {logger.info("消息消费成功！消息ID={}，消息内容：{}", item3.getMsgId(), new String(item3.getBody(), "UTF-8"));consumer.updateConsumeOffset(item2, result.getNextBeginOffset());} catch (Exception e) {logger.error("遍历消息信息失败！" , e);}});break;}case NO_NEW_MSG:{logger.info("没有最新消息！");break;}case NO_MATCHED_MSG: {logger.info("没有匹配的消息！");break;}case OFFSET_ILLEGAL: {logger.error("偏移量非法，当前偏移量为{}", offset);break;}}}} catch (Exception e) {logger.error("遍历消息队列失败！", e);}});} catch (MQClientException e) {logger.error("遍历主题失败！", e);}});}} catch (MQClientException e) {logger.error("消息拉取失败！", e);}
  

  • 📌随机获取消息队列消息

    DefaultLitePullConsumer consumer = null;
    try {consumer = new DefaultLitePullConsumer("group_rocketmq_asyncDemo");consumer.setNamesrvAddr("192.168.113.101:9876");consumer.subscribe("rocketmq_asyncDemo", "*");consumer.start();while (true) {List<MessageExt> messageExtList = consumer.poll();messageExtList.forEach(item -> {try {logger.info("获取消息成功！消息队列ID={}，消息ID={}，消息内容{}", item.getQueueId(),item.getMsgId(), new String(item.getBody(), "UTF-8"));} catch (Exception e) {logger.error("获取消息异常！",e);}});}
    } catch (MQClientException e) {logger.error("获取消息异常！",e);
    } finally {if (consumer != null) {consumer.shutdown();}
    }
    

  • 📌指定消息队列获取消息

    	//指定第一个消息队列消费consumer.seek(messageQueueList.get(0), 10);
    

🍓顺序消息

• ⭐生产者需要将有序消息发送到同一个队列

• ⭐消费者push模式，通过加锁的方式，使得一个队列同时只有一个消费者，每隔一段时间就会延长锁的时间（有超时机制），直到整个队列的消息全部消费

• ⭐消费者pull模式，只要消费者自己能保证消息顺序消费就行

• ⭐消费线程数需设置为1

• ⭐生产者代码

  //i 队列序号
  for (int i = 0; i < 5; i++) {//j 消息序号for (int j = 0; j < 100; j++) {Message message = new Message("rocketmq_orderDemo", "tag", ("Hello world!" + j).getBytes("UTF-8"));producer.send(message, new MessageQueueSelector() {/*** * @param list 队列集合* @param message 消息 （send函数第一个参数）* @param o 队列序号 （send函数第三个参数）* @return 消息队列*/@Overridepublic MessageQueue select(List<MessageQueue> list, Message message, Object o) {return list.get(Integer.parseInt(o.toString()));}}, i);}
  }
  

• ⭐消费者代码

  //MessageListenerOrderly有序消息监听（不要使用并发消费监听）
  consumer.setMessageListener(new MessageListenerOrderly() {@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {list.forEach(item -> {try {logger.info("消息接收成功！消息队列={}，消息ID={},消息内容={}", item.getQueueId(), item.getMsgId(), new String(item.getBody(), "UTF-8"));} catch (Exception e) {logger.error("消息接收异常！", e);}});return ConsumeOrderlyStatus.SUCCESS;}
  });
  

🍓广播消息

生产者：

 //设置为广播模式consumer.setMessageModel(MessageModel.BROADCASTING);

🍓延时消息

生产者：

//1-18 对应 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h （消费者会跟生产者设置的时间延迟接收消息）
//message.setDelayTimeLevel(3);//设置自定义时间，单位毫秒
message.setDelayTimeMs(10000L);

🍓批量消息

• ⭐优点：减少网络OA，提高吞吐量

• 限制：

  • 消息大小不能超过4M
  • 相同的topic
  • 相同的waitStoreMsgOk
  • 不能是延迟消息、事务消息等

• ⭐切割消息工具

  /*** 消息集合切割* 消息大小 = 消息长度 + 主题长度 + 消息自定义属性key长度 + 消息自定义属性val长度 + 20（日志空余）* @author xuhaixiang* @date 2023-09-10*/
  public class ListSplitter implements Iterator<List<Message>> {/*** 消息大小限制 1MB*/private static final int SIZE_LIMIT = 10 * 1000;/*** 消息集合*/private final List<Message> messageList;/*** 当前索引*/private int currentIndex;public ListSplitter(List<Message> messageList) {this.messageList = messageList;}@Overridepublic boolean hasNext() {return currentIndex < messageList.size();}@Overridepublic List<Message> next() {int nextIndex = currentIndex;int totalSize = 0;for (; nextIndex < messageList.size(); nextIndex++) {Message message = messageList.get(nextIndex);int messageSize = message.getBody().length + message.getTopic().length();Map<String, String> properties = message.getProperties();for (String key : properties.keySet()) {String val = properties.get(key);messageSize += key.length() + val.length();}messageSize += 20;totalSize += messageSize;if (totalSize > SIZE_LIMIT) {nextIndex = nextIndex - 1;break;}}List<Message> result = messageList.subList(currentIndex, nextIndex);currentIndex = nextIndex;return result;}
  }
  

• ⭐消息批量发送

  List<Message> messages = new ArrayList<>();
  for (int i = 0; i < 2000; i++) {String body = i + "Lorem ipsum dolor sit amet consectetur adipisicing elit. Quasi exercitationem laudantium repellendus quisquam aspernatur est neque quidem vitae nostrum! Quia voluptatibus vitae tempore! Repellendus quam aspernatur, nam neque hic esse!";//参数一：主题、参数二：过滤、参数三：消息内容Message message = new Message("rocketmq_syncDemo","tag", body.getBytes("UTF-8"));messages.add(message);
  }
  ListSplitter listSplitter = new ListSplitter(messages);
  while (listSplitter.hasNext()) {List<Message> messageList = listSplitter.next();SendResult result = producer.send(messageList);String msgId = result.getMsgId();SendStatus sendStatus = result.getSendStatus();logger.info("{}消息发送状态为{}", msgId, sendStatus);
  }
  

🍓过滤消息

• ⭐tag过滤

  • 📌生产者

    String[] tagArr = {"tagA", "tagB", "tagC"};
    for (int i = 0; i < 2; i++) {for (String tag : tagArr) {String body = tag + ", Hello zhang " + i;//参数一：主题、参数二：过滤、参数三：消息内容Message message = new Message("rocketmq_syncDemo",tag, body.getBytes("UTF-8"));SendResult result = producer.send(message);String msgId = result.getMsgId();SendStatus sendStatus = result.getSendStatus();logger.info("{}消息发送状态为{}", msgId, sendStatus);}
    }
    

  • 📌消费者·

    //参数一：topic、参数二：过滤（*表示不过滤），多个tag可以使用||
    consumer.subscribe("rocketmq_syncDemo", "tagA || tagC");
    

• ⭐SQL过滤

  • 📌生产者

    message.putUserProperty("type", "elg_" + i);
    

  • 📌消费者（必须推模式）

     //过滤方式二（注意该sql里面字段是区分大小写的）//sql过滤方式，borker配置文件必须设置属性enablePropertyFilter=true，并且消费者必须是推模式//另外消息过滤行为是在broker端进行的，可以提升网络传输性能，但是会增加服务器的压力（将过滤sql推送给broker）consumer.subscribe("rocketmq_syncDemo", MessageSelector.bySql("TAGS is not null and TAGS in ('tagA','tagC') and type = 'elg_0'"));
    

🍓事务消息

• ⭐事务消息是分布式系统中保证最终一致性的两阶段提交的消息实现。他可以保证本地事务执行与消息发送两个操作的原子性，也就是两个操作一起成功或者一起失败。

• ⭐事务消息机制的关键是在发送消息时会将消息转为一个half消息，并存入rocketmq内部的一个Topic(RMQ_SYS_TRANS_HALF_TOPIC)，这个topic对消费者是不可见的。再经过一系列事务检查通过后，再将消息转存到目标topic，这样消费者就可见了。

• ⭐事务消息实现原理主要通过两个发送阶段和一个确认阶段来实现

• ⭐本地事务消息执行器（本地事务执行和本地事务回查，用于向rocketmq发送提交、回滚、无状态三种结果）

  import org.apache.rocketmq.client.producer.LocalTransactionState;
  import org.apache.rocketmq.client.producer.TransactionListener;
  import org.apache.rocketmq.common.message.Message;
  import org.apache.rocketmq.common.message.MessageExt;
  import org.apache.rocketmq.logging.org.slf4j.Logger;
  import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;/*** 本地事务实现类* @author xuhaixiang* @date 2023-09-11*/
  public class TransactionListenerImpl implements TransactionListener {/*** 日志对象*/private static  final Logger logger = LoggerFactory.getLogger(TransactionListenerImpl.class);/*** 本地事务执行* @param message* @param o* @return*/@Overridepublic LocalTransactionState executeLocalTransaction(Message message, Object o) {String tags = message.getTags();logger.info("{}本地事务执行", tags);if ("tagA".equals(tags)) {//tagA允许发送return LocalTransactionState.COMMIT_MESSAGE;}if ("tagB".equals(tags)) {//tagB消息回滚return LocalTransactionState.ROLLBACK_MESSAGE;}//其他消息无状态，无状态消息会进行本地事务回查return LocalTransactionState.UNKNOW;}/*** 本地事务回查* @param messageExt* @return*/@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt messageExt) {String tags = messageExt.getTags();logger.info("{}本地事务回查", tags);if ("tagC".equals(tags)) {//tagC本地事务回查允许发送return LocalTransactionState.COMMIT_MESSAGE;}return LocalTransactionState.UNKNOW;}
  }
  

• ⭐生产者

  TransactionMQProducer producer = null;
  try {producer = new TransactionMQProducer("transactionProductor");producer.setNamesrvAddr("192.168.113.101:9876");//开启异步线程，用于异步执行本地事务执行和回查两个动作ExecutorService service = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS ,new ArrayBlockingQueue<>(20000), new ThreadFactory(){@Overridepublic Thread newThread(Runnable r) {Thread thread = new Thread(r);thread.setName("transaction");return thread;}});producer.setExecutorService(service);//设置本地事务执行器producer.setTransactionListener(new TransactionListenerImpl());producer.start();String[] tags = {"tagA", "tagB", "tagC", "tagD", "tagE"};for (int i = 0; i < 10; i++) {for (String tag : tags) {Message message = new Message("rocketmq_transactionDemo", tag, (tag + " Hello world!" + i).getBytes("UTF-8"));TransactionSendResult result = producer.sendMessageInTransaction(message, null);logger.info("消息发送成功！消息ID={}" + result.getMsgId());}}//让生产者存活一段时间可以回调本地事务执行和本地事务回查Thread.sleep(60000);
  } catch (Exception e) {logger.error("消息发送异常！", e);
  } finally {if (producer != null) {producer.shutdown();}
  }
  

• ⭐消费者。事务与消费没有任何关系，消费者正常消费消息就行。

🍓如何保证消息不丢失

消息丢失的几种情况：

1. 生产者将消息发送给broker，当网络发生异常，消息可能会丢失
   解决：消息发送后会有ack返回，当我们发现消息发送失败，可以做一个重试机制
2. 消费者拿到消息，会立即发送ack告诉broker收到，但是在接下来处理消息时发生了异常，可能会导致消息丢失，消息无法重新消费
   解决：先处理完消息之后，再返回ack给broker
3. broker存储消息阶段，异步刷盘可能会出现问题导致消息丢失
   解决：使用同步刷盘机制；集群模式采用同步复制

🍓消息持久化机制

rocketmq的消息持久化机制是指将消息存储在磁盘上，以确保消息能够可靠存储和检索
rocketmq消息持久化涉及以下三个角色

• ⭐CommitLog消息存储文件
  • 📌存储方式：
    • 🍁同步刷盘:消息存储到内存，再从内存存储到commitLog，然后返回生产者ack
    • 🍁异步刷盘:消息存储到内存，然后返回生产者ack，再异步存储到commitLog
  • 📌文件固定大小1G，超过则新开辟一个文件
• ⭐ConsumeQueue
  存储commitLog当前读取的偏移量、消息大小、tags值
• ⭐IndexFile
  存储消息自定义的属性、与之对应的消息偏移量、时间参数、下一个Index偏移量

🍓rocketmq保证消息有序

• ⭐生产者需要将有序消息发送到同一个队列
• ⭐消费者push模式，通过加锁的方式，使得一个队列同时只有一个消费者，每隔一段时间就会延长锁的时间（有超时机制），直到整个队列的消息全部消费
• ⭐消费者pull模式，只要消费者自己能保证消息顺序消费就行
• ⭐消费线程数需设置为1