MetaQ原理简介(一)

MetaQ是阿里巴巴中间件团队开发的一款消息队列中间件,说起MetaQ的命名呢,也是有点意思。MetaQ最早是基于Kafka的设计并使用Java进行了完全重写,而Kafka(卡夫卡)作家最著名的作品,大家都清楚,叫做《变形记》,英文名叫Metamorphosis。
MetaQ集群架构 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中分为masterslave两种角色。然后通过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存储结构
MetaQ原理简介(一)
文章图片
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的消息写入文件中

MetaQ原理简介(一)
文章图片
消息写入
核心代码如下
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(); }

【MetaQ原理简介(一)】putMessageLock这里提供了两种上锁方式,一种是默认的自旋锁,使用compareAndSet实现(用于low-race condition);一种是可重入锁,使用ReentrantLock实现
参考
  • RocketMQ_原理简介

    推荐阅读