《RocketMQ技术内幕》学习笔记_rocketmq中的ssb

一、RocketMQ源码目录

RocketMQ核心目录说明：

• broker：broker模块
• client：消息客户端：包括消息生产者、消息消费者等
• common：公共包
• dev：开发者信息（非源代码）
• distribution：部署实例文件夹（非源代码）
• example：RocketMQ示例代码
• filter：消息过滤相关基础类
• filtersrv：消息过滤服务器相关实现类
• logappender：日志实现相关类
• namesrv：NameServer实现相关类
• openmessaging：消息开放标准，正在定制定
• remoting：远程通信模块，基于Netty
• srvutil：服务器工具类
• store：消息存储相关实现类
• style：checkstyle相关实现
• test：测试相关类
• tools：工具类，监控命令相关实现类。

二、RocketMQ设计

2.1、设计理念

RocketMQ的设计基于topic的sub/pub模式，其核心功能包括消息发送、消息存储、消息消费。

整体设计追求简单与性能第一，主要体现在如下三个方面：
1、NameServer：摒弃了zookeeper，自研了NameServer来做元数据的管理（topic路由信息等）。由于topic路由信息能容忍分钟级别的不一致，所以不必强一致，做到最终一致性即可。所以RocketMQ的NameServer集群之间互不通信，极大地降低了NameServer实现的复杂度，但性能比zk强许多。-- 这个是对kafka zk的优化
2、IO存储机制：RocketMQ的消息存储文件设计成文件组，组内单个文件的大小固定，引入了内存映射机制（啥是内存映射？答：基于Netty NIO），所有topic消息的存储都是顺序写磁盘，极大提升了写性能。为了兼顾消息消息与消息查找，引入了消息消费队列文件与索引文件。-- 这些基本都是学的kafka啊。
3、容忍设计缺陷，将某些MQ的工作留给MQ的使用方来处理。即消息如何做到不丢、不重。RocketMQ只保证消息不丢，不保证消息不重。消息不重需要使用在在消费后自己实现幂等。

2.2、设计目标

RocketMQ作为一款消息中间件，用来解决下面的问题。

1、架构模式

RocketMQ与大多数消息中间件一样，采用sub/pub模式，参与组件主要包括：
消息生产者、消息存储（broker）、消息消费者、路由发现。

2、顺序消息

所谓顺序消息，指消息消费者按照消息到达消息存储服务器的顺序消费。RocketMQ可以严格保证消息有序。

3、消息过滤：这点kafka没有

消息过滤指，消息消费者可以按照指定规则，只消费topic下自己感兴趣的消息。RocketMQ消息过滤支持在服务端过滤和消费端过滤：
1）消息在broker过滤：broker只把consumer感兴趣的消息发给consumer
2）消息在consumer过滤：由consumer自己实现过滤，但缺点是很多无用的消息会从broker传输到consumer。

4、消息存储

消息存储会考虑：消息积压能力 和 消息存储写性能。
1）消息存储写性能：RocketMQ引入内存映射机制，所有topic的消息，顺序存储到了同一个文件中。 – kafka是写入多个文件中（多个partition），所以kafka无序，RocketMQ有序。
2）消息积压能力：为了避免消息无限制在服务器中堆积，引入了消息文件过期机制 与 文件存储空间报警 机制。-- 这与kafka类似

5、消息高可用性

可能影响消息可用性的场景：
1）Broker正常关机
2）Broker异常Crash
3）OS Crash
4）集群断电，但能立即恢复供电
5）集群无法开机（可能CPU、主板、内存等关键设备损坏）
6）磁盘设备损坏

对于上述情况，1~4 RocketMQ在同步刷盘时可保证消息不丢；在异步刷盘的模式下会丢少量消息。
5~6属于单点故障，一旦发生，该节点上的消息全部丢失；如果开启了异步复制机制，RocketMQ只丢少量消息。（怎样异步复制？）

6、消息消费低延迟

RocketMQ在消息不发送积压时，以长轮询模式实现准实时的消息推送。-- 是拉不是推

7、确保消息必须被消费一次：消息不丢

RocketMQ通过ack来确保消息至少被消费一次，但有重复消费的可能。

8、回溯消息

回溯消息指消息已经被consumer成功消费，但由于业务需求，还要重新消费。RocketMQ支持回溯消息，时间维度可精确到毫秒，还可以向前会向后回溯。-- 这点kafka只支持用offset回溯？

9、消息堆积

RocketMQ消息存储使用磁盘文件（内存映射机制），在物理布局上为多个大小相等的文件组成逻辑文件组，可以无限循环使用。RocketMQ的消息存储文件默认保留3天。-- 跟kafka类似

10、定时消息

定时消息指消息发送到broker后，不会马上被consumer消费，要等到特定时间点才被消费。RocketMQ不支持任意精度的定时消息，只支持特定延迟级别。
说明：如果要支持按任意精度的定时消息消费，需要再服务端对消息排序，势必极大损耗性能。所以RocketMQ不支持。

11、消息重试机制

消息重试指消息在被消费时，如果发送异常，RocketMQ支持消息重新发送。

三、路由中心NameServer

RocketMQ由NameServer来负责路由管理、服务注册、服务发现。
路由管理指让服务调用方找到”远方“的服务提供者。

3.1、NameServer架构设计

RocketMQ部署图：

Broker在启动时，会向所有的NameServer注册，producer在发消息前，先从NameServer获取Broker服务器地址列表，然后根据负载均衡算法，从列表中选择一台broker进行消息发送。

NameServer与每台Broker保持长连接，并间隔30s检测Broker是否存活，如果发现Broker宕机，则从路由注册表中将其移除。

这样路由变化了，Producer并不会马上知道，为什么要这样设计呢？这是为了降低NameServer实现的复杂性，在Producer提供容错机制来保证消息发送的高可用性。

NameServer本身的高可用是通过部署多台NameServer服务器来实现的，但不同NameServer之间彼此不相互通信，也就是说NameServer服务器之间在某一时刻的数据并不完全相同，但这对消息发送不会造成任何影响，这也是RocketMQ设计的一个亮点。RocketMQ的设计追求简单高效。

3.2、NameServer启动流程

NameServer启动类：org.apache.rocketmq.namesrv.NamesrvStartup.java
Step1、先解析配置文件，需要填充NameServerConfig、NettyServerConfig的属性值

NameServerConfig属性：

    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
    private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
    private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
    private String productEnvName = "center";
    private boolean clusterTest = false;
    private boolean orderMessageEnable = false;
1
2
3
4
5
6
7

NettyServerConfig属性：

private int listenPort = 8888;
private int serverWorkerThreads = 8;
private int serverCallbackExecutorThreads = 0;
private int serverSelectorThreads = 3;
private int serverOnewaySemaphoreValue = 256;
private int serverAsyncSemaphoreValue = 64;
private int serverChannelMaxIdleTimeSeconds = 120;

private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
private boolean serverPooledByteBufAllocatorEnable = true;
1
2
3
4
5
6
7
8
9
10
11

相关参数含义：

• rocketmqhome：rocketmq主目录，可通过 -Drocketmq.home.dir=path 或通过配置环境变量 ROCKETMQ_HOME来设置RocketMQ的主目录
• kvConfigPath：NameServer存储KV配置属性的持久化路径。
• configStortPath：nameServer默认配置文件路径，不省心。nameServer启动时如果要通过配置文件配置NameServer启动属性的话，请使用-c选项。
• orderMessageEnable：是否支持顺序消息，默认为不支持。
• listenPort：NameServer监听端口，默认为9876
• serverWorkerThreads：Netty业务线程池线程个数
• serverCallbackExecutorThreads：Netty public任务线程池线程个数， Netty 网络设计， 根据业务类型会创建不同的线程池，比如处理消息发送、消息消费、心跳检测等 。 如果该业务类型(RequestCode)未注册线程池， 则由 public线程池执行。
• serverSelectorThreads：IO线程池线程个数，主要是NameServer、Broker端解析请求、返回相应的线程个数，这类线程主要是处理网络请求的，解析请求包，然后转发到各个业务线完成具体的业务操作，然后将结果再返回调用方。
• serverOnewaySemaphoreValue：send oneway 消息请求井发度( Broker 端参数) 。
• serverAsyncSemaphoreValue：异步消息发送最大并发度（Broker端参数）
• serverChannelMaxldleTimeSeconds：网络连接最大空闲时间，默认120s。如果连接空闲时间超过该参数设置的值，连接将被关闭。
• serverSocketSndBufSize：网络socket发送缓存区大小，默认64k
• serverSocketRcvBufSize：网络socket接收缓存区大小，默认64k
• serverPooledByteBufAllocatorEnable：ByteBuffer是否开启缓存，建议开启
• useEpoolNativeSelector：是否启用Epoll IO模型，Linux环境建议开启

