RocketMQ 的介绍 RocketMQ 作为目前主流的消息中间件之一, 而消息队列主要应用于以下场景：

• 异步（不需要同步等待）
• 解耦（应用之间不互相依赖）
• 削峰（避免流量激增导致系统性能问题）

RocketMQ 具有以下特性：

• 消息的订阅与发布（消息队列的基本功能）
• 顺序消息（消费的顺序与发送的顺序一致，包括全局有序和分区有序，全局有序的topic只有一个消息队列，适用场景：性能要求不高，所有消息需要严格有序。分区有序的所有消息根据 sharding key进行分区。同一个分区内的消息按照 FIFO 顺序进行发布和消费。Sharding key 是消息中用来区分不同分区的关键字段，和普通消息的 Key 是完全不同的概念。 适用场景：性能要求高，对于某一类消息需要有序，同时有一个sharding key 作为分区字段。）
• 定时消息（消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。）
• 事务消息（应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。通过事务消息能达到分布式事务的最终一致。）
• 消息重试（消费者消费失败时，RocketMQ 提供重试机制使得消息可以被再次消费。）
• 流量控制（当Broker处理消息的能力达到瓶颈时，通过拒绝生产者的发送请求进行流量控制，当消费者的消息能力达到瓶颈时，通过降低消息的拉取频率进行流量控制）
• 死信队列（当一条消息达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。）

RocketMQ 的概念

• 消息（MESSAGE）
  系统之间互相传递的消息，每个消息都属于某一个主题，每个消息使用唯一的Message ID进行标识，同时消息可以带有标签（TAG）和 键（KEY）。
• 标签(TAG)
  每条消息可以携带标签，用于同一主题下区分不同类型的消息。
• 键（KEY）
  除了标签，RocketMQ的消息还可以带上KEY，可以有多个KEY。
• 主题(TOPIC)
  消息对应的主题，用于表示消息的类别，一个主题可以对应多条消息，生产者生产消息时需要制定主题，消费者消费消息时也需要制定主题
• 生产者（PRODUCER）
  发送消息的应用称为生产者。
• 生产者组(PRODUCER GROUP)
  同一类的生产者的集合，每个生存者组包含多个生产者。
• 消费者(CONSUMER)
  消费消息的应用称为消费者。
• 消费者组(CONSUMER GROUP)
  同一类的消费者的集合，每个消费者组包含多个消费者。
• 代理服务器(BROKER)
  消息中转角色，负责存储消息、转发消息。接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。
• 名字服务器(NAME SERVER)
  类似于注册中心，消费者和生产者的注册和发现都依赖于名字服务器，同时会保存broker的信息， 生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。
• 消费者获取消息的方式：
  
  • 拉取式消费
    应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。
  • 推动式消费
    该模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性较高。
• 消费模式：
  
  • 集群消费
    集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。
  • 广播消费
    广播消费模式下，相同Consumer Group的每个Consumer实例都接收全量的消息。

RocketMQ 的架构设计 整体架构
整个RocketMQ 的架构如下：

文章图片

可以整个架构由四部分组成，分别是Producer, Conusmer, Name Server, Broker。
整个RocketMQ 的工作流程如下：

1. 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
2. Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
3. 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
5. Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

消息存储

文章图片

• CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G, 文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；
• ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；
• IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：$HOME \store\index${fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

RocketMQ采用的是混合型的存储结构，Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。RocketMQ使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。
消息过滤
前面有提到Consumer端订阅消息是通过ConsumeQueue拿到消息的索引，然后再从CommitLog里面读取真正的消息实体内容，ConsumeQueue的存储结构如下，可以看到其中有8个字节存储的Message Tag的哈希值，基于Tag的消息过滤正是基于这个字段值的。
RocketMQ 支持两种消息过滤方式：

• Tag过滤方式：Consumer端在订阅消息时除了指定Topic还可以指定TAG，如果一个消息有多个TAG，可以用||分隔。其中，Consumer端会将这个订阅请求构建成一个 SubscriptionData，发送一个Pull消息的请求给Broker端。Broker端从RocketMQ的文件存储层—Store读取数据之前，会用这些数据先构建一个MessageFilter，然后传给Store。Store从 ConsumeQueue读取到一条记录后，会用它记录的消息tag hash值去做过滤，由于在服务端只是根据hashcode进行判断，无法精确对tag原始字符串进行过滤，故在消息消费端拉取到消息后，还需要对消息的原始tag字符串进行比对，如果不同，则丢弃该消息，不进行消息消费 （如果存在tag的hashcode一致，则可能导致丢失消息）。
• SQL92的过滤方式：这种方式的大致做法和上面的Tag过滤方式一样，只是在Store层的具体过滤过程不太一样，真正的 SQL expression 的构建和执行由rocketmq-filter模块负责的。每次过滤都去执行SQL表达式会影响效率，所以RocketMQ使用了BloomFilter避免了每次都去执行。SQL92的表达式上下文为消息的属性。

消息查询
RocketMQ支持两种方式查询：

• 按照Message Id查询消息。
  RocketMQ中的MessageId的长度总共有16字节，其中包含了消息存储主机地址（IP地址和端口），消息Commit Log offset。“按照MessageId查询消息”在RocketMQ中具体做法是：Client端从MessageId中解析出Broker的地址（IP地址和端口）和Commit Log的偏移地址后封装成一个RPC请求后通过Remoting通信层发送（业务请求码：VIEW_MESSAGE_BY_ID）。Broker端走的是QueryMessageProcessor，读取消息的过程用其中的 commitLog offset 和 size 去 commitLog 中找到真正的记录并解析成一个完整的消息返回。
• 按照Message Key查询消息。
  “按照Message Key查询消息”，主要是基于RocketMQ的IndexFile索引文件来实现的。RocketMQ的索引文件逻辑结构，类似JDK中HashMap的实现。索引文件的具体结构如下：
  
  文章图片
  
  IndexFile索引文件为用户提供通过“按照Message Key查询消息”的消息索引查询服务。如果消息的properties中设置了UNIQ_KEY这个属性，就用 topic + “#” + UNIQ_KEY的value作为 key 来做写入操作。如果消息设置了KEYS属性（多个KEY以空格分隔），也会用 topic + “#” + KEY 来做索引。其中的索引数据包含了Key Hash/CommitLog Offset/Timestamp/Next Index offset 这四个字段，一共20 Byte。SlotTable中保存每个Slot对应的链表的头指针，NextIndexOffset 保存的是下一个节点的地址。
  按照Message Key查询消息”的方式，RocketMQ的具体做法是，主要通过Broker端的QueryMessageProcessor业务处理器来查询，读取消息的过程就是用topic和key找到IndexFile索引文件中的一条记录，根据其中的commitLog offset从CommitLog文件中读取消息的实体内容。

事务消息
RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示。

文章图片

上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。

1. 事务消息发送及提交：
   (1) 发送消息（half消息）。
   (2) 服务端响应消息写入结果。
   (3) 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。
   (4) 根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）
2. 补偿流程：
   (1) 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”
   (2) Producer收到回查消息，检查回查消息对应的本地事务的状态
   (3) 根据本地事务状态，重新Commit或者Rollback
   其中，补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

事务消息的实现

1. 第一阶段的消息写入
   写入的如果事务消息，将消息的Topic替换为RMQ_SYS_TRANS_HALF_TOPIC，同时将原来的Topic和Queue信息存储到消息的属性中，正因为消息主题被替换，故消息并不会转发到该原主题的消息消费队列，消费者无法感知消息的存在，不会消费。同时RocketMQ会开启一个定时任务，从Topic为RMQ_SYS_TRANS_HALF_TOPIC中拉取消息进行消费，根据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息。需要注意的是此时半消息是Pending状态的。
2. 第二阶段的消息写入
   因为消息是顺序写入ConmmitLog的，无法做到删除一个消息，为了标志半消息的状态，引入了Op消息来标志是Commit状态还是Rollback状态。Op消息里面保存了消息的一些基本信息，基本信息里最主要的就是消息的Offset。通过Offset可以索引到具体的消息。
   RocketMQ将Op消息写入到全局一个特定的Topic中通过源码中的方法—TransactionalMessageUtil.buildOpTopic()；这个Topic是一个内部的Topic（像Half消息的Topic一样），不会被用户消费。Op消息的内容为对应的Half消息的存储的Offset，这样通过Op消息能索引到Half消息进行后续的回查操作。
   
   文章图片
   
   第二阶段如果是Rollback操作，则通过发送一条Op消息标志该消息的状态为Rollback，同时表明该消息已经处理过了。如果第二阶段为Commit的操作，则需要发送一条Op消息标志该消息的状态为Commit,同时需要将半消息读出，并恢复出一条完整的普通消息进行发送。
3. 回查消息
   如果在RocketMQ事务消息的二阶段过程中失败了，例如在做Commit操作时，出现网络问题导致Commit失败，那么需要通过一定的策略使这条消息最终被Commit。RocketMQ采用了一种补偿机制，称为“回查”。Broker端对未确定状态的消息发起回查，将消息发送到对应的Producer端（同一个Group的Producer），由Producer根据消息来检查本地事务的状态，进而执行Commit或者Rollback。Broker端通过对比Half消息和Op消息进行事务消息的回查并且推进CheckPoint（记录那些事务消息的状态是确定的）。
   值得注意的是，rocketmq并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，rocketmq默认回滚该消息。

负载均衡
RocketMQ 的负载均衡都是在Client端完成的，分为生产者的负载均衡和消费者的负载均衡。

• 生产者的负载均衡
  Producer端在发送消息的时候，会先根据Topic找到指定的TopicPublishInfo，在获取了TopicPublishInfo路由信息后，RocketMQ的客户端在默认方式下selectOneMessageQueue()方法会从TopicPublishInfo中的messageQueueList中选择一个队列（MessageQueue）进行发送消息。具体的容错策略均在MQFaultStrategy这个类中定义。
• 消费者的负载均衡
  在RocketMQ中，Consumer端的两种消费模式（Push/Pull）都是基于拉模式来获取消息的，而在Push模式只是对pull模式的一种封装，其本质实现为消息拉取线程在从服务器拉取到一批消息后，然后提交到消息消费线程池后，又“马不停蹄”的继续向服务器再次尝试拉取消息。如果未拉取到消息，则延迟一下又继续拉取。在两种基于拉模式的消费方式（Push/Pull）中，均需要Consumer端在知道从Broker端的哪一个消息队列—队列中去获取消息。
  集群模式下的具体的步骤如下：
  
  1. Conumser 向broker发送心跳包（包含了消息消费分组名称、订阅关系集合、消息通信模式和客户端id的值等信息），Broker端在收到Consumer的心跳消息后，会将它维护在ConsumerManager的本地缓存变量—consumerTable，同时并将封装后的客户端网络通道信息保存在本地缓存变量—channelInfoTable中。
  2. 启动负载均衡服务线程RebalanceService， 首先从rebalanceImpl实例的本地缓存变量—topicSubscribeInfoTable中，获取该Topic主题下的消息消费队列集合（mqSet），然后根据topic和consumerGroup为参数调用mQClientFactory.findConsumerIdList()方法向Broker端发送获取该消费组下消费者Id列表的RPC通信请求。对Topic下的消息消费队列、消费者Id排序，然后用消息队列分配策略算法（默认为：消息队列的平均分配算法），计算出待拉取的消息队列，最后根据分配队列的结果更新ProccessQueueTable，尝试删除不在待拉去的消息队列中的队列，或者在分配中的队列中但是已经过期的队列。为过滤后的消息队列集合（mqSet）中的每个MessageQueue创建一个ProcessQueue对象并存入RebalanceImpl的processQueueTable队列中，根据processQueueTable构建长轮询对象PullRequest对象，会从broker获取消费的进度。

RocketMQ 的最佳实践 顺序消息
FIFOProducer.java

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.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; public class FIFOProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("FIFOProducerGroup"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start(); String[] tags = new String[]{"TagA", "TagC", "TagD"}; // 订单列表 List orderList = new FIFOProducer().buildOrders(); Date date = new Date(); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); String dateStr = sdf.format(date); for (int i = 0; i < 10; i++) { // 加个时间前缀 String body = dateStr + " Hello RocketMQ " + orderList.get(i); Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes()); SendResult sendResult = producer.send(msg, new MessageQueueSelector() { @Override public MessageQueue select(List mqs, Message msg, Object arg) { Long id = (Long) arg; //根据订单id选择发送queue long index = id % mqs.size(); return mqs.get((int) index); } }, orderList.get(i).getOrderId()); //订单idSystem.out.println(String.format("SendResult status:%s, queueId:%d, body:%s", sendResult.getSendStatus(), sendResult.getMessageQueue().getQueueId(), body)); }producer.shutdown(); }/** * 订单的步骤 */ private static 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 + '\'' + '}'; } }/** * 生成模拟订单数据 */ private List buildOrders() { List orderList = new ArrayList(); OrderStep orderDemo = new OrderStep(); orderDemo.setOrderId(15103111039L); orderDemo.setDesc("创建"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103111065L); orderDemo.setDesc("创建"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103111039L); orderDemo.setDesc("付款"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103117235L); orderDemo.setDesc("创建"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103111065L); orderDemo.setDesc("付款"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103117235L); orderDemo.setDesc("付款"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103111065L); orderDemo.setDesc("完成"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103111039L); orderDemo.setDesc("推送"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103117235L); orderDemo.setDesc("完成"); orderList.add(orderDemo); orderDemo = new OrderStep(); orderDemo.setOrderId(15103111039L); orderDemo.setDesc("完成"); orderList.add(orderDemo); return orderList; } }

FIFOConsumer.java

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.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; import java.util.List; import java.util.Random; import java.util.concurrent.TimeUnit; public class FIFOConsumer { public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("FIFOConsumerGroup"); consumer.setNamesrvAddr("127.0.0.1:9876"); /** * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
* 如果非第一次启动，那么按照上次消费的位置继续消费 */ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); consumer.subscribe("TopicTest", "TagA || TagC || TagD"); consumer.registerMessageListener(new MessageListenerOrderly() {Random random = new Random(); @Override public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) { context.setAutoCommit(true); for (MessageExt msg : msgs) { // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序 System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody())); }try { //模拟业务逻辑处理中... TimeUnit.SECONDS.sleep(random.nextInt(10)); } catch (Exception e) { e.printStackTrace(); } return ConsumeOrderlyStatus.SUCCESS; } }); consumer.start(); System.out.println("Consumer Started."); } }

事务消息
TransactionProducer.java

import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.TransactionListener; import org.apache.rocketmq.client.producer.TransactionMQProducer; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; import java.io.UnsupportedEncodingException; import java.util.concurrent.*; public class TransactionProducer { public static void main(String[] args) throws MQClientException, InterruptedException { TransactionListener transactionListener = new TransactionListenerImpl(); TransactionMQProducer producer = new TransactionMQProducer("TransactionProducerGroup"); producer.setNamesrvAddr("127.0.0.1:9876"); ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName("client-transaction-msg-check-thread"); return thread; } }); producer.setExecutorService(executorService); producer.setTransactionListener(transactionListener); producer.start(); String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"}; for (int i = 0; i < 10; i++) { try { Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, ("Hello Luo " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult = producer.sendMessageInTransaction(msg, null); System.out.printf("%s%n", sendResult); Thread.sleep(10); } catch (MQClientException | UnsupportedEncodingException e) { e.printStackTrace(); } } for (int i = 0; i < 100000; i++) { Thread.sleep(1000); } producer.shutdown(); } }

TransactionListenerImpl.java

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 java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; public class TransactionListenerImpl implements TransactionListener { private AtomicInteger transactionIndex = new AtomicInteger(0); private ConcurrentHashMap localTrans = new ConcurrentHashMap<>(); @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { int value = https://www.it610.com/article/transactionIndex.getAndIncrement(); int status = value % 3; localTrans.put(msg.getTransactionId(), status); return LocalTransactionState.UNKNOW; } @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { Integer status = localTrans.get(msg.getTransactionId()); if (null != status) { switch (status) { case 0: return LocalTransactionState.UNKNOW; case 1: return LocalTransactionState.COMMIT_MESSAGE; case 2: return LocalTransactionState.ROLLBACK_MESSAGE; } } return LocalTransactionState.COMMIT_MESSAGE; } }

RocketMQ 的使用建议

• producer使用建议
  
  1. 请控制producer实例数，过多的连接会造成server端的资源浪费；
  2. 请给 producer 设置优雅关闭(producer.shutdown())，服务关闭可以立即释放资源
  3. 接入多个rocketmq集群，为防止nameserver覆盖问题，需要设置InstanceName
• consumer使用建议
  
  1. 尽量使用push consumer
  2. 接入多个rocketmq集群，为防止nameserver覆盖问题，需要设置InstanceName
  3. 当使用TAG的时候，避免同一个consumerGroup 的Consumer 使用的TAG 的类型不同。

参考 【RocketMQ使用指南】RocketMQ 官方文档

推荐阅读

• 

  mysql实现排名 mysqlsql排名函数怎么用
• 

  大豆蛋白被是什么材料
• 

  金庸的武侠代表作品是什么
• 

  技嘉主板超频内存 技嘉Z390主板超频指南
• 

  2017年出的关于现代校园恋爱的动漫,有没有什么推荐？
• 

  新鲜桂花怎么处理和保存
• 

  为什么iPhone手机不踏足折叠屏市场？或许离不开这五点原因
• 

  SQL|SQL 导出表结构到Excel
• 

  N95 12月25日起赣榆区投放8万个医用防护口罩
• 

  什么条件下才能领取失业保险金?不领取还是领好?
• 

  出现菲斯曼锅炉显示0a故障怎么办,故障原因说明与3种解决方法
• 

  定金合同在什么情况下无效 有哪些情况定金合同无效
• 

  大金空调打台风后不制冷故障排除图解,主要还是受气温的影响
• 

  40岁自学go语言 零基础学go语言多久
• 

  办理加油卡有什么好处 加油卡怎么办理划算
• 

  这样搭配更具有视觉美观性 浅蓝色t恤配什么颜色裤子好看
• 

  夏天什么时候种白菜 什么时候种白菜
• 

  Win11如何禁用资源管理器历史搜索？Win11禁用资源管理器历史搜索的方法
• 

  特仑苏纯牛奶多少钱一瓶 「特仑苏纯牛奶两种包装」
• 

  三菱空调定时闪烁不制冷是什么原因,辅助加热

• iPhone手机这些使用指南|iPhone手机这些使用指南 你知道多少（）
• 2018-03-28|2018-03-28 搭建RocketMQ踩的坑。。。
• Android关于BottomNavigationView使用指南
• RocketMQ学习十-消息存储
• 2.|2. RocketMQ进阶1_架构与集群部署模式
• RocketMQ学习八-NameServer分析
• RocketMQ学习七-Netty的应用
• RocketMQ学习六-消息发送错误与解决方案
• RocketMQ学习五-选择队列等特性
• RocketMQ - 理论篇