RocketMQ原理分享

之前在组内分享过一次RocketMQ，虽然写了PPT，还是口述+黑板画图，现场分享效果更佳。RocketMQ架构清晰，只有NameServer、Broker、Producer和Consumer四个组件，通过共享commitlog，即使数十万topic，依然能保持高性能；是通过工程巧妙解决业务难题的典范。RocketMQ真正做到了“简单”，例如重复消费这种功能，完全抛给消费侧；通过构建逻辑队列，每个消息都会指定唯一队列，解决消息顺序的问题。

下面介绍整体架构和四个组件，然后分析一条消息从诞生到被消费的全过程，从中了解RocketMQ设计的精妙。

整体架构

pic1

NameServer在内存中保存broker路由信息，broker定期向其发送心跳包，保证连接；NameServer存储的路由包括：topic及其队列(包括集群信息)、brokername和机器ip（master/slave）、cluster name和broker name信息、broker探活信息等。

broker存储数据，包括消息、consumer和订阅关系、消费偏移、延迟队列消费进度、topic信息（namesrv保存的是topic的集群信息，具体数据都在broker）。broker和namesrv之间有长连接。

producer和consumer从namesrv获取broker地址并向broker发送请求。

Broker

每台机器在配置文件中指定相同的broker_name，名字相同的组成master+slave结构的broker。主要功能如下：

  • 消息持久化 —— 后面着重介绍
  • 保存下面这些重要信息，slave会定时请求master同步：
    • topic：读写queue的数量和权限等（每个topic都由多个queue组成，每个queue有queueid）
    • consumer group：包括组内消费者、订阅关系、消费进度、消费方式（是否广播模式、pull或push）、消费失败重试次数、broker切换等
    • producer group：缓存消费组对应的生产者信息

pic2

NameServer

NameServer作为名字中心，管理broker路由信息、topic（与其对应queue）集群信息，producer和consumer通过它获取broker地址。

  • 无状态，只在内存中保存数据；且节点之间没有相互通信；重启之后，只需等待broker重新注册即可
  • broker与所有nameserver保持长连接，定期更新自身信息
  • 对broker有存活检测机制，摘除不活跃的broker地址
public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    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;
    ...
}

Producer

Producer启动时需要指定nameserver地址，并与其中一个server建立长连接。定期从nameserver获取topic对应的broker路由信息，与所有master节点建立长连接。

public void updateTopicRouteInfoFromNameServer() {
    Set<String> topicList = new HashSet<String>();

    ...

    // Producer
    {
        Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, MQProducerInner> entry = it.next();
            MQProducerInner impl = entry.getValue();
            if (impl != null) {
                Set<String> lst = impl.getPublishTopicList();
                topicList.addAll(lst);
            }
        }
    }

    for (String topic : topicList) {
        this.updateTopicRouteInfoFromNameServer(topic);
    }
}

Producer生产消息需指定topic+queueid，queueid由producer发送时指定，兼具负载均衡和顺序发送。

public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) {
    TopicPublishInfo info = new TopicPublishInfo();
    info.setTopicRouteData(route);
    if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) {
        String[] brokers = route.getOrderTopicConf().split(";");
        for (String broker : brokers) {
            String[] item = broker.split(":");
            int nums = Integer.parseInt(item[1]);
            for (int i = 0; i < nums; i++) {
                MessageQueue mq = new MessageQueue(topic, item[0], i);
                info.getMessageQueueList().add(mq);
            }
        }

        info.setOrderTopic(true);
    } else {
        ...
    }

    return info;
}

Consumer

Consumer连接Broker和NameServer与Producer相同，也在updateTopicRouteInfoFromNameServer。获取消息有两种模式：pull和push，不过最终都是类似pull的方式，拉取消息。

Consumer从broker获取Consumer Group内所有消费者数量，以及topic的MessageQueueList，在本地做负载均衡；消费进度Offset会保存在本地作为参考，定期与broker同步；消费失败或其它异常，broker会把消息放到重试队列或者死信队列。

消息流

消息从“生”到“死”的整个过程如下，为保证完整，topic创建等内容也有所涉及。