Step2、根据启动属性，创建NamesrvController实例， 并初始化该实例，NameServerController实例为NameServer的核心控制器。

public boolean initialize() {
        this.kvConfigManager.load();
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
        this.remotingExecutor =
            Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
        this.registerProcessor();
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);
        ...
        return true;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

加载KV配置，创建NettyServer网络处理对象，然后开启两个定时任务，在RocketMQ中此类定时任务统称为心跳检测。

• 定时任务1：NameServer每隔10s扫描一次Broker，移除处于不激活状态的Broker
• 定时任务2：nameServer每隔10分钟打印一次KV配置。

Step3、注册JVM钩子函数，并启动服务器，以便监听Broker、Producer的请求。

        Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                controller.shutdown();
                return null;
            }
        }));
1
2
3
4
5
6
7

这里使用钩子的目的是在JVM进程关闭前，先关闭线程池，及时释放资源。

3.2、NameServer路由注册、故障剔除

3.2.1、路由元信息

NameServer路由实现类：org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager。

RouteInfoManager路由元数据：

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
1
2
3
4
5

看各元数据的含义：

• topicQueueTable：Topic消息队列路由信息，消息发送时根据路由表进行负载均衡。
• brokerAddrTable：Broker基础信息，包括brokerName, 所属集群名称，主备Broker地址。
• clusterAddrTable：Broker集群信息，存储集群中所有Broker名称
• brokerLiveTable：Broker状态信息。NameServer每次收到心跳包时会替换该信息。
• filterServerTable：Broker上的FilterServer列表，用于类模式消息过滤。

RocketMQ基于sub/pub机制，一个topic有多个消息队列，一个broker为每一topic默认创建4个读队列和4个写队列。多个broker组成一个集群，BrokerName由相同的多台Broker组成Master-Slave架构，brokerId为0代表Master，大于0表示Slave。BrokerLiveInfo中的lastUpdateTimestamp存储上次收到Broker心跳包的时间。

3.3.2、路由注册

RocketMQ是通过Broker与NameServer之间的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳语句，每隔30s向集群中所有的NameServer发送心跳包，NameServer收到心跳包时会更新brokerLiveTable，如果连续120s没有收到心跳包，NameServer将移除该Broker的路由信息，同时关闭Socket连接。

1、Broker发送心跳包：BrokerController.java:

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
                } catch (Throwable e) {
                    log.error("registerBrokerAll Exception", e);
                }
            }
        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
1
2
3
4
5
6
7
8
9
10
11

BrokerOuterAPI: registerBrokerAll

        final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
        List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
        if (nameServerAddressList != null && nameServerAddressList.size() > 0) {

            final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
            requestHeader.setBrokerAddr(brokerAddr);
            requestHeader.setBrokerId(brokerId);
            requestHeader.setBrokerName(brokerName);
            requestHeader.setClusterName(clusterName);
            requestHeader.setHaServerAddr(haServerAddr);
            requestHeader.setCompressed(compressed);

            RegisterBrokerBody requestBody = new RegisterBrokerBody();
            requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
            requestBody.setFilterServerList(filterServerList);
            final byte[] body = requestBody.encode(compressed);
            final int bodyCrc32 = UtilAll.crc32(body);
            requestHeader.setBodyCrc32(bodyCrc32);
            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
            for (final String namesrvAddr : nameServerAddressList) { // 遍历所有NamServer
                brokerOuterExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);//分别向NameServer注册
                            if (result != null) {
                                registerBrokerResultList.add(result);
                            }

                            log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                        } catch (Exception e) {
                            log.warn("registerBroker Exception, {}", namesrvAddr, e);
                        } finally {
                            countDownLatch.countDown();
                        }
                    }
                });
            }

            try {
                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
            }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

2、NameServer处理心跳包

org.apache.rocketmq.namesrv,processor.DefaultRequestProcessor 解析请求类型，如果请求类型为 RequestCode.REGISTER_BROKER，则请求转发到RouteInfoManager#registerBroker：

            this.lock.writeLock().lockInterruptibly();
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            if (null == brokerNames) {
                brokerNames = new HashSet<String>();
                this.clusterAddrTable.put(clusterName, brokerNames);
            }
            brokerNames.add(brokerName);
1
2
3
4
5
6
7

Step1、路由注册需要加锁，防止并发修改RoutInfoManager中的路由表。首先判断Broker所属集群是否存在，不存在则创建，然后将Broker名加入到集群Broker集合中。

Step2、维护BrokerData信息：首先从brokerAddrTable根据BrokerName尝试获取Broker信息，如果不存在，则新建BrokerData放入brokerAddrTable，registerFirst设置为true；如果存在，直接替换原先的，registerFirst设置为false，表示非第一次注册。

