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 的工作流程如下：
- 启动NameServer，NameServer起来后监听端口，期待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
- Broker启动，跟所有的NameServer放弃长连贯，定时发送心跳包。心跳包中蕴含以后Broker信息(IP+端口等)以及存储所有Topic信息。注册胜利后，NameServer集群中就有Topic跟Broker的映射关系。
- 收发音讯前，先创立Topic，创立Topic时须要指定该Topic要存储在哪些Broker上，也能够在发送音讯时主动创立Topic。
- Producer发送音讯，启动时先跟NameServer集群中的其中一台建设长连贯，并从NameServer中获取以后发送的Topic存在哪些Broker上，轮询从队列列表中抉择一个队列，而后与队列所在的Broker建设长连贯从而向Broker发消息。
- 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) 发送音讯（half音讯）。
(2) 服务端响应音讯写入后果。
(3) 依据发送后果执行本地事务（如果写入失败，此时half音讯对业务不可见，本地逻辑不执行）。
(4) 依据本地事务状态执行Commit或者Rollback（Commit操作生成音讯索引，音讯对消费者可见） - 弥补流程：
(1) 对没有Commit/Rollback的事务音讯（pending状态的音讯），从服务端发动一次“回查”
(2) Producer收到回查音讯，查看回查音讯对应的本地事务的状态
(3) 依据本地事务状态，从新Commit或者Rollback
其中，弥补阶段用于解决音讯Commit或者Rollback产生超时或者失败的状况。
事务音讯的实现
- 第一阶段的音讯写入
写入的如果事务音讯，将音讯的Topic替换为RMQ_SYS_TRANS_HALF_TOPIC，同时将原来的Topic和Queue信息存储到音讯的属性中，正因为音讯主题被替换，故音讯并不会转发到该原主题的音讯生产队列，消费者无奈感知音讯的存在，不会生产。同时RocketMQ会开启一个定时工作，从Topic为RMQ_SYS_TRANS_HALF_TOPIC中拉取音讯进行生产，依据生产者组获取一个服务提供者发送回查事务状态申请，依据事务状态来决定是提交或回滚音讯。须要留神的是此时半音讯是Pending状态的。 - 第二阶段的音讯写入
因为音讯是程序写入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,同时须要将半音讯读出，并复原出一条残缺的一般音讯进行发送。 - 回查音讯
如果在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端的哪一个音讯队列—队列中去获取音讯。
集群模式下的具体的步骤如下：- Conumser 向broker发送心跳包（蕴含了音讯生产分组名称、订阅关系汇合、音讯通信模式和客户端id的值等信息），Broker端在收到Consumer的心跳音讯后，会将它保护在ConsumerManager的本地缓存变量—consumerTable，同时并将封装后的客户端网络通道信息保留在本地缓存变量—channelInfoTable中。
- 启动负载平衡服务线程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<OrderStep> 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<MessageQueue> 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());//订单id System.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<OrderStep> buildOrders() { List<OrderStep> orderList = new ArrayList<OrderStep>(); 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第一次启动是从队列头部开始生产还是队列尾部开始生产<br> * 如果非第一次启动，那么依照上次生产的地位持续生产 */ 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<MessageExt> 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<Runnable>(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<String, Integer> localTrans = new ConcurrentHashMap<>(); @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { int value = 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应用倡议
- 请管制producer实例数，过多的连贯会造成server端的资源节约；
- 请给 producer 设置优雅敞开(producer.shutdown())，服务敞开能够立刻开释资源
- 接入多个rocketmq集群，为避免nameserver笼罩问题，须要设置InstanceName
consumer应用倡议
- 尽量应用push consumer
- 接入多个rocketmq集群，为避免nameserver笼罩问题，须要设置InstanceName
- 当应用TAG的时候，防止同一个consumerGroup 的Consumer 应用的TAG 的类型不同。
参考
RocketMQ 官网文档