MetaQ原理简介（一）

MetaQ是阿里巴巴中间件团队开发的一款消息队列中间件，说起MetaQ的命名呢，也是有点意思。MetaQ最早是基于Kafka的设计并使用Java进行了完全重写，而Kafka（卡夫卡）作家最著名的作品，大家都清楚，叫做《变形记》，英文名叫Metamorphosis。

MetaQ集群架构

---

MetaQ集群架构

• NameServer集群：MetaQ基于NameServer，也是基于阿里内部中间件Config Server。可以把它理解为类似zookeeper的角色
• Broker：消息中转角色，负责存储消息，转发消息
• Consumer：消息消费者，负责消费消息，一般是后台系统负责异步消费。MetaQ提供两种消费模型
  • Push Consumer ：向Consumer对象注册一个Listener接口，收到消息后回调Listener接口方法，采用长轮询实现push
  • Pull Consumer：主动由Consumer主动拉取信息，同kafka
• Producer：消息生产者，负责产生消息，一般由业务系统负责产生消息

消息结构模型

• Message：单位消息
• Topic：消息主题，软分区，对应相同的topic时，生产者对应消费者的分区标识
• Tag：消息在topic基础上的二级分类
• Message Queue：硬分区，物理上区分topic，一个topic对应多个message queue。在 MetaQ 中，所有消息队列都是持久化，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用 Offset 来访问，offset 为 java long 类型，64 位，理论上在 100 年内不会溢出，所以认为是长度无限，另外队列中只保存最近几天的数据，之前的数据会按照过期时间来 删除。
• Group：Consumer Group，一类 Consumer 的集合名称，这类 Consumer 通常消费一类消息，且消费逻辑一致；Producer Group，一类 Producer 的集合名称，这类 Producer 通常发送一类消息，且发送逻辑一致。
• Offset：绝对偏移值，message queue中有两类offset（commitOffset和offset），前者存储在OffsetStore中表示消费到的位置，后者是在PullRequest中为拉取消息位置。

Broker

---

Broker以组为单位向Consumer提供消息服务，group中分为master和slave两种角色。然后通过NameServer暴露给Consumer具体通信地址，采用message queue消息队列结构来提供消费接口。针对某一topic情况下，message queue会根据queue id分布在不同的broker上，Consumer的消息消费压力则会分摊在不同的Broker上的message queue，从而达到负载均衡的作用。
虽然每个topic下面有很多message queue，但是message queue本身并不存储消息。真正的消息存储会写在CommitLog的文件，message queue只是存储CommitLog中对应的位置信息，方便通过message queue找到对应存储在CommitLog的消息。不同的topic，message queue都是写到相同的CommitLog 文件，也就是说CommitLog完全的顺序写，而顺序读写是metaq高吞吐量的基础。

Broker存储结构

Broker存储结构

• 重试队列：%RETRY%+consumergroup，push consumer默认订阅用于消费失败后的重试消费
• 死信队列：多次（默认16次）消费失败后进入DLQ队列，需要人工处理
• 定时队列：用于定时和延时消息
• ConsumeQueue: 即message queue，根据topic和queueId区分的消息队列，对MappedFileQueue进行封装
• CommitLog： Broker中顺序存储的消息结构，管理消息commit和flush，对MappedFileQueue进行封装
• MappedFileQueue： 对~/store/commitlog/中MappedFile封装成文件队列，进行文件大小格式检查，对mappedFile进行管理。
• MappedFile： 实际broker数据文件映射成的类，即~/store/commitlog/中00000000000000000000、00000000001073741824等文件,每个文件默认大小上限为1G。

消息写入

CommitLog负责将Producer的消息写入文件中

消息写入

核心代码如下

putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        try {
            long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
            this.beginTimeInLock = beginLockTimestamp;

            // Here settings are stored timestamp, in order to ensure an orderly
            // global
            msg.setStoreTimestamp(beginLockTimestamp);

            if (null == mappedFile || mappedFile.isFull()) {
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
            if (null == mappedFile) {
                log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
            }

            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            switch (result.getStatus()) {
                case PUT_OK:
                    break;
                case END_OF_FILE:
                    unlockMappedFile = mappedFile;
                    // Create a new file, re-write the message
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                    if (null == mappedFile) {
                        // XXX: warn and notify me
                        log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                        beginTimeInLock = 0;
                        return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
                    }
                    result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                    break;
                case MESSAGE_SIZE_EXCEEDED:
                case PROPERTIES_SIZE_EXCEEDED:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
                case UNKNOWN_ERROR:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
                default:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
            }

            eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
            beginTimeInLock = 0;
        } finally {
            putMessageLock.unlock();
        }

putMessageLock这里提供了两种上锁方式，一种是默认的自旋锁，使用compareAndSet实现（用于low-race condition）；一种是可重入锁，使用ReentrantLock实现

参考

---

• RocketMQ_原理简介