if (null != topicConfigWrapper
    && MixAll.MASTER_ID == brokerId) {
    if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
        || registerFirst) {
        ConcurrentMap<String, TopicConfig> tcTable =
            topicConfigWrapper.getTopicConfigTable();
        if (tcTable != null) {
            for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                this.createAndUpdateQueueData(brokerName, entry.getValue());
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

Step3、如果Broker为Master，并且Broker Topic配置信息发送变化，或者初次注册，则需要创建或更新Topic路由元数据，填充topicQueueTable，其实就是为默认主题自动注册路由信息，其中包含MixAll.DEFAULT_TOPIC的路由信息。当producer发送topic时，如果该topic为创建，并且BrokerConfig的autoCreateTopicEnable为true时，将返回MixAll.DEFAULT_TOPIC的路由信息。

private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
        QueueData queueData = new QueueData();
        queueData.setBrokerName(brokerName);
        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
        queueData.setPerm(topicConfig.getPerm());
        queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());

        List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
        if (null == queueDataList) {
            queueDataList = new LinkedList<QueueData>();
            queueDataList.add(queueData);
            this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
            log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
        } else {
            boolean addNewOne = true;

            Iterator<QueueData> it = queueDataList.iterator();
            while (it.hasNext()) {
                QueueData qd = it.next();
                if (qd.getBrokerName().equals(brokerName)) {
                    if (qd.equals(queueData)) {
                        addNewOne = false;
                    } else {
                        log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
                            queueData);
                        it.remove();
                    }
                }
            }

            if (addNewOne) {
                queueDataList.add(queueData);
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

Step4、更新BrokerLiveInfo，存活Broker信息表，BrokerLiveInfo是执行路由删除的重要依据。
Step5、注册Broker的过滤器Server地址列表，一个Broker上会关联对哦个FilterServer消息过滤服务器。

设计亮点：NameServer与Broker保持长链接，Broker状态存储在brokerLiveTable中，NameServer每收到一个心跳包，将更新brokerLiveTable中关于broker的状态信息及路由表（topicQueueTable, brokerAddrtable，brokerLiveTable, filterServerTable）。更新上述路由表（HashTable）使用了锁力度较少的读写锁，允许多个producer并发读。但同一时刻NameServer只处理一个Broker心跳包，多个心跳包请求串行执行。

3.3.3、路由删除

Broker每隔30s向NameServer发送一个心跳包，心跳包中包含BrokerId, Broker地址、Broker所属集群名称、Broker关联的FilterServer列表。

NameServer会每隔10s扫描brokerListTable状态表，如果BrokerLive的lastUpdateTimestamp的时间距离当前时间超过120s，则认为broker失效，移除该broker，关闭与broker连接，并同时更新topicQueueTable, brokerAddrTable, brokerLiveTable, filterServerTable、

3.3.4、路由发现

当Topic路由出现变化后，NameServer不主动推送给客户端，而是由客户端定时拉取Topic最新的路由。根据Topic名拉取路由信息的命令编码为：GET_ROUTENINTO_BY_TOPIC。

3.4、总结

四、RocketMQ消息发送

4.1、RocketMQ消息发送概述

RocketMQ支持3种消息发送方式：

• 同步（sync）：producer向MQ发送消息时，同步等待，直到消息服务器返回发送结果。
• 异步（async）：producer向MQ发送消息时，指定消息发送成功后的回调函数，然后调用消息发送API后，立即返回，producer线程不阻塞，直到运行结束，producer发送成功或失败的回调任务在一个新的线程中执行。
• 单向（oneway）：producer想MQ发送消息时，直接返回，不等待MQ的结果，也不注册回调函数。也就是只管发，不管发送是否成功。

RocketMQ消息发送需要考虑以下几个问题：

• 消息队列如何进行负载均衡
• 消息发送如何实现高可用
• 批量消息发送如何实现一致性

4.2、RocketMQ消息构成

RocketMQ的消息封装类是org.apache.rocketmq.common.message.Message

Message的属性主要包括topic、flag（RocketMQ不做处理）、扩展属性、消息体。

RocketMQ定义的MessageFlag：

Message的扩展属性属于包括：

• tag：消息tag，用于做消息过滤
• keys：Message索引键，多个用空格隔开，RocketMQ可以根据这些key快速检索到消息。
• waitStoreMsgOK：消息发送时是否等待消息存储完成后再返回
• delayTimeLevel：消息延迟级别，用于定时消息或消息重试
  这些扩展属性存储在了Message的properties中。

4.3、Producer启动流程

Producer代码都在client模块中，相对于RocketMQ来说，他就是客户端。

4.3.1、DefaultMQProducer消息发送者

DefaultMQProducer是默认的producer实现类，它实现了MQAdmin接口。


下面看DefaultMQProducer的主要方法：
1）void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) ：
创建Topic：
key：目前未实际作用，可与newTopic相同
newTopic：主题名称
queueNum：队列数量
topicSysFlag：主题系统标签，默认为0

2）long searchOffset(final M巳ssageQueue mq, final long timestamp)
根据时间戳从队列中查找其偏移量

3）long maxOffset(final MessageQueue mq)
查找该消息队列中最大的物理偏移量

4）long minOffset(final MessageQueue mq)
查找该消息队列中最小的物理偏移量

5）MessageExt viewMessage(final String offsetMsgId）
根据消息偏移量查找消息

6）QueryResult queryMessage(final String topic, final String key, final int maxNum, final
long begin, final long end)
感觉条件查询消息
topic：消息主题
key：消息索引字段
maxNum：本次最多取出消息条数
begin：开始时间
end：结束时间

7）MessageExt viewMessage(String topic,String msgld)
根据topic与消息ID查找消息

8）List fetchPublishMessageQueues(final String topic)
查找topic下所有的消息队列

9）SendResult send(final Message msg)
同步发送消息，具体发送到topic中的哪个消息队列由负载均衡算法决定

10）SendResult send(final Message msg, final long timeout)
同步发送消息，如果发送超过timeout则抛出超时异常

11）void send(final Message msg, final SendCallback sendCallback)
异步发送消息，sendCallback参数是消息发送成功后的回调方法

12）void send(final Message msg, final SendCallback sendCallback, final long timeout)
异步发送消息，超时则抛出异常

13）void sendOneway(final Message msg)
单向发消息，不等待返回

14）SendResult send(final Message msg, final MessageQueue mq)
同步发消息，发送到指定队列

15）void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback)
异步发送消息，发送到指定队列

16）void sendOneway(final Message msg, final MessageQueue mq)
单向发送消息，发到指定队列

17）SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg)
消息发送，指定消息负载均衡算法，覆盖消息默认的消息队列负载均衡算法

18）SendResult send(final Collection msgs, final MessageQueue mq, final long timeout)
同步批量发送消息

DefaultMQProducer核心属性：

protected final transient DefaultMQProducerImpl defaultMQProducerImpl;
private String producerGroup;
private String createTopicKey = TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC;
private volatile int defaultTopicQueueNums = 4;
private int sendMsgTimeout = 3000;
private int compressMsgBodyOverHowmuch = 1024 * 4;
private int retryTimesWhenSendFailed = 2;
private int retryTimesWhenSendAsyncFailed = 2;
private boolean retryAnotherBrokerWhenNotStoreOK = false;
private int maxMessageSize = 1024 * 1024 * 4; // 4M
private TraceDispatcher traceDispatcher = null;
1
2
3
4
5
6
7
8
9
10
11

producerGroup: 生产者所属组，消息服务器在回查事务状态时会随机选择该组中任何一 个生产者发起事务回查请求 。
createTopicKey：默认topicKey
defaultTopicQueueNums：默认topic在每个broker的队列数量
sendMsgTimeout：发消息默认超时时间，默认3s
compressMsgBodyOverHowmuch：消息体超过该阈值则启用压缩，默认4K
retryTimesWhenSendFailed：同步发送消息重试次数，默认为2，总共执行3次。
retryTimesWhenSendAsyncFailed：异步发送消息重试次数，默认为2
retryAnotherBrokerWhenNotStoreOK：消息重试时选择另外一个Broker时，是否不等待存储就返回，默认为false
maxMessageSize：允许发送的最大消息长度，默认为4M，该值最大值为2^32-1

4.3.2、消息生产者启动流程

Step1、检查productGroup是否符合要求，并改变生产者的instanceName为进程ID。
Step2、创建MQClientInstance实例。‘
Step3、向MQClientInstance注册，将当前producer加入到MQClientInstance管理中，方便后续调用网络请求、进行心跳检测等。
Step4、启动MQClientInstance，如果MQClientInstance已经启动，则本次启动不会正在执行。

4.4、消息发送基本流程

消息发送流程的主要步骤：验证消息、查找路由、消息发送。

默认消息发送为同步发送，默认超时时间为3s。

4.4.1、消息长度验证

消息发送前，先验证producer是否处于运行状态，然后验证消息是否符合规范：

• topic名称、消息体不能为空
• 消息长度不能为0且默认不允许超过最大长度4M

4.4.2、查找主题路由信息

消息发送前，先获取topic的路由信息，这样才知道要发送给哪个broker