topic创建

包括topic创建、消费组定义以及订阅关系产生。

管理工具发起创建topic命令，指定cluster和queue数量、topic对应queue数量；然后请求NameServer获取broker地址；通过UPDATE_AND_CREATE_TOPIC请求，broker更新config并向NameServer注册，后者保存topic信息。

多主架构中，每个master都有独立的topic队列，共同组成全量数据。

fetchMasterAddrByClusterName -> createTopic -> registerBroker

消息发送

生产者调用getRouteInfoByTopic(GET_ROUTEINTO_BY_TOPIC请求)从NameServer获取topic路由信息TopicRouteData;然后调用topicRouteData2TopicPublishInfo方法得到包含MessageQueue list的TopicPublishInfo。

public class TopicRouteData extends RemotingSerializable {
    private String orderTopicConf;
    private List<QueueData> queueDatas;
    private List<BrokerData> brokerDatas;
    private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
    ...
}

public class QueueData implements Comparable<QueueData> {
    private String brokerName;
    private int readQueueNums;
    private int writeQueueNums;
    private int perm;
    private int topicSynFlag;
    ...
}

MessageQueue与master*queue number对应，selectOneMessageQueue通过轮询方式选择要把消息发往哪个broker的哪个queue。

消息存储

每台master有一个commitlog，所有topic共享，收到消息后写入，并定期刷盘（多种刷盘策略）；后台线程定期对每个topic生成consume squeue，存储commitlog偏移，供消费者消费（取出offset，然后读commitlog）；利用定长的内存映射文件（1G），写入速度非常快，读操作也只是读内存。

上层所谓offset偏移，一般指queue队列偏移，而非commitlog文件偏移。

public class DefaultMessageStore implements MessageStore {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    private final MessageStoreConfig messageStoreConfig;
    // CommitLog
    private final CommitLog commitLog;
    private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueue>> consumeQueueTable;
    private final FlushConsumeQueueService flushConsumeQueueService;
    private final CleanCommitLogService cleanCommitLogService;
    private final CleanConsumeQueueService cleanConsumeQueueService;
    private final IndexService indexService;
    private final AllocateMappedFileService allocateMappedFileService;
    private final ReputMessageService reputMessageService;
    private final HAService haService;
    private final ScheduleMessageService scheduleMessageService;
    private final StoreStatsService storeStatsService;
    private final TransientStorePool transientStorePool;
    ...
}

ReputMessageService的doReput方法定期把commitlog写入，实际执行写入的是CommitLogDispatcherBuildConsumeQueue类。

commitlog文件结构如下，多种刷盘策略：同步刷盘或异步刷盘，同时CommitRealTimeService定期写入fileChannel（并不是立即刷盘，会有自己的缓存）。

pic3

Checkpoint文件记录commitlog、consume queue刷盘时间，用于恢复数据。

高可用

Rocketmq高可用通过master broker和slave broker同步数据（topic配置、消费进度、订阅关系）实现，slave从nameserver获取master地址，定期请求master获取config和消息（两者同步频率不同）；master根据同步策略向slave同步发送消息或者定期更新消息内容。

消息消费

消费者通过rebalance模块实现负载均衡，获取属于自己的consume queue，并创建对应的数据请求：从nameserver获取broker地址，然后请求数据，将数据放入ProcessQueue作为缓存。

ConsumeMessageService服务将缓存的数据发送给用户，并更新offset；消息拉取、负载均衡、消息消费三个步骤，都是通过一个PullRequest队列异步完成的。新的消费者或broker加入时，会通知所有消费者，重新执行rebalance操作。

pic4

RocketMQ其它特色

  • 事务消息：通过 prepare消息——执行事务——commit消息三个阶段完成，以写入consume queue为准
  • 延时消费：线程定时扫描不同粒度的延时队列，写入consume queue
  • 全局commitlog：读写分离，topic数量不受限制
  • 广播消费：消费组管理offset，没有消息重试
  • 顺序消费：利用tag字段，producer写入同一个queue，consumer用锁保证消费