asyncPutMessage方法真正的用来存储消息。
1.asyncPutMessage存储普通消息
DefaultMessageStore#asyncPutMessage()
1、 checkStoreStatus,checkMessage,checkLmqMessage校验；
2. CommitLog#asyncPutMessage存储消息, 更新耗时时间和失败次数。
@Override
public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
PutMessageStatus checkStoreStatus = this.checkStoreStatus();
if (checkStoreStatus != PutMessageStatus.PUT_OK) {
return CompletableFuture.completedFuture(new PutMessageResult(checkStoreStatus, null));
}
PutMessageStatus msgCheckStatus = this.checkMessage(msg);
if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null));
}
long beginTime = this.getSystemClock().now();
CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
putResultFuture.thenAccept((result) -> {
long elapsedTime = this.getSystemClock().now() - beginTime;
if (elapsedTime > 500) {
log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
}
this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);
if (null == result || !result.isOk()) {
this.storeStatsService.getPutMessageFailedTimes().add(1);
}
});
return putResultFuture;
}
1.1 checkStoreStatus检查存储状态
1、 如果DefaultMessageStore是shutdown状态,返回SERVICE_NOT_AVAILABLE；
2、 如果broker是SLAVE角色,返回SERVICE_NOT_AVAILABLE,不能将消息写入SLAVE角色；
3、 如果不支持写入,返回SERVICE_NOT_AVAILABLE,可能因为broker的磁盘已满、写入逻辑队列错误、写入索引文件错误等等原因‘；
4、 如果操作系统页缓存繁忙,返回OS_PAGECACHE_BUSY,如果broker持有锁的时间超过osPageCacheBusyTimeOutMills,算作操作系统页缓存繁忙；
5、 最后返回PUT_OK,表示可用存储消息；
/**
* DefaultMessageStore的方法
* <p>
* 检查存储状态
*/
private PutMessageStatus checkStoreStatus() {
//如果DefaultMessageStore是shutdown状态，返回SERVICE_NOT_AVAILABLE
if (this.shutdown) {
log.warn("message store has shutdown, so putMessage is forbidden");
return PutMessageStatus.SERVICE_NOT_AVAILABLE;
}
//如果broker是SLAVE角色，则返回SERVICE_NOT_AVAILABLE，不能将消息写入SLAVE角色
if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
long value = this.printTimes.getAndIncrement();
if ((value % 50000) == 0) {
log.warn("broke role is slave, so putMessage is forbidden");
}
return PutMessageStatus.SERVICE_NOT_AVAILABLE;
}
//如果不支持写入，那么返回SERVICE_NOT_AVAILABLE
//可能因为broker的磁盘已满、写入逻辑队列错误、写入索引文件错误等等原因
if (!this.runningFlags.isWriteable()) {
long value = this.printTimes.getAndIncrement();
if ((value % 50000) == 0) {
log.warn("the message store is not writable. It may be caused by one of the following reasons: " +
"the broker's disk is full, write to logic queue error, write to index file error, etc");
}
return PutMessageStatus.SERVICE_NOT_AVAILABLE;
} else {
this.printTimes.set(0);
}
//如果操作系统页缓存繁忙，则返回OS_PAGECACHE_BUSY
//如果broker持有锁的时间超过osPageCacheBusyTimeOutMills，则算作操作系统页缓存繁忙
if (this.isOSPageCacheBusy()) {
return PutMessageStatus.OS_PAGECACHE_BUSY;
}
//返回PUT_OK，表示可以存储消息
return PutMessageStatus.PUT_OK;
}
1.2 checkMessage检查消息
1、 如果topic长度大于127,返回MESSAGE_ILLEGAL；
2、 如果设置的属性长度大于32767,返回MESSAGE_ILLEGAL,properties过长；
3、 返回PUT_OK,检查通过；
/**
* DefaultMessageStore的方法
* <p>
* 检查消息
*/
private PutMessageStatus checkMessage(MessageExtBrokerInner msg) {
//如果topic长度大于127，则返回MESSAGE_ILLEGAL，表示topic过长了
if (msg.getTopic().length() > Byte.MAX_VALUE) {
log.warn("putMessage message topic length too long " + msg.getTopic().length());
return PutMessageStatus.MESSAGE_ILLEGAL;
}
//如果设置的属性长度大于32767，则返回MESSAGE_ILLEGAL，表示properties过长了
if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
log.warn("putMessage message properties length too long " + msg.getPropertiesString().length());
return PutMessageStatus.MESSAGE_ILLEGAL;
}
return PutMessageStatus.PUT_OK;
}
2.CommitLog#asyncPutMessage异步存储消息
1、 处理延迟消息的逻辑；
1、 如果是延迟消息,即DelayTimeLevel大于0,替换topic为SCHEDULE_TOPIC_XXXX,替换queueId为延迟队列id,id=level-1,如果延迟级别大于最大级别,则设置为最大级别18,默认延迟2h,这些参数可以在broker端配置类MessageStoreConfig中配置；
2、 最后保存真实topic到消息的REAL_TOPIC属性,保存queueId到消息的REAL_QID属性,方便后面恢复；
2、 消息编码获取线程的本地变量,包含一个线程独立的encoder和keyBuilder对象将消息内容编码,存储到encoder中的encoderBuffer中,它是通过ByteBuffer.allocateDirect(size)得到的一个直接缓冲区消息写入之后,调用encoderBuffer.flip(),将Buffer从写模式切换到读模式,可以读取到数据；
3、 加锁并写入消息；
1、 一个broker将所有的消息都追加到同一个逻辑CommitLog日志文件中,需要通过获取putMessageLock锁来控制并发一种锁是ReentrantLock可重入锁,一种是CAS锁,根据StoreConfig的useReentrantLockWhenPutMessage决定是否是ReentrantLock锁,默认为true,使用ReentrantLock；
2、 从mappedFileQueue中的mappedFiles集合中获取最后一个MappedFile如果最新的mappedFile为null,或者mappedFile满了,会新建mappedFile；
3、 通过mappedFile调用appendMessage方法追加消息,仅仅是追加消息到byteBuffer的内存中如果是writeBuffer则表示消息写入了堆外内存中,如果是mappedByteBuffer,则表示消息写入了pagechache中；
4、 追加成功之后解锁如果是剩余空间不足,则会重新初始化一个MappedFile并再次尝试追加；
4、 如果存在写满的MappedFile并且启用了文件内存预热,那么这里调用unlockMappedFile对MappedFile执行解锁；
5、 更新消息统计信息随后调用submitFlushRequest方法提交刷盘请求,将会根据刷盘策略进行刷盘随后调用submitReplicaRequest方法提交副本请求,用于主从主从同步；
/**
* CommitLog的方法
* <p>
* 异步存储消息
*
* @param msg
* @return
*/
public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
// Set the storage time
//设置存储时间
msg.setStoreTimestamp(System.currentTimeMillis());
// Set the message body BODY CRC (consider the most appropriate setting
// on the client)
//设置消息正文CRC
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
// Back to Results
AppendMessageResult result = null;
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
String topic = msg.getTopic();
// int queueId msg.getQueueId();
/*
* 1 处理延迟消息的逻辑
*
* 替换topic和queueId，保存真实topic和queueId
*/
//根据sysFlag获取事务状态，普通消息的sysFlag为0
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
//如果不是事务消息，或者commit提交事务小i
if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
|| tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
// Delay Delivery
//获取延迟级别，判断是否是延迟消息
if (msg.getDelayTimeLevel() > 0) {
//如果延迟级别大于最大级别，则设置为最大级别
if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
}
//获取延迟队列的topic，固定为 SCHEDULE_TOPIC_XXXX
topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
//根据延迟等级获取对应的延迟队列id， id = level - 1
int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// Backup real topic, queueId
//使用扩展属性REAL_TOPIC 记录真实topic
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
//使用扩展属性REAL_QID 记录真实queueId
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
//更改topic和queueId为延迟队列的topic和queueId
msg.setTopic(topic);
msg.setQueueId(queueId);
}
}
//发送消息的地址
InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
if (bornSocketAddress.getAddress() instanceof Inet6Address) {
msg.setBornHostV6Flag();
}
//存储消息的地址
InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
if (storeSocketAddress.getAddress() instanceof Inet6Address) {
msg.setStoreHostAddressV6Flag();
}
/*
* 2 消息编码
*/
//获取线程本地变量，其内部包含一个线程独立的encoder和keyBuilder对象
PutMessageThreadLocal putMessageThreadLocal = this.putMessageThreadLocal.get();
//将消息内容编码，存储到encoder内部的encoderBuffer中，它是通过ByteBuffer.allocateDirect(size)得到的一个直接缓冲区
//消息写入之后，会调用encoderBuffer.flip()方法，将Buffer从写模式切换到读模式，可以读取到数据
PutMessageResult encodeResult = putMessageThreadLocal.getEncoder().encode(msg);
if (encodeResult != null) {
return CompletableFuture.completedFuture(encodeResult);
}
//编码后的encoderBuffer暂时存入msg的encodedBuff中
msg.setEncodedBuff(putMessageThreadLocal.getEncoder().encoderBuffer);
//存储消息上下文
PutMessageContext putMessageContext = new PutMessageContext(generateKey(putMessageThreadLocal.getKeyBuilder(), msg));
/*
* 3 加锁并写入消息
* 一个broker将所有的消息都追加到同一个逻辑CommitLog日志文件中，因此需要通过获取putMessageLock锁来控制并发。
*/
//持有锁的时间
long elapsedTimeInLock = 0;
MappedFile unlockMappedFile = null;
/*
* 有两种锁，一种是ReentrantLock可重入锁，另一种spin则是CAS锁
* 根据StoreConfig的useReentrantLockWhenPutMessage决定是否使用可重入锁，默认为true，使用可重入锁。
*/
putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
try {
/*
* 从mappedFileQueue中的mappedFiles集合中获取最后一个MappedFile
*/
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
//加锁后的起始时间
long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
this.beginTimeInLock = beginLockTimestamp;
// Here settings are stored timestamp, in order to ensure an orderly
// global
//设置存储的时间戳为加锁后的起始时间，保证有序
msg.setStoreTimestamp(beginLockTimestamp);
/*
* 如果最新mappedFile为null，或者mappedFile满了，那么会新建mappedFile并返回
*/
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());
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));
}
/*
* 追加存储消息
*/
result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
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());
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));
}
result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
break;
case MESSAGE_SIZE_EXCEEDED:
case PROPERTIES_SIZE_EXCEEDED:
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
case UNKNOWN_ERROR:
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
default:
return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
}
//加锁的持续时间
elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
} finally {
//重置开始时间，释放锁
beginTimeInLock = 0;
putMessageLock.unlock();
}
if (elapsedTimeInLock > 500) {
log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
}
//如果存在写满的MappedFile并且启用了文件内存预热，那么这里对MappedFile执行解锁
if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
}
PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
// Statistics
//存储数据的统计信息更新
storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).add(1);
storeStatsService.getSinglePutMessageTopicSizeTotal(topic).add(result.getWroteBytes());
/*
* 4 提交刷盘请求，将会根据刷盘策略进行刷盘
*/
CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
/*
* 5 提交副本请求，用于主从同步
*/
CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
if (flushStatus != PutMessageStatus.PUT_OK) {
putMessageResult.setPutMessageStatus(flushStatus);
}
if (replicaStatus != PutMessageStatus.PUT_OK) {
putMessageResult.setPutMessageStatus(replicaStatus);
}
return putMessageResult;
});
}
2.1 处理延迟消息
- 如果DelayTimeLevel大于0, 则表示延迟消息, topic换为"SCHEDULE_TOPIC_XXXX", queueId为延迟队列id, id = level - 1, 保存真实topic到消息的REAL_TOPIC属性, 保存queueId到消息的REAL_QID属性。
/*
* 1 处理延迟消息的逻辑
*
* 替换topic和queueId，保存真实topic和queueId
*/
//根据sysFlag获取事务状态，普通消息的sysFlag为0
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
//如果不是事务消息，或者commit提交事务消息
if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
|| tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
// Delay Delivery
//获取延迟级别，判断是否是延迟消息
if (msg.getDelayTimeLevel() > 0) {
//如果延迟级别大于最大级别，则设置为最大级别
if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
}
//获取延迟队列的topic，固定为 SCHEDULE_TOPIC_XXXX
topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
//根据延迟等级获取对应的延迟队列id， id = level - 1
int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// Backup real topic, queueId
//使用扩展属性REAL_TOPIC 记录真实topic
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
//使用扩展属性REAL_QID 记录真实queueId
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
//更改topic和queueId为延迟队列的topic和queueId
msg.setTopic(topic);
msg.setQueueId(queueId);
}
}
2.2 获取最新mappedFile
- 从mappedFileQueue中的mappedFiles集合中获取最后一个MappedFile。
/**
* MappedFileQueue的方法
* <p>
* 获取最新的MappedFile
*/
public MappedFile getLastMappedFile() {
MappedFile mappedFileLast = null;
while (!this.mappedFiles.isEmpty()) {
try {
//从mappedFiles中获取最后一个mappedFile
mappedFileLast = this.mappedFiles.get(this.mappedFiles.size() - 1);
break;
} catch (IndexOutOfBoundsException e) {
//continue;
} catch (Exception e) {
log.error("getLastMappedFile has exception.", e);
break;
}
}
return mappedFileLast;
}
- 最新mappedFile为null, 或者mappedFile满了, 创建mappedFile。
/**
* MappedFileQueue的方法
* <p>
* 创建新的MappedFile
*
* @param startOffset 指定起始offset
*/
public MappedFile getLastMappedFile(final long startOffset) {
return getLastMappedFile(startOffset, true);
}
/**
* MappedFileQueue的方法
* <p>
* 创建或者获取最新的MappedFile
*
* @param startOffset 起始offset
* @param needCreate 是否创建
*/
public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
long createOffset = -1;
//从mappedFiles集合中获取最后一个MappedFile
MappedFile mappedFileLast = getLastMappedFile();
//如果为null，那么设置创建索引，默认为0，即新建的文件为第一个mappedFile文件，从0开始
if (mappedFileLast == null) {
createOffset = startOffset - (startOffset % this.mappedFileSize);
}
//如果满了，那么设置新mappedFile文件的创建索引 = 上一个文件的起始索引（即文件名） + mappedFileSize
if (mappedFileLast != null && mappedFileLast.isFull()) {
createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
}
//如果需要创建新mappedFile，那么根据起始索引创建新的mappedFile
if (createOffset != -1 && needCreate) {
return tryCreateMappedFile(createOffset);
}
return mappedFileLast;
}
2.2.1 tryCreateMappedFile创建新的MappedFile#
- 获取下两个MappedFile的路径nextFilePath和nextNextFilePath, 然后调用doCreateMappedFile真正创建, 一次请求对于2个mappedFile, 2个commitlog。
- commitlog文件预创建或者文件预分配, 如果启用了MappedFile预分配服务, 那么在创建MappedFile时会同时创建两个MappedFile, 一个用于同步创建并返回用于本次实际使用, 一个后台异步创建用于下次取用。避免等到当前文件真正用完了才创建下一个文件, 提供性能。
/**
* MappedFileQueue的方法
* <p>
* 创建commitlog文件，映射MappedFile
*
* @param createOffset 起始索引，即新文件的文件名
*/
protected MappedFile tryCreateMappedFile(long createOffset) {
//下一个文件路径 {storePathCommitLog}/createOffset，即文件名为createOffset，即起始物理offset
String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
//下下一个文件路径 {storePathCommitLog}/createOffset+mappedFileSize，即文件名为createOffset + mappedFileSize，即起始offset
String nextNextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset
+ this.mappedFileSize);
//真正创建文件
return doCreateMappedFile(nextFilePath, nextNextFilePath);
}
判断如果allocateMappedFileService不为null, 那么异步的创建MappedFile, 否则同步创建MappedFile。
/**
* MappedFileQueue的方法
* <p>
* 创建commitlog文件，映射MappedFile
*
* @param nextFilePath 要创建的下一个文件路径
* @param nextNextFilePath 要创建的下下一个文件路径
*/
protected MappedFile doCreateMappedFile(String nextFilePath, String nextNextFilePath) {
MappedFile mappedFile = null;
//如果allocateMappedFileService不为null，那么异步的创建MappedFile
//CommitLog的MappedFileQueue初始化时会初始化allocateMappedFileService，因此一般都不为null
if (this.allocateMappedFileService != null) {
//添加两个请求到处理任务池，然后阻塞等待异步创建默认1G大小的MappedFile
mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
nextNextFilePath, this.mappedFileSize);
} else {
try {
//同步创建MappedFile
mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
} catch (IOException e) {
log.error("create mappedFile exception", e);
}
}
if (mappedFile != null) {
//如果是第一次创建，那么设置标志位firstCreateInQueue为true
if (this.mappedFiles.isEmpty()) {
mappedFile.setFirstCreateInQueue(true);
}
//将创建的mappedFile加入mappedFiles集合中
this.mappedFiles.add(mappedFile);
}
return mappedFile;
}
2.2.2 putRequestAndReturnMappedFile异步创建MappedFile#
- MappedFile作为一个RocketMQ的物理文件在Java中的映射类。commitLog consumerQueue、indexFile3种文件磁盘的读写都是通过MappedFile操作的。它的构造器中会对当前文件进行mmap内存映射操作。
- putRequestAndReturnMappedFile用于创建MappedFile, 会同时创建两个MappedFile, 一个同步创建并返回用于本次实际使用, 一个后台异步创建用于下次取用。可以避免等到当前文件真正用完了才创建下一个文件, 提升性能。
- 同步和异步实际上都是通过一个服务线程执行的, 该方法只是提交两个映射文件创建请求AllocateRequest, 提交到requestTable和requestQueue中。随后当前线程只会同步等待第一个映射文件的创建, 最多等待5s, 如果创建成功则返回, 较大的offset那一个映射文件则会异步的创建, 不会等待。
- 这里线程等待使用的是倒计数器CountDownLatch, 一个请求一个AllocateRequest对象, 其内部还持有一个CountDownLatch对象, 当请求对应的MappedFile创建之后, 会调用内部的CountDownLatch#countDown, 唤醒等待的线程。
/**
* AllocateMappedFileService的方法
* 添加两个请求到处理任务池，然后阻塞等待异步创建并返回MappedFile
*
* @param nextFilePath
* @param nextNextFilePath
* @param fileSize 文件大小默认1G
* @return
*/
public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
//可以提交的请求
int canSubmitRequests = 2;
//如果当前节点不是从节点，并且是异步刷盘策略，并且transientStorePoolEnable参数配置为true，并且fastFailIfNoBufferInStorePool为true
//那么重新计算最多可以提交几个文件创建请求
if (this.messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
if (this.messageStore.getMessageStoreConfig().isFastFailIfNoBufferInStorePool()
&& BrokerRole.SLAVE != this.messageStore.getMessageStoreConfig().getBrokerRole()) {
//if broker is slave, don't fast fail even no buffer in pool
canSubmitRequests = this.messageStore.getTransientStorePool().availableBufferNums() - this.requestQueue.size();
}
}
//根据nextFilePath创建一个请求对象，并将请求对象存入requestTable这个map集合中
AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
//如果存入成功
if (nextPutOK) {
if (canSubmitRequests <= 0) {
log.warn("[NOTIFYME]TransientStorePool is not enough, so create mapped file error, " +
"RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().availableBufferNums());
this.requestTable.remove(nextFilePath);
return null;
}
//将请求存入requestQueue中
boolean offerOK = this.requestQueue.offer(nextReq);
if (!offerOK) {
log.warn("never expected here, add a request to preallocate queue failed");
}
//可以提交的请求数量自减
canSubmitRequests--;
}
//根据nextNextFilePath创建另一个请求对象，并将请求对象存入requestTable这个map集合中
AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);
boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;
if (nextNextPutOK) {
if (canSubmitRequests <= 0) {
log.warn("[NOTIFYME]TransientStorePool is not enough, so skip preallocate mapped file, " +
"RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().availableBufferNums());
this.requestTable.remove(nextNextFilePath);
} else {
//将请求存入requestQueue中
boolean offerOK = this.requestQueue.offer(nextNextReq);
if (!offerOK) {
log.warn("never expected here, add a request to preallocate queue failed");
}
}
}
//有异常就直接返回
if (hasException) {
log.warn(this.getServiceName() + " service has exception. so return null");
return null;
}
//获取此前存入的nextFilePath对应的请求
AllocateRequest result = this.requestTable.get(nextFilePath);
try {
if (result != null) {
//同步等待最多5s
boolean waitOK = result.getCountDownLatch().await(waitTimeOut, TimeUnit.MILLISECONDS);
if (!waitOK) {
//超时
log.warn("create mmap timeout " + result.getFilePath() + " " + result.getFileSize());
return null;
} else {
//如果nextFilePath对应的MappedFile创建成功，那么从requestTable移除对应的请求
this.requestTable.remove(nextFilePath);
//返回创建的mappedFile
return result.getMappedFile();
}
} else {
log.error("find preallocate mmap failed, this never happen");
}
} catch (InterruptedException e) {
log.warn(this.getServiceName() + " service has exception. ", e);
}
return null;
}
2.2.3 AllocateMappedFileService创建MappedFile#
- AllocateMappedFileService继承了ServiceThread, ServiceThread实现了Runnable接口。
/**
* ServiceThread的方法
* 启动一个线程执行线程任务
*/
public void start() {
log.info("Try to start service thread:{} started:{} lastThread:{}", getServiceName(), started.get(), thread);
//只能启动一次
if (!started.compareAndSet(false, true)) {
return;
}
stopped = false;
//新建线程
this.thread = new Thread(this, getServiceName());
//后台线程
this.thread.setDaemon(isDaemon);
//启动线程
this.thread.start();
}
AllocateMappedFileService#run()\
/**
* AllocateMappedFileService的方法
* 创建mappedFile
*/
public void run() {
log.info(this.getServiceName() + " service started");
//死循环
//如果服务没有停止，并且没有被线程中断，那么一直循环执行mmapOperation方法
while (!this.isStopped() && this.mmapOperation()) {
}
log.info(this.getServiceName() + " service end");
}
2.2.4 mmapOperation执行mmp操作#
创建MappedFile:
1、 从requestQueue中获取优先级最高的一个请求,即文件名最小或者说起始offset最小的请求requestQueue是一个优先级队列；
2、 判断是否需要通过堆外内存创建MappedFile,如果当前节点不是从节点,而且是异步刷盘策略,transientStorePoolEnable参数为true,那么使用堆外内存,默认不使用；
1、 RocketMQ中引入的transientStorePoolEnable能缓解pagecache的压力,原理是基于DirectByteBuffer和MappedByteBuffer的读写分离；
2、 消息先写入DirectByteBuffer(堆外内存),随后从MappedByteBuffer(pagecache)读取；
3、 如果没有启动堆外内存,采用普通方式创建mappedFile,使用mmap操作；
4、 如果mappedFile大小大于等于1G并且warmMapedFileEnable参数为true,那么预热,就是所谓的内存预热或者文件预热注意warmMapedFileEnable参数默认为false,默认不开启预热；
5、 创建成功后,将请求对象中的countDownLatch释放计数,可以唤醒putRequestAndReturnMappedFile方法中阻塞的线程；
/**
* AllocateMappedFileService的方法
* <p>
* mmap 操作，只有被外部线程中断，才会返回false
*/
private boolean mmapOperation() {
boolean isSuccess = false;
AllocateRequest req = null;
try {
//从requestQueue中获取优先级最高的一个请求，即文件名最小或者说起始offset最小的请求
//requestQueue是一个优先级队列
req = this.requestQueue.take();
//从requestTable获取对应的请求
AllocateRequest expectedRequest = this.requestTable.get(req.getFilePath());
if (null == expectedRequest) {
log.warn("this mmap request expired, maybe cause timeout " + req.getFilePath() + " "
+ req.getFileSize());
return true;
}
if (expectedRequest != req) {
log.warn("never expected here, maybe cause timeout " + req.getFilePath() + " "
+ req.getFileSize() + ", req:" + req + ", expectedRequest:" + expectedRequest);
return true;
}
//获取对应的mappedFile，如果为null则创建
if (req.getMappedFile() == null) {
//起始时间
long beginTime = System.currentTimeMillis();
MappedFile mappedFile;
//如果当前节点不是从节点，并且是异步刷盘策略，并且transientStorePoolEnable参数配置为true，那么使用堆外内存，默认不使用
//RocketMQ中引入的 transientStorePoolEnable 能缓解 pagecache 的压力，其原理是基于DirectByteBuffer和MappedByteBuffer的读写分离
//消息先写入DirectByteBuffer（堆外内存），随后从MappedByteBuffer（pageCache）读取。
if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
try {
//可以基于SPI机制获取自定义的MappedFile
mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
//初始化
mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
} catch (RuntimeException e) {
log.warn("Use default implementation.");
mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
}
} else {
//普通方式创建mappedFile，并且进行mmap
mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
}
//创建mappedFile消耗的时间
long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(beginTime);
if (elapsedTime > 10) {
int queueSize = this.requestQueue.size();
log.warn("create mappedFile spent time(ms) " + elapsedTime + " queue size " + queueSize
+ " " + req.getFilePath() + " " + req.getFileSize());
}
// pre write mappedFile
//如果mappedFile大小大于等于1G并且warmMapedFileEnable参数为true，那么预写mappedFile，也就是所谓的内存预热或者文件预热
//注意warmMapedFileEnable参数默认为false，即默认不开启文件预热，因此选哟手动开启
if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig()
.getMappedFileSizeCommitLog()
&&
this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
//预热文件
mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),
this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
}
req.setMappedFile(mappedFile);
this.hasException = false;
isSuccess = true;
}
} catch (InterruptedException e) {
log.warn(this.getServiceName() + " interrupted, possibly by shutdown.");
this.hasException = true;
return false;
} catch (IOException e) {
log.warn(this.getServiceName() + " service has exception. ", e);
this.hasException = true;
if (null != req) {
requestQueue.offer(req);
try {
Thread.sleep(1);
} catch (InterruptedException ignored) {
}
}
} finally {
//如果创建成功，那么将请求对象中的countDownLatch释放计数，这样就可以唤醒在putRequestAndReturnMappedFile方法中被阻塞的线程了
if (req != null && isSuccess)
req.getCountDownLatch().countDown();
}
return true;
}
2.2.4.1 mmap方法#
使用普通构建mappedFile的时候, 会初始化参数, 会把commitlog文件从磁盘空间完全的映射到虚拟内存, 即内存映射, 为mmap操作。
public MappedFile(final String fileName, final int fileSize) throws IOException {
//调用init初始化
init(fileName, fileSize);
}
private void init(final String fileName, final int fileSize) throws IOException {
//文件名。长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0
this.fileName = fileName;
//文件大小。默认1G=1073741824
this.fileSize = fileSize;
//构建file对象
this.file = new File(fileName);
//构建文件起始索引，就是取自文件名
this.fileFromOffset = Long.parseLong(this.file.getName());
boolean ok = false;
//确保文件目录存在
ensureDirOK(this.file.getParent());
try {
//对当前commitlog文件构建文件通道fileChannel
this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
//把commitlog文件完全的映射到虚拟内存，也就是内存映射，即mmap，提升读写性能
this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
//记录数据
TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
TOTAL_MAPPED_FILES.incrementAndGet();
ok = true;
} catch (FileNotFoundException e) {
log.error("Failed to create file " + this.fileName, e);
throw e;
} catch (IOException e) {
log.error("Failed to map file " + this.fileName, e);
throw e;
} finally {
//释放fileChannel，注意释放fileChannel不会对之前的mappedByteBuffer映射产生影响
if (!ok && this.fileChannel != null) {
this.fileChannel.close();
}
}
}
2.2.4.2 采用堆外内存#
判断是否开启了堆外内存, 相对于mmap方法, 会多设置一个writeBuffer。
public void init(final String fileName, final int fileSize,
final TransientStorePool transientStorePool) throws IOException {
//普通初始化
init(fileName, fileSize);
//设置写buffer，采用堆外内存
this.writeBuffer = transientStorePool.borrowBuffer();
this.transientStorePool = transientStorePool;
}
- borrowBuffer方法中会返回TransientStorePool中的 availableBuffer, 如果堆外内存开启, 那么在broker启动创建DefaultMessageStore的时候将会执行TransientStorePool#init方法, 该方法会初始化5个1G大小的堆外内存并且锁定住。是个重量级初始化方法, 会延长broker启动时间。
- 堆外内存就是通过ByteBuffer.allocateDirect方法分配的, 这5块内存可以反复使用。
TransientStorePool#init():
/**
* TransientStorePool的方法
*
* It's a heavy init method.
*/
public void init() {
//默认5个
for (int i = 0; i < poolSize; i++) {
//分配堆外内存，默认大小1G
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);
}
}
如果是普通模式, 那么采用mmap方法, 如果是读写分离模式, 那么broker会将消息写入writeBuffer, 先写入DirectByteBuffer(堆外内存), 直接返回。然后异步服务CommitRealTimeService不断从堆外内存批量Commit到Page Cache中, 消费者始终从mappedByteBuffer(page Cache)读取数据。
高并发下写入 page cache 可能会造成刷脏页时磁盘压力较高, 写入时发生毛刺现象。读写分离可以缓解page Cache压力, 但会影响消息不一致性, 数据一致性会降低。
2.2.5 warmMappedFile文件预热#
- mmap操作对于OS来说只是建立虚拟内存地址至物理地址的映射关系, 即将进程使用的虚拟内存地址映射到物理地址上。并不会加载任何MappedFile数据至内存中, 也并不会分配指定的大小的内存。当程序要访问数据时, 如果发现这部分数据页并没有实际加载到内存中, 则处理器自动触发一个缺页异常, 使得进入内核空间再分配物理内存, 一次默认为4k。一个G大小的commitLog, 如果靠着缺页中断来分配实际内存, 会触发26w多次缺页中断, 会造成很大开销。
- RocketMQ避免频繁发生却也异常的做法是采用文件预热, 提前让os分配物理内存空间, 防止在写入消息时发生缺页异常才进行分配。
/**
* MappedFile的方法
*
* 建立了进程虚拟地址空间映射之后，并没有分配虚拟内存对应的物理内存，这里进行内存预热
*
* @param type 消息刷盘类型，默认 FlushDiskType.ASYNC_FLUSH;
* @param pages 一页大小，默认4k
*/
public void warmMappedFile(FlushDiskType type, int pages) {
long beginTime = System.currentTimeMillis();
// 创建一个新的字节缓冲区
ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
int flush = 0;
long time = System.currentTimeMillis();
//每隔4k大小写入一个0
for (int i = 0, j = 0; i < this.fileSize; i += MappedFile.OS_PAGE_SIZE, j++) {
//每隔4k大小写入一个0
byteBuffer.put(i, (byte) 0);
// force flush when flush disk type is sync
//如果是同步刷盘，则每次写入都要强制刷盘
if (type == FlushDiskType.SYNC_FLUSH) {
if ((i / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE) >= pages) {
flush = i;
mappedByteBuffer.force();
}
}
// prevent gc
//调用Thread.sleep(0)当前线程主动放弃CPU资源，立即进入就绪状态
//防止因为多次循环导致该线程一直抢占着CPU资源不释放，
if (j % 1000 == 0) {
log.info("j={}, costTime={}", j, System.currentTimeMillis() - time);
time = System.currentTimeMillis();
try {
Thread.sleep(0);
} catch (InterruptedException e) {
log.error("Interrupted", e);
}
}
}
// force flush when prepare load finished
//把剩余的数据强制刷新到磁盘中
if (type == FlushDiskType.SYNC_FLUSH) {
log.info("mapped file warm-up done, force to disk, mappedFile={}, costTime={}",
this.getFileName(), System.currentTimeMillis() - beginTime);
mappedByteBuffer.force();
}
log.info("mapped file warm-up done. mappedFile={}, costTime={}", this.getFileName(),
System.currentTimeMillis() - beginTime);
//锁定内存
this.mlock();
}
- RocketMQ对于MappedFile每隔OS_PAGE_SIZE大小写入一个0, 来让操作系统预先分配1G大小的全额物理内存, 预先分配内存。
2.2.6 mlock锁定内存#
- 虽然预热了文件, 短时间不会读取数据不会引发缺页异常, 但是内存不足时, 一部分不常使用的内存还是会被交换到swap空间中, 程序读取交换出的数据时候会产生缺页异常。
- mlock方法调用系统mlock函数, 锁定文件的page cache, 防止把预热的文件交换到swap空间。还会调用系统madvise函数, 尝试一次性将一段数据读入到映射内存区域, 减少了缺页异常。
/**
* MappedFile的方法
* 锁定内存
*/
public void mlock() {
final long beginTime = System.currentTimeMillis();
final long address = ((DirectBuffer) (this.mappedByteBuffer)).address();
Pointer pointer = new Pointer(address);
{
//mlock调用
int ret = LibC.INSTANCE.mlock(pointer, new NativeLong(this.fileSize));
log.info("mlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime);
}
{
//madvise调用
int ret = LibC.INSTANCE.madvise(pointer, new NativeLong(this.fileSize), LibC.MADV_WILLNEED);
log.info("madvise {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime);
}
}
2.3 appendMessage追加存储消息
- 当获取到mappedFile之后, 调用mappedFile#appendMessage方法追加消息。
/**
* MappedFile的方法
* <p>
* 追加消息
*
* @param msg 消息
* @param cb 回调函数
* @param putMessageContext 存放消息上下文
*/
public AppendMessageResult appendMessage(final MessageExtBrokerInner msg, final AppendMessageCallback cb,
PutMessageContext putMessageContext) {
//调用appendMessagesInner方法
return appendMessagesInner(msg, cb, putMessageContext);
}
首先获取当前文件的写指针, 如果写指针小于文件的大小, 进行消息追加, doAppend方法。最后更新写指针位置, 和存储时间。
/**
* MappedFile的方法
* <p>
* 追加消息
*
* @param messageExt 消息
* @param cb 回调函数
* @param putMessageContext 存放消息上下文
*/
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb,
PutMessageContext putMessageContext) {
assert messageExt != null;
assert cb != null;
//获取写入指针的位置
int currentPos = this.wrotePosition.get();
//如果小于文件大小，那么可以写入
if (currentPos < this.fileSize) {
//如果存在writeBuffer，即支持堆外缓存，那么则使用writeBuffer进行读写分离，否则使用mmap的方式写
ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
//设置写入位置
byteBuffer.position(currentPos);
AppendMessageResult result;
/*
* 通过回调函数执行实际写入
*/
if (messageExt instanceof MessageExtBrokerInner) {
//单条消息
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,
(MessageExtBrokerInner) messageExt, putMessageContext);
} else if (messageExt instanceof MessageExtBatch) {
//批量消息
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,
(MessageExtBatch) messageExt, putMessageContext);
} else {
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
//更新写指针的位置
this.wrotePosition.addAndGet(result.getWroteBytes());
//更新存储实时间
this.storeTimestamp = result.getStoreTimestamp();
return result;
}
log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
2.3.1 doAppend执行追加#
- 通过AppendMessageCallback回调函数的doAppend方法执行的。回调函数的具体实现是DefaultAppendMessageCallback, 是CommitLog里面的一个内部类的实现。
1、 获取消息物理偏移量,创建服务端消息Id生成器,4个字节ip+4个字节端口+8个消息偏移量从topicQueueTable中获取Queue队列的最大相对偏移量；
2、 判断如果消息的长度加上文件结束符子节数大于maxBlank,表示commitlog剩余大小不足存储消息,返回END_OF_FILE在asyncPutMessage方法中判断到该code之后将会新建一个MappedFile并尝试再次存储；
3、 如果空间足够,消息进行编码,编码后的消息写入到byteBuffer中,byteBuffer可能是writeBffer,即直接缓冲区,也可能为普通缓冲区mappedByteBuffer；
4、 返回AppendMessageResult对象,内部包含消息追加状态,消息写入物理偏移量,消息写入长度,消息id生成器,、消息开始追加的时间戳、消息队列偏移量、消息开始写入的时间戳等属性；
/**
* DefaultAppendMessageCallback的方法
* <p>
* 追加消息回调
*
* @param fileFromOffset 文件起始索引
* @param byteBuffer 缓冲区
* @param maxBlank 最大空闲区
* @param msgInner 消息
* @param putMessageContext 上下文
*/
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBrokerInner msgInner, PutMessageContext putMessageContext) {
// STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
// PHY OFFSET
//获取物理偏移量索引
long wroteOffset = fileFromOffset + byteBuffer.position();
/*
* 构建msgId,也就是broker端的唯一id,在发送消息的时候,在客户端producer也会生成一个唯一id是的。
*/
Supplier<String> msgIdSupplier = () -> {
//系统标识
int sysflag = msgInner.getSysFlag();
//长度16
int msgIdLen = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
//分配16字节的缓冲区
ByteBuffer msgIdBuffer = ByteBuffer.allocate(msgIdLen);
//ip4个字节、host4个字节
MessageExt.socketAddress2ByteBuffer(msgInner.getStoreHost(), msgIdBuffer);
//清除缓冲区,因为因为socketAddress2ByteBuffer会翻转缓冲区
msgIdBuffer.clear();//because socketAddress2ByteBuffer flip the buffer
//8个字节存储commitLog的物理偏移量
msgIdBuffer.putLong(msgIdLen - 8, wroteOffset);
return UtilAll.bytes2string(msgIdBuffer.array());
};
// Record ConsumeQueue information
//记录ConsumeQueue信息
//key = "topic-queueId"
String key = putMessageContext.getTopicQueueTableKey();
//获取该队列的最大相对偏移量
Long queueOffset = CommitLog.this.topicQueueTable.get(key);
if (null == queueOffset) {
//如果为null则设置为0,并且存入topicQueueTable
queueOffset = 0L;
CommitLog.this.topicQueueTable.put(key, queueOffset);
}
//light message queue(LMQ)支持
boolean multiDispatchWrapResult = CommitLog.this.multiDispatch.wrapMultiDispatch(msgInner);
if (!multiDispatchWrapResult) {
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
// Transaction messages that require special handling
//需要特殊处理的事务消息
final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
switch (tranType) {
// Prepared and Rollback message is not consumed, will not enter the
// consumer queuec
//准备和回滚消息不会被消费，不会进入消费队列
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
queueOffset = 0L;
break;
//非事务消息和提交消息会被消费
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
default:
break;
}
/*
* 消息编码序列化
*/
//获取编码的ByteBuffer
ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
final int msgLen = preEncodeBuffer.getInt(0);
// Determines whether there is sufficient free space
//消息编码
if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
this.msgStoreItemMemory.clear();
// 1 TOTALSIZE
this.msgStoreItemMemory.putInt(maxBlank);
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
// 3 The remaining space may be any value
// Here the length of the specially set maxBlank
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
byteBuffer.put(this.msgStoreItemMemory.array(), 0, 8);
return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset,
maxBlank, /* only wrote 8 bytes, but declare wrote maxBlank for compute write position */
msgIdSupplier, msgInner.getStoreTimestamp(),
queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
}
int pos = 4 + 4 + 4 + 4 + 4;
// 6 QUEUEOFFSET
preEncodeBuffer.putLong(pos, queueOffset);
pos += 8;
// 7 PHYSICALOFFSET
preEncodeBuffer.putLong(pos, fileFromOffset + byteBuffer.position());
int ipLen = (msgInner.getSysFlag() & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
// 8 SYSFLAG, 9 BORNTIMESTAMP, 10 BORNHOST, 11 STORETIMESTAMP
pos += 8 + 4 + 8 + ipLen;
// refresh store time stamp in lock
preEncodeBuffer.putLong(pos, msgInner.getStoreTimestamp());
//存储消息起始时间
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
// Write messages to the queue buffer
/*
* 将消息写入到byteBuffer中，这里的byteBuffer可能是writeBuffer，即直接缓冲区，也有可能是普通缓冲区mappedByteBuffer
*/
byteBuffer.put(preEncodeBuffer);
msgInner.setEncodedBuff(null);
//返回AppendMessageResult，包括消息追加状态、消息写入偏移量、消息写入长度、消息ID生成器、消息开始追加的时间戳、消息队列偏移量、消息开始写入的时间戳
AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgIdSupplier,
msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
switch (tranType) {
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
break;
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
// The next update ConsumeQueue information
CommitLog.this.topicQueueTable.put(key, ++queueOffset);
CommitLog.this.multiDispatch.updateMultiQueueOffset(msgInner);
break;
default:
break;
}
return result;
}
2.3.2 消息序列化#
Broker的commitlog只会存储序列化后的消息。
3.存储高性能设计总结
通过获取mappedFile, RocketMQ对于commitlog的性能采取了很多措施:
1、 commitlog文件预创建和文件预分配；
2、 mmap；
3、 文件预热和内存预热；
4、 内存锁定；
5、 读写分离；