private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
    TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    if (null == topicPublishInfo || !topicPublishInfo.ok()) {
        this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
    }

    if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
        return topicPublishInfo;
    } else {
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
        return topicPublishInfo;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

tryToFindTopicPublishInfo是查找topic路由信息的方法。如果Producer中缓存了topic的路由信息，如果该路由信息中包含了消息队列，则直接返回路由信息，如果没有缓存或没有包含消息队列，则想NameServer查询该topic的路由信息。如果最终未找到路由信息，则抛出异常。

下面看下TopicPublishInfo的属性：

• orderTopic：是否是顺序消息
• List messageQueueList：该主题的消息队列
• sendWhichQueue：每选择一次消息队列，该值会自增1，如果Integer.MAX_VALUE，则重置为0，用于选择消息队列。
• List brokerDatas：topic分布的broker元数据
• HashMap<String /* brokerAddress*/, List<String /filterServer/>: broker上过滤服务器地址列表

4.4.3、选择消息队列

去NameServer获取消息队列信息，如topicA在broker-a, broker-b上分别创建4个队列，那么返回信息为：
[{ “brokerName”:“broker-a”,“queueId”:0},
{“brokerName”:“broker-a”:“queueId”:1},
{“brokerName”:“broker-a”:“queueId”:2},
{“brokerName”:“broker-a”:“queueId”:3},
{“brokerName”:“broker-b”:“queueId”:0},
{“brokerName”:“broker-b”:“queueId”:1},
{“brokerName”:“broker-b”:“queueId”:2},
{“brokerName”:“broker-b”:“queueId”:3}]
, 那么RocketMQ会如何选择消息队列呢？

消息发送端采用重试机制，由retryTimesWhenSendFailed指定同步方式重试次数，异步重试机制是在收到消息发送结果后执行回调之前进行重试，有retryTimesWhendSendAsyncFailed指定。

然后循环执行，选择消息队列、发送消息、发送成功则返回，收到异常则重试。

选择消息队列有两种方式：
1）sendLatencyFaultEnable=false，默认不启用Broker故障延迟机制：为默认机制。
在一次消息发送过程中，如果Broker宕机，随后会选择宕机Broker的第二个队列，这事消息发送很可能会失败，再次引发重试。

2）sendLatencyFaultEnable=true, 启用Broker故障延迟机制
当Broker宕机时，一次发送消息失败，下次会把宕机的Broker排除在消息队列的选择范围之外。

4.4.4、消息发送

消息发送API接口：

private SendResult sendKernelImpl(final Message msg,
        final MessageQueue mq,
        final CommunicationMode communicationMode,
        final SendCallback sendCallback,
        final TopicPublishInfo topicPublishInfo,
        final long timeout)
1
2
3
4
5
6

参数详解：
1）Message msg：待发送消息
2）MessageQueue mq：消息将发送到该消息队列上
3）CommunicationMode communicationMode：消息发送模式：sync, async, oneway
4）SendCallback sendCallback：异步消息回调函数
5）TopicPublishInfo topicPublishInfo：topic路由消息
6）long timeout：消息发送超时时间

4.4.5、批量消息发送

批量消息发送至将同一topic的多条消息一起打包发送，减少网络调用次数。单批发送消息的总长度不能超过DefaultMQProducer#maxMessageSize。

4.5、总结

消息发送的整个流程，充电：
1）消息producer启动流程
重点理解MQClientInstance、消息生产者之间的关系

2）消息队列负载均衡机制
消息producer在发送消息时，如果本地路由表中未缓存topic路由信息，向NameServer发送获取路由信息，更新本地路由信息表，并且消息Producer每隔30s从NameServer更新路由表。

3）消息发送异常机制
消息发送高可用主要通过两个手段：重试与Broker规避。
broker规避就是在一次消息发送过程中发现错误，在某一时间段内，Producer不会选择该Broker上的消息队列，提高消息的成功率。

4）批量消息发送

五、RocketMQ消息存储

5.1、存储设计概要

RocketMQ的存储主要包括：Commitlog文件、ConsumerQueue文件、IndexFile文件。

RocketMQ将所有Topic的消息都存储在同一个文件中，确保消息发送时顺序写文件，这样可以最大限度保证消息发送的高性能与高吞吐。
但由于消息中间件一般是基于topic订阅机制，这样给按消息Topic查找消息带来了极大的不便。

为了提高消息消费的效率，RocketMQ引入了ConsumerQueue的消息队列文件，每个消息tpic包含多个消息消费队列，每一个消息队列有一个消息文件。

IndexFile索引文件，主要设计理念是为了加速消息的检索新能，根据消息的属性，快速从Commitlog文件中检索消息。

RocketMQ是一款高性能的消息中间件，存储部分是设计核心，存储的核心是IO访问性能。

1）Commitlog：消息存储文件，所有消息topic的消息都存储在Commitlog文件中
2）ConsumeQueue：消息消费队列，消息到达Commitlog文件后，将一部转发到消息消费队列，供消息消费者消费。
3）IndexFile：消息索引文件，主要存储消息Key与Offset的对应关系
4）事务状态服务：存储每条消息的事务状态
5）定时消息服务：每一个延迟级别对应一个消息消费队列，存储延迟队列的消息拉取进度。

5.2、消息存储结构

消息存储实现类：org.apache.rocketmq.store.DefaultMessageStore：

5.3、消息发送存储流程

Step1、如果当前Broker停止工作，或Broker为slave角色或当前Rocket不支持写入，则拒绝消息写入；如果topic超过256个字符，消息属性长度超过65536个字符，将拒绝该消息的写入。

Step2、如果消息延迟级别大于0，将消息的原topic名称与原消息队列ID存入消息属性中，用延迟消息topic SCHEDULE_TOPIC、消息度列ID更新原先消息的主题与队列，这是并发消息重试的关键一步。

Step3、获取当前可以写入的Commitlog文件。

Step4、在写入Commitlog之前，先申请putMessageLock，也就是将消息存储到Commitlog文件中是串行的

Step5、设置消息的存储时间

Step6、将消息追加到MappedFile中。

Step7、创建全局唯一的消息ID，消息ID有16个字节，消息ID的组成如下图：

Step8、获取该消息在消息队列的偏移量。Commitlog中保存了当前所有消息队列的当前待写入偏移量。

Step9、根据消息体程度、topic长度、属性的长度，结合消息存储格式，计算消息的总长度

Step10、如果消息长度+END_FILE_MIN_BLANK_LANGTH大于Commitlog文件的空闲空间，则返回AppendMessageStatus.END_OF_FILE, Broker会重新创建一个新的Commitlog文件来存储该消息。

Step11、将消息内容存储到ByteBuffer中，然后创建AppendMessageResult。这里只讲消息存储在MappedFile对应的内存映射Buffer中，并没有刷鞋到磁盘。

Step12、更新消息队列逻辑偏移量

Step13、处理完消息追加逻辑后，将释放putMesssageLock锁。

Step14、DefaultAppendMessageCallback#doAppend只是将消息追加到内存中，需要根据同步刷还是异步刷盘方式，将内存中的数据持久化到磁盘。

5.4、存储文件组织与内存映射

RocketMQ通过使用内存映射稳居来提供IO访问性能，不论是Commitlog，ConsumerQueue，还是IndexFile，单个文件都被设计为固定长度，如果一个文件写满以后再创建一个新文件，文件名就为该文件第一条消息对应的全局offset。例如Commitlog的文件组织方式如下图：

RocketMQ使用MappedFile, MappedFileQueue来封装存储文件，其关系如下图：

5.4.1、MappedFileQueue映射文件队列

MappedFileQueue是MappedFile的管理容器，MappedFileQueue是对存储目录的封装，例如Commitlog文件存储路径 ${ROCKET_HOME/store/commitlog}，该目录下会存在多个内存映射文件（MappedFile）。MappedFileQueue类图：

说明：这个设计抄的Kafka吧：Kafka的partition就是文件夹，然后segment文件放到文件夹里。

5.4.2、TransientStorePool 存储池

TransientStorePool：RocketMQ单独创建一个MappedByteBuffer内存缓存池，用来临时存储数据，数据先写入该内存映射中，然后有commit线程定时将数据从该内存复制到目的物理文件对应的内存映射中。RocketMQ引入该机制的主要原因是提供一种内存锁，将当前堆外内存一直锁定在内存中，避免被进程将内存交换到磁盘。

看TransientStorePool#init代码：

public void init() {
    for (int i = 0; i < poolSize; i++) {
        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);

        final long address = ((DirectBuffer) byteBuffer).address();
        Pointer pointer = new Pointer(address);
        LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));

        availableBuffers.offer(byteBuffer);
    }
}
1
2
3
4
5
6
7
8
9
10
11

创建poolSize个堆外内存，并利用som.sun.jna.Library类库将该批内存锁定，避免被置换到交换区，提高存储性能。

5.5、RocketMQ存储文件

RocketMQ存储路径为 ${ROMCT_HOME}/store，主要存储文件如下图：

1）commitlog：消息存储目录
2）config：运行期间一些配置信息，主要包括：
consumerFilter.json：主题消息过滤信息
consumerOffset.json：集群消费模式消息消费进度
delayOffset.json：延时消息队列拉取进度
subscriptionGroup.json：消息消费组配置信息
topics.json：topic配置属性
3）consumerqueue：消息消费队列存储目录
4）index：消息索引文件存储目录
5）abort：如果存在abort文件说明Broker非正常关闭，该文件默认启动时创建，正常退出之前删除。
6）checkpoint：文件检测点，存储commitlog文件最后一次刷盘时间戳，comsumequeue最后一次刷盘时间，Index索引文件最后一次刷盘时间戳。

5.5.1、Commitlog文件

Commitlog文件存储逻辑如下图，每条消息的签名4个字节存储该条消息的总长度：

Commitlog文件存储目录默认为 ${ROCKET_HOME}/store/commitlog，可以通过再broker配置文件中设置storePathRootDir来改变。commitlog文件默认大小为1G，可在broker配置文件中设置mappedFileSizeCommitlog属性来修改。

5.5.2、ConsumeQueue文件

ConsumeQueue中并不需要存储消息的内容，而存储的是消息在CommitLog中的offset。也就是说，ConsumeQueue其实是CommitLog的一个索引文件。

RocketMQ基于topic sub/pub模式实现消息消费，消费者关心的是一个topic下的所有消息，但由于同一topic下的消息不连续地存储在commitlog文件中，如果从commitlog中直接消费，那么性能极差。RocketMQ设计了消息消费队列文件ConsumeQueue，该文件可以视为Commitlog关于消息消费的”索引“文件，comsumequeue的第一集目录为消息topic, 第二季目录为topic的消息队列：

