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中为拉取消息位置。
虽然每个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();
}
【MetaQ原理简介(一)】putMessageLock这里提供了两种上锁方式,一种是默认的自旋锁,使用compareAndSet实现(用于low-race condition);一种是可重入锁,使用ReentrantLock实现
参考
- RocketMQ_原理简介
推荐阅读
- 做一件事情的基本原理是什么()
- 【读书笔记】贝叶斯原理
- SG平滑轨迹算法的原理和实现
- “写作宝典”《金字塔原理》之读书笔记
- java简介|Java是什么(Java能用来干什么?)
- 7、前端--jQuery简介、基本选择器、基本筛选器、属性选择器、表单选择器、筛选器方法、节点操作、绑定事件
- 什么是张拉膜结构雨棚(有哪些特点?)
- Spring|Spring 框架之 AOP 原理剖析已经出炉!!!预定的童鞋可以识别下发二维码去看了
- ACI注册国际心理咨询师简介和可靠性分析
- Spring|Spring Boot 自动配置的原理、核心注解以及利用自动配置实现了自定义 Starter 组件