为了加速Consumequeue的消息检索速度，并节省磁盘空间，每一个Consumequeue不会存储消息的全量信息，其存储格式如下图：

单个Consumequeue文件中默认包含了30万个条目，单个文件的长度为30w * 20字节，单个Consumequeue文件可以看做是一个ConsumeQueue条目的数组，其下标为ConsumeQueue的逻辑偏移量，消息消费进度存储的偏移量即逻辑偏移量。ComsumeQueue即为Commitlog文件的索引文件，其构建机制是当消息到达Commitlog文件后，由专门的线程产生消息转发任务，从而构建消息消费队列文件与索引文件。

5.5.3、Index索引文件

如果我们需要根据消息ID，来查找消息，consumequeue 中没有存储消息ID, 如果不采取其他措施，又得遍历 commitlog文件了，indexFile就是为了解决这个问题的文件

IndexFile的存储内容是什么？
hash槽存储的是节点的索引位置。index条目存储的是key的hash值，物理的offset，与beginTimeStamp的差值、上一个hash冲突的索引位置。

怎么把一条消息放入到IndexFile?
确定hash槽
先是根据key计算hashcode，对500w取模，就可以知道位于哪个hash槽。indexHead占了文件的前面的40字节。然后每个hash槽占4个字节。具体在文件的位置是由公式40 + keyIndex*4计算得到的。

计算index条目位置
一条消息hash槽的位置是根据key决定的，index条目的位置是放入的顺序决定的，这叫顺序写。
index条目首先要跨过indexHead和500w个hash槽的大小。然后根据当前是第几条index条目，就放入到第几个位置去。
计算公式是:40个字节的indexHead+500w个 * 4字节的hash槽大小 + 当前index索引的值 * 20字节

怎么查询indexFile？
先是根据key计算hashcode，对500w取模，就可以知道位于哪个hash槽。根据槽值的内容，再通过计算index条目位置，获取到index条目，再依次获取上一个hash冲突节点的index条目。

为什么需要indexFile？
在客户端（生产者和消费者）和admin接口提供了根据key查询消息的实现。为了方便用户查询具体某条消息。但是关键得找到消息的key。

为什么这么设计IndexFile？
使用文件实现，应该是为了轻量级，不依赖其他组件
indexHead可以根据时间快速定位要查找的key在哪个indexFile。
使用了hash槽的设计，提供O(1)的时间复杂度即可定位到index条目。
使用hash槽存index条目位置，index条目顺序写入，提供了写的性能

5.5.4、checkpoint文件

checkpoint文件的作用是记录Commitlog, ConsumeQueue, Index文件的刷盘时间点，文件固定长度为4k, 其中只用该文件的前面24个字节：

5.6、实时更新消息消费队列与索引文件

ConsumeQueue文件、Index文件都是基于CommitLog文件构建的，当消息Producer提交的消息存储在CommitLog文件组宏，ConsumeQueue、IndexFile需要及时更新，否则消息无法及时被消费，根据消息属性查找消息也会出现较大延迟。

RocketMQ通过开启一个县城ReputMessageService来准实时转发CommitLog文件更新时间，响应的任务处理器根据转发的消息及时更新ConsumeQueue, IndexFile 文件。

5.7、消息队列与索引文件恢复

由于RocketMQ存储首先将消息全量存储在CommitLog文件中，然后异步生成转发任务，更新ConsumeQueue, Index文件。如果消息成功存储到CommitLog文件中，而转发文物为成功执行，此时消息服务器Broker由于某个原因宕机，导致Commitlog, Consumequeue, IndexFile文件数据不一致。如果不加以人工修复的话，会有一部分消息即便在Commitlog文件中存在，但由于没有转发到ComsumeQueue，这部分消息永远不会被消费者消费。那RocketMQ是如何使Commitlog, ConsumeQueue达到最终一致性的呢？

可从RocketMQ对存储文件的加载流程来看：

Step1、判断上一次退出是否正常：其实现机制为Broker在启动时创建 ${ROCKET_HOME}/store/abort文件，在退出时通过注册JVM狗子函数删除abort文件。如果下一次启动时abort文件存在，说明broker是异常退出的，Commitlog与ConsumeQueue数据有可能不一致，需要进行修复。

Step2、加载延迟队列，RocketMQ定时消息相关

Step3、加载Commitlog文件，加载 ${ROCKET_HOME}/store/commitlog 目录下所有文件，并按文件名排序。如果文件大小与配置文件的单个文件大小不一致，将忽略该目录下所有文件，然后创建MappedFile对象。

Step4、加载消息消费队列，调用DefaultMessageStore#loadConsumeQueue，其思路与Commitlog基本一致，遍历消息消费队列根目录，获取该Broker存储的所有主题，然后遍历每个主题目录，获取该主题下的所有消息消费队列，然后分别加载每个消息消费队列下的文件，构建ConumeQueue对象, 主要初始化ConsumeQueue的topic, queueId, storePath, mappedFileSize属性。

Step5、加载checkpoint文件，主要记录commitlog文件，ConsumeQueue文件，Index索引文件的刷盘点

Step6、加载索引文件，如果上次异常退出，而且索引文件上次刷盘时间小于该索引文件最大的消息时间戳，该文件将立即销毁。

Step7、根据Broker是否是正常停止，执行不同的恢复策略。

Step8、恢复ConsumeQueue文件后，将在Commitlog实例中保存每个消息消费队列当前的存储逻辑偏移量，这也是消息中不近存储 主题、消息队列ID，还存储了消息度列偏移量的关键所在。

5.8、文件刷盘机制

RocketMQ的存储与读写是基于JDK NIO的内存映射机制（MappedByteBuffer），消息存储时，首先将消息追加到内存，在根据配置的刷盘策略，在不同时间进行刷鞋磁盘。

如果是同步刷盘，消息追加到内存后，将同步调用MapptedByteBuffer的 force() 方法；
如果是异步刷盘，在消息追加到内存后，立即返回给消息发送端。RocketMQ使用一个单独的线程，安装某一个设定的频率执行刷盘操作。通过再broker配置文件中配置flushDiskType来设置刷盘方式，可选值为 ASYNC_FLUSH、SYNC_FLUSH，默认为异步刷盘。
ROCKETMQ处理刷盘的实现方法为Commitlog#handleDiskFlush()方法，刷盘流程作为消息发送、消息存储的子流程。

值得注意的事，索引文件的刷盘并不是采取定时刷盘机制，而是每更新一次索引文件，就会将上一次的改动刷写到磁盘。

5.9、过期文件删除机制

由于RocketMQ操作Commitlog, ConsumeQueue文件时基于内存映射机制，并在启动的时候加载commitlog, ConsumeQueue目录下的所有文件，为了避免内存与磁盘的浪费，不可能将消息用具存储在消息服务器上，所以需要删除已过期文件。

RocketMQ顺序写Commitlog文件、ConsumeQueue文件，所有写操作全部落在最后一个Commitlog或ConsumeQueue文件上，之前的文件在下一个文件创建后，将不会再被更新。

RocketMQ清除过期文件的方法是：如果非当前写文件，在一定时间间隔内没有再次被更新，则认为是过期文件，可以被删除，RocketMQ不会关注这个文件上的消息是否全部被消费。每个文件的过期时间默认为72小时，通过再Broker配置文件中设置fileReservedTime来改变过期时间，单位为小时。
但磁盘空间不足时，也会触发清除文件操作；也可以手动触发清除文件操作。

5.10、总结

RocketMQ主要存储文件包括：消息文件（CommitLog）、消息消费队列文件（ConsumeQueue）、Hash索引文件（IndexFile）、检测点文件（checkpoint）、异常关闭文件（abort)。
单个消息存储文件、消息消费队列文件、Hash索引文件长度固定，以便使用内存映射机制进行文件的读写操作。

RocketMQ组织文件以文件的起始offset来命名文件，目的是根据Offset能快速定位到真实的物理文件。RocketMQ基于内存映射文件机制，提供了同步刷盘与异步刷盘2种机制，异步刷盘指在消息存储时先追加到内存映射文件，然后启动专门的刷盘线程，定时将内存中的数据写到磁盘。

Commitlog, 消息存储文件：RocketMQ为了保险消息发送的高吞吐量，采用单一文件存储所有topic的消息，保证消息存储是完全的顺序写，但这样给文件读取同样代理了不便，为此，RocketMQ构建了消息消费队列文件，基于topic与队列进行组织，同时RocketMQ为消息实现了Hash索引，可以为消息设置索引键，根据索引能快速从Commitlog文件中检索消息。

当消息达到Commitlog文件后，会通过ReputMessageService线程接近实时地将消息转发给消息消费队列文件与索引文件。为了安全起见，RocketMQ引入了abort文件，记录Broker的停机是正常关闭还是异常关闭，在重启Broker时为了保证Commitlog文件与消息消费队列文件与Hash索引文件的正确性，分别采取不同的策略来恢复文件。

RocketMQ不会永久存储消息文件，消息消费队列文件，而是启用文件过期机制并在磁盘空间不足或默认在凌晨4点删除过期文件，文件默认保存72小时，并且在删除文件时并不会判断该消息文件上的消息是否被消费。

六、RocketMQ消息消费

6.1、消息消费概述

与Kafka一样，消息消费以group为单位，一个group内可以有多个consumer，每个consumer可以订阅多个topic。

group之间有集群模式和广播模式两种消费模式：

• 集群模式：topic下的同一条消息只允许被其中一个consumer消费
• 广播模式：topic下的同一条消息将被集群内的所有consumer消费一次。

Broker和Consumer之间的消息传递也分为两种模式：

• 推模式：消息到达broker后，推送给consumer
• 拉模式：consumer主动发起拉消息请求
  RocketMQ的消息推模式的实现是基于拉模式：在拉模式上包装了一层，一个拉取任务完成后，开始下一个拉取任务。

在集群模式下，多个消费者如何对消息队列进行负载呢？消息队列负载遵循一个通用的思想：一个消息队列同一时间只允许被一个消费者消费，一个消费者可以消费多个消息队列。

RocketMQ支持保证同一个消息队列上的消息顺序消费，不支持消息全局顺序消费。如果要实现某一topic的全局消息有序消费，可以将该topic的队列数设置为1，牺牲高可用性。

RocketMQ支持两种消息过滤模式：表达式（TAG, SQL92）与类过滤模式。

消息拉模式，主要是由客户端手动调用消息拉取API，而消息推模式是消息服务器主动将消息推送到消息消费端。

6.2、消息消费简介

消息消费分为推和拉两种模式。

6.2.1、推模式

先介绍推模式：消费者MQPushConsume的主要API：

1 ) void sendMessageBack (MessageExt msg, int delayLevel, String brokerName ) 发送消息 ACK确认。
msg: 消息。
delayLevel：消息延迟级别。
brokerName：消息服务器名称。

2）Set fetchSubscribeMessageQueues (final String topic)
获取消息topic分配了哪些消息队列

3. void registerMessageListener ( final MessageListenerConcurrently messageListener )
   注册并发消息事件监听器

4. void registerMessageListener ( final MessageListenerOrderly messageListener )
   注册顺序消息事件监听器

5. void subscribe (final String topic, final String subExpression)
   基于主题订阅消息
   topic：消息主题
   subExpression：消息过滤表达式，TAG或SQL92表达式

6）void subscribe (final String topic, final String ful!ClassName,final String filterClassSource)
基于主题订阅消息，消息过滤方式使用类模式
topic：消息主题
fullClassName：过滤类全路径名
filterClassSource：过滤类代码

7）void unsubscribe ( final String topic )
取消消息订阅

DefaultMQPushConsumer（推模式消息消费者）主要属性图：

1）consumerGroup：消费者组
2）messageModel: 消息消费模式，分为集群模式、广播模式，默认为集群模式 。
3) ConsumeFromWhere consumeFromWhere，根据消息进度从消息服务器拉取不到消息时重新计算消费策略 。
CONSUME_FROM_LAST_OFFSET：从队列当前最大偏移量开始消费
CONSUME_FROM_FIRST_OFFSET：从队列当前最小偏移量开始消费
CONSUME_FROM_TIMESTAMP：从消费者启动时间戳开始消费
FROM TIMESTAMP :从消费者启动时间戳开始消费 。
注意：如果从消息进度服务OffsetStoreduqu dao MessageQueue中的Offset不小于0，则使用读取到的偏移量，只有在读到的偏移量小于0时，上述策略才会生效。
4）allocateMessageQueueStrategy：集群模式下消息队列负载策略
5）Map<String /topic/, String /*sub expression */> subscription：订阅消息
6）MessageListener messageListener：消息业务监听器
7）private OffsetStore offsetStore：消息消费进度存储器
8）int consumeThreadMin = 20, 消息最新线程数
9）int consumeThreadMax = 64, 消费者最大线程数，由于消费者线程池使用无界队列，所以消费者线程数最多只有consumeThreadMin个
10）int consumeConcurrentlyMaxSpan, 并发消息消费时处理队列对打跨度，默认为2000.
11）int pullThresholdForQueue：默认值1000，每1000次流控后打印流控日志
12）long pullInterval = 0, 推模式下拉取任务间隔时间，默认一次拉取任务完成绩效拉取。
13）int pullBatchSize：每次消息拉取所拉取的条数，默认32条
14）int consumeMessageBatchMaxSize ：消息并发消费时一次消费消息条数
15）postSubscriptionWhenPull：是否每次拉取消息都更新等于消息，默认为false
16）postSubscriptionTimes：最大消费重试次数。如果消息消费次数超过maxReconsumeTimes还未成功，则该消息转移到一个失败队列，等待被删除
17）suspendConcurrentQueueTimeMillis：延迟将该队列的消息提交到消费者线程的等待时间，默认延迟1s。
18）long consumeTimeout, 消息消费超时时间，默认为15，单位为分钟。

6.2.2、拉模式

RocketMQ使用一个单独的线程PullMessageService来负责消息的拉取。

消息拉取分3步：
1）消息拉取客户端对消息拉取请求封装
2）消息服务器查找并返回消息
3）消息拉取客户端处理返回的消息

6.2.3、消息拉取长轮询机制分析

RocketMQ并没有真正实现推模式，而是消费者主动向消息服务器拉取消息，RocketMQ推模式是循环向消息服务端发送消息拉取请求，如果消息消费者向RocketMQ发送消息拉取时，消息并未到达消费队列，如果不启用长轮询机制，则会在服务端等待shortPollingTimeMills时间后（挂起）再去判断消息是否已到达消息队列，如果消息未到达，则提示消息拉取客户端PULL_NOT_FOUND（消息不存在）；如果开启长轮询模式，RocketMQ一方面会每5s轮询检查一次消息是否可达，同时一有新消息到达后，立即通知挂起线程，再次验证新消息是否是自己感兴趣的消息，如果是，则从commitlog文件提取消息，返回给消息拉取客户端，否则直到挂起超时，超时时间由消息拉取方在消息拉取时封装在请求参数中，PUSH模式默认为15s，PULL模式通过DefaultMQPullConsumer#setBrokerSuspendMaxTimeMillis设置。RocketMQ通过再Broker端配置longPollingEnable=true来开启长轮询模式。

6.3、总结

消息消费重点包括：消息消费方式、消息队列负载、消息拉取、消息消费、消息消费进度存储、消息过滤、定时消息、顺序消息。

RocketMQ消息消费方式分别为集群模式和广播模式。

消息队列负载由RebalanceService线程默认每20s进行一次消息队列负载均衡，根据当前消费组内消费者个数与主题队列数量，按照某一种负载算法进行队列分配，分配原则为同一个消费者可以分配多个消息消费度列，同一个消息消费队列同一时间只会分配给一个消费者。

消息拉取由PullMessageService线程根据RebalanceService线程创建的拉取任务进行拉取，默认一批拉取32条消息，提交给消费者消费线程池后，继续下一次的消息拉取。如果消息消费过慢，产生消息堆积会触发消息消费拉取流控。

并发消息消费，指消费线程池中的线程可以并发地从一个消息消费队列的消息进行消费，消费成功后，取出消息处理队列中最小消息偏移量作为消息消费进度偏移量存储到消息消费进度存储文件中，集群模式消息进度存储在Broker，广播模式消息进度存储在消费者端。如果业务方返回RECONSUME_LATER，则RocketMQ启用消息消费重试机制，将原消息的主题与队列存储在消息属性中，将消息存储在主题名为SCHEDULE_TOPIC_XXX的消息消费队列中，等待指定时间后，RocketMQ将自动将该消息重新拉取，并在此将消息存储在commitlog进而转发到原主要的消息消费队列，并消费者消费，消息消费重试主题为%RETRY%消费者组名。

RocketMQ消息消费支持表达式与类过滤模式。其中表达式消息过滤，又分为基于TAG模式与SQL92表达式，TAG模式就是为消息设定一个TAG，然后消息消费者订阅TAG，如果消费者订阅的TAG列表包含消息的TAG则消费该消息。SQL92表达式基于消息属性实现SQL条件表达式的过滤模式。

顺序消息消费一般使用集群模式，是指消息消费者内的线程池中的线程，对消息消费队列只能串行消费。与并发消息消费最本质的区别是，消费消息时，必须成功锁定消息消费队列在Broker端存储消息消费队列的锁占用情况。

七、消息过滤FilterServer

这里看下基于类模式的消息过滤模式。基于类模式的过滤是指在Broker端运行1个或多个消息过滤服务器（FilterServer），RocketMQ允许消息消费者自定义消息过滤实现类，并将其代码上次到FilterServer上，消息消费者向FilterServer拉取消息，FilterServer将消息消费者的拉取命令，转发到Broker，然后对返回的消息执行过滤逻辑，最终将消息返回给消费端。

FilterServer的工作原理：
1）Broker进程所在的服务器会启动多个FilterServer进程
2）消费者在订阅消息主题时，会上次一个自定义的消息过滤实现类，FilterServer加载并实例化
3）消息消费者向FilterServer发送消息拉取请求，FilterServer接收到消息消费者消息拉取请求后，FilterServer将消息拉取请求转发给Broker，Broker返回消息后在FilterServer端执行消息过滤逻辑，然后返回符合订阅信息的消息给消息消费者进行消费。

通常消息消费者是直接向Broker订阅主题，然后从Broker上拉取消息。而类模式过滤的特别之处在于，消息消费者是从FilterServer拉取消息。

总结：类模式过滤，相比TAG模式过滤，有如下优势：
1）基于TAG模式，由于在消息服务端进行消息过滤，是匹配消息TAG的hashcode, 导致服务端过滤并不十分准确，从服务端返回的消息最终并不一定是消息消费者订阅的消息，造成网络带宽的浪费，而基于类模式的消息过滤，所有的过滤操作全部在FilterServer端进行。
2）由于FilterServer和Broker运行在同一台机器上，消息的传输是通过本地通信，不会浪费Broker的网络资源。

八、Rocket主从同步（HA）机制

8.1、RocketMQ主从复制原理

为了避免Broker单点故障引起Broker上的消息无法及时被消费，RocketMQ引入了Broker主备机制，即消息消费到达主服务器后，需要将消息同步到从服务器；如果主服务器Broker宕机后，消息消费者可以从从服务器拉取消息。

RocketMQ的HA核心实现类图如下：

RocketMQ HA由7个核心类实现，分别为：
1）HAService：RocketMQ主从同步核心实现类
2）HAService#AcceptSocketService：HA Master端监听客户端连接实现类
3）HAService#GroupTransferService：主从同步通知实现类
4）HAService#HAClient：HA CLient端实现类
5）HAConnection：HA Master服务端HA连接对象的峰值，与Broker从服务器的网络读写实现类
6）HAConnection#ReadSocketService：HA Master网络读实现类
7）HAConnection#WriteSocketService：HA Master网络写实现类

RocketMQ HA实现原理如下：
1）主服务器启动，并在特定端口上监听从服务器的连接
2）从服务器主动连接主服务器，主服务器接收客户端的连接，并建立相关TCP连接
3）从服务器主动向主服务器发送待拉取消息偏移量，主服务器解析请求，并返回消息给从服务器
4）从服务器保存消息，并继续发送新的消息同步请求。

九、RocketMQ事务消息

9.1、事务消息实现思想

RocketMQ事务消息的实现原理是基于两阶段提交和定时事务状态回查，来决定消息最终提交还是回滚。

1）应用程序在事务内完成相关业务数据落库后，需要同步调用RocketMQ消息发送接口，发送状态为prepare消息。消息发送成功后，RocketMQ服务器会回调RocketMQ消息发送者的事件监听程序，记录消息的本地事务状态，该相关标记与本地业务操作，同属一个事务，确保消息发送与本地事务的原子性。
2）RocketMQ在收到类型为prepare的消息时，会首先备份消息的原主题与原消息消费队列，然后将消息存储在主题为RMQ_SYS_TRANS_HALF_TOPIC的消息消费队列中
3）RocketMQ消息服务器开启一个定时任务，消费RMQ_SYS_TRANS_HALF_TOPIC的消息，向消息发送端（应用程序）发起消息事务状态回查，应用程序根据保存的事务状态，反馈消息服务器事务的状态（提交、回滚、未知），如果是提交或回滚，则消息服务器提交或回滚消息，如果是未知，待下一次回查，RocketMQ允许设置一条消息的互查间隔与回查次数，如果在超时回查次数后依然无法获知消息的事务状态，则默认回滚消息。

总结：RocketMQ事务消息时基于两阶段提交和事务状态回查机制来实现，所谓两阶段提交，即首先发送prepare消息，待事务提交或回滚是发送commit、rollback命令。再结合定时任务，RocketMQ使用专门的线程以特定的频率对RocketMQ服务器上的prepare信息进行处理，向发送端查询事务消息的状态来决定是否提交或回滚消息。

十、RocketMQ实战

10.1、消息批量发送

RocketMQ消息批量发送是将同一topic的多条消息一起打包，发送到消息服务端，减少网络调用次数。
代码如下：

public class SimpleBatchProducer {
	public static void main (String[] args) throws Exception {
		DefaultMQProducer producer = new DefaultMQProducer("BatchProducerGroupName");
		producer.setNamesrvAddr("127.0.0.1:9876");
		producer.start();

		String topic = "BatchTest";
		List<Message> messages = new ArrayList<>();
		messages.add(new Message(topic, "Tag", "OrderID001", "Hello world1".getBytes()));
		messages.add(new Message(topic, "Tag", "OrderID002", "Hello world2".getBytes()));
		messages.add(new Message(topic, "Tag", "OrderID003", "Hello world3".getBytes()));
		System.out.println(producer.send(message));
		producer.shutdown();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

10.2、消息发送队列自选择

消息发送默认根据topic的路由信息（topic消息队列）进行负载均衡，负载均衡机制为轮询。例如现在有这样一个场景，订单的状态变更消息发送到特定主题，为了避免消息消费者同时消费同一订单的不同状态的变更消息，我们应该使用顺序消息。为了提高消息消费的并发度，如果我们根据某种负载算法，相同订单的不同消息能同一发到同一个消息消费队列上，则可以避免引入分布式锁，RocketMQ在消息发送时，提供了消息队列选择器MessageQueueSelector。
消息发送自定义分片算法，代码如下：

String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 100; i++) {
	int orderId = i % 10;
	Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i, (Hello RocketMQ " + i).getBytes(RomotingHelper.DEFAULT_CHARSET));
	SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
		public MessageQueue select (List<MessageQueue> msg, Message msg, Object arg) {
			Integer id = (Integer)arg;
			int index = id & msg.size();
			return msg.get(index);
		}
	}, orderId);
}
1
2
3
4
5
6
7
8
9
10
11
12

10.3、消息过滤

10.3.1、TAG模式过滤

消息发送者发送带TAG的消息：

for (int i = 0; i < 10; i++) {
	if (i % 2 == 0) {
		Message msg = new Message("TopicFilter7", "TOPICA_TAG_ALL", "OrderID001", "Helloworld".getBytes(RomotingHelper.DEFAULT_CHARSET));
		System.out.println("%s%n", "producer.send(msg));
	} else {
		Message msg = new Message("TopicFilter7", "TOPICA_TAG_ORD", "OrderID001". "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET));
		System.out.println("%s%n", producer.send(msg));
	}
}
1
2
3
4
5
6
7
8
9

Step1: 在发送消息时，我们可以为每条消息设置一个Tag标签，消息消费者订阅自己感兴趣的Tag，一般使用的场景是，对于同一类的功能（数据同步）创建一个主题，但对于该主题下的数据，可能不同的系统关心的数据不一样，但基础数据各个系统都需要同步，设置标签为TOPICA_TAG_ALL, 而订单数据只有订单下游子系统关系，其他系统并不关系，则设置标签为TOPICA_TAG_ORD, 库存子系统则关注库存相关的数据，设置标签为TOPICA_TAG_CAPCITY。

消息消费者订阅TAG模式：

// 订单系统消费组
DefaultMQPushConsumer orderConsumer = new DefaultMQPushConsumer("Order_Data_Syn");
orderConsumer.subscribe("TopicFilter7", "TOPICA_TAG_ALL | TOPIC_TAG_ORD");
// 库存子系统消费组
DefaultMQPushConsumer kuCunConsumer = new DefaultMQPushConsumer("CAPACITY_Data_Syn");
kuCunConsumer.subscribe("TopicFilter7", "TOPIC_TAG_ALL | TOPIC_TAG_CAPCITY");
1
2
3
4
5
6

Step2: 消费者组订阅相同的topic，不同的TAG，多个TAG用 ”|“ 分割，注意：同一个消费者组订阅的topic，tag必须相同。

10.3.2、SQL表达式过滤

SQL表达式消息发送方式：

Message msg = new Message ("TopicTest" /*Topic*/, "TagA" /*Tag*/, (Hello RocketMQ " + i).getBytes(RomotingHelper.DEFAULT_CHARSET));
msg.putUserProperty("orderStatus", "1");
SendResult sendResult = producer.send(msg);
System.out.println("%s%n", sendResult);
1
2
3
4

Step1: 基于SQL92表达式消息过滤，其实是对消息的属性运用SQL过滤表达式进行调节匹配，所以消息发送时，应该调用putUserProperty方法设置消息属性。

基于SQL过滤消息消费者构建示例：

consumer.subscribe("TopicTest", MessageSelector.bySql("(orderStatus is not null and orderStatus > 0 )"));
1

Step2: 订阅模式为一条SQL条件过滤表达式，上下文环境为消息的属性。

10.3.3、类过滤模式

Step1: 实现自定义消息过滤器，实现org.apache.rocketmq.common.filter.MessageFilter, MessageExt实例中封装了整体消息的所有信息。
自定义消息过滤类，实现org.apache.rocketmq.common.filter.MessageFilter接口

RocketMQ通过定义消息过滤类的接口来实现自定义消息过滤，代码如下：

package org.apache.rocketmq.example.filter;
import org.apache.rocketmq.common.filter.FilterContext;
import org.apache.rocketmq.filter.MessageFilter;
public class MessageFilterImpl implements MessageFilter {
	@Override
	public boolean match(MessasgeExt msg, FilterContext context) {
		String property = msg.getProperty("SequenceId");
		if (property != null) {
			int id = Integer.parseInt(property);
			if (((id % 10) == 0) && (id > 100)) {
				return true;
			}
		}
		return false;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

Step2：消息消费者订阅主题，并上传自定义订阅类源码

DefaultMQPushConsumer consumer = DefaultMQPushConsumer("ConsumerGroupNamecc4");
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
File classFile = new File(classLoader.getResource("MessageFilterImpl.java".getFile());
String filterCode = MixAll.file2String(classFile);
consumer.subscribe("TopicTest", "org.apache.rocketmq.example.filter.MessageFilterImpl", filterCode);
1
2
3
4
5

10.3.4、事务消息

以一个订单流转流程来距离，例如订单子系统创建订单，需要将订单数据下发到其他子系统（与第三方系统对接）这个场景。我们通常会将两个系统进行解耦，不直接使用服务调用的方式进行交互。其业务的实现步骤通常有下面几步：
1）A系统创建订单并入库
2）发送消息到MQ
3）MQ消费者消费消息，发送远程RPC服务调用，完成订单数据的同步。

这里，”创建订单并入本地库“ 与 ”发送消息到MQ，远程消费MQ“ 应该是个事务。

方案1：
1、执行下订单相关业务流，例如操作本地数据库落库相关代码
2、调用消息发送端API发送消息到MQ
3、返回结果，提交事务。

方案1的弊端：
1、如果消息发送到MQ成功，在提交事务时，JVM突然挂掉，事务没有提交成功，导致两个系统之间数据不一致。（A系统中由于没有提交事务，导致订单不存在；但写入了MQ，B系统能消费到，所以B系统里有这个订单）
2、由于消息是在事务提交之前发送，发送的消息内容是订单实体的内容，会造成在消费端进行消费时，如果需要验证订单是否存在时，可能出现订单不存在的情况。

方案2：
在MQ不支持事务消息的前提下，可采用下面的方式优化：
1、执行下订单相关的业务流程，例如操作本地数据库落库相关代码
2、生成事务消息唯一业务标识，将该业务组装到待发送的消息体中
3、往待发送消息标中插入一条记录，本次唯一消息发送业务ID，消息JSON（消息主题、消息tag、消息体）、创建时间、发送状态
4、将消息体返回控制器层
5、返回结果，提交事务
然后在控制器层异步发送消息，同时需要引入定时机制，去扫描待发送消息记录，避免消息丢失。

方案2弊端：
1、消息有可能重复发送，但在消费端可通过唯一业务变化来进行去重
2、实现过于复杂，为了避免极端情况下消息丢失，需要使用定时任务

方案3：基于RocketMQ4.3版本事务消息
1、执行下订单相关的业务流程，例如操作本地数据库落库相关代码
2、生成事务消息唯一业务标识，将该业务标识组装到待发送的消息体中，方便消息端进行幂等消费
3、调用消息客户端API，发送事务prepare消息消费
上述是第一步，发送事务消息，接下来需要实现TransactionListener，实现执行本地事务与本地事务回查。代码如下：

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;
@SuppressW arnings( "unused ")
public class OrderTransactionListenerImpl implements TransactionListener {
	private ConcurrentHashMap<Str工ng, Integer> countHashMap = new ConcurrentHashMap<>();
	private final static int MAX_COUNT = 5; 

	@Override
	public LocalTransactionState executeLocalTransaction (Message msg, Object arg) {
		String bizUniNo = msg.getUserProperty(”bizUniN。”); //从消息中获取业务唯一ID 
		// 将 bizUniNo 入库，表名 : t message transaction， 表结构 bizUniNo ( 主键 }， 业务类型。 
		return LocalTransactionState.UNKNOW;
	}

	@Override
	public LocalTransactionState checkLocalTransaction(MessageExt msg) {
		Integer status = O;
		// 从数据库查询 t_message_transaction表，如果该表中存在记录，则提交，
		String bizUniNo = msg.getUserProperty(”bizUniN。”); //从消息中获取业务唯一 ID。
		// 然后查询 t_message_transaction表 ，是否存在bizUniNo，如果存在，则返回COMMIT_MESSAGE,
		// 不存在，则记录查询次数，未超过次数，返回UNKNOW，超过次数，返回ROLLBACK_MESSAGE 
		if(query(bizUniNo) > 0) {
			return LocalTransactionState.ROLLBACK_MESSAGE;
		}
		return rollBackOrUnown(bizUniNo);
	}

	public int query (String bizUniNo) {
		return 1; // select count(1) from t_message_transaction a where a.biz_uni_no=#{bizUniNo}
	}

	public LocalTransactionState rollBackOrUnown(String bizUniNo) {
		Integer num = countHashMap.get(bizUniNo);
		if (num != null && ++num > MAX_COUNT) {
			countHashMap.remove(bizUniNo);
			return LocalTransactionState.ROLLBACK_MESSAGE;
		}
		if (num == null) {
			num = new Integer();
		}
		countHashMap.put(bizUniNo, num);
		return LocalTransactionState.UNKNOW;
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

TransactionListener实现要点：
1）executeLocalTransaction：该方法主要设置本地事务状态，与业务方代码在一个事务中，只要本地事务提交成功，该方法也会提交成功。所以在这里，主要是向t_message_transation添加一条记录，在事务回查是，如果存在记录，就任务该消息需要提交，其返回值建议返回LocalTransactionState.UNKNOW.
2）checkLocalTransaction：该方法主要是告知RocketMQ消息是需要提交还是回滚，如果本地事务表t_message_transaction存在记录，则认为提交；如果不存在，可以设置回查次数，如果指定次数内还是未查到消息，则回滚，否则返回未知。RocketMQ会按一定的频率回查事务，当然回查次数也有限制，默认为5次，可配置。