RocketMQ5.0.0消息存储<二>_消息存储流程
目录
一、消息存储概览
二、Broker接收消息
三、消息存储流程
1. DefaultMessageStore类
2. 存储流程
1):同步与异步存储
2):CommitLog异步存储消息
3):提交消息(Commit)
四、参考资料
一、消息存储概览
如下图所示,是消息从生产者发送消息到消费者消费消息的大致流程。
- step1:生产者发送消息到消息存储Broker端;
- step2:单一文件Commitlog存储所有主题消息,确保顺序写入,提高吞吐量;
- step3:消息通过堆外缓存,Commit消息写入文件内存映射,然后Flush写入磁盘;
- step4:消息Flush磁盘后,把消息转发到ConsumeQueue、IndexFile供消费者消费;
- step5:主题下消费队列内容相同,但是一个消费队列在同一时刻只能被一个消费者消费;
- step6:消费者根据集群/广播模式、PUSH/PULL模式来消费消息。
如何实现顺序存储的呢?通过org.apache.rocketmq.store.PutMessageLock接口,在消息追加文件内存映射时,加锁实现存储消息串行。
消息存储模式:同步、异步。默认异步存储,但是无论同步还是异步,最终执行存储方法是org.apache.rocketmq.store.CommitLog#asyncPutMessage(异步执行,提高存储效率),而同步需要等待存储结果才能返回。
本章主要介绍生产者发送消息,Broker如何接收消息,如何Commit写入文件内存映射,并没有介绍如何刷盘、转发到ConsumeQueue和IndexFile、HA主从同步等内容。
二、Broker接收消息
org.apache.rocketmq.broker.processor.SendMessageProcessor是生产者发送消息后,Broker接收消息的核心实现类。 发送消息请求码是RequestCode.SEND_MESSAGE。发送消息参考《RocketMQ5.0.0消息发送》。
org.apache.rocketmq.broker.processor.SendMessageProcessor#processRequest不仅处理生产者发来的消息,同时还是处理消费端消费ACK的处理请求。其核心逻辑是sendMessage或sendBatchMessage处理方法,都是Broker端存储消息。
以下代码是org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage存储之前对消息的预处理。 核心逻辑如下:
- randomQueueId():发送时是否指定消费队列,若没有指定,则随机选择;
- handleRetryAndDLQ():消息是否延迟或重试消息,并处理;
- sendTransactionPrepareMessage变量:判定是否事务消息,true事务消息;
- 异步存储消息(默认):
事务消息存储:TransactionalMessageServiceImpl#asyncPrepareMessage
普通消息存储:DefaultMessageStore#asyncPutMessage
- 同步存储消息:
事务消息存储:TransactionalMessageServiceImpl#prepareMessage
普通消息存储:DefaultMessageStore#putMessage
/*** 存储之前,对消息的处理* step1:预发送处理,如:检查消息、主题是否符合规范* step2:发送消息时,是否指定消费队列,若没有则随机选择* step3:消息是否进入重试或延迟队列中(重试次数失败)* step4:消息是否是事务消息,若是则存储为prepare消息* step5:BrokerConfig#asyncSendEnable是否开启异步存储,默认开启true* (异步存储、同步存储)*/
public RemotingCommand sendMessage(final ChannelHandlerContext ctx,final RemotingCommand request,final SendMessageContext sendMessageContext,final SendMessageRequestHeader requestHeader,final TopicQueueMappingContext mappingContext,final SendMessageCallback sendMessageCallback) throws RemotingCommandException {// 预发送处理,如:检查消息、主题是否符合规范final RemotingCommand response = preSend(ctx, request, requestHeader);if (response.getCode() != -1) {return response;}final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) response.readCustomHeader();// 获取消息体final byte[] body = request.getBody();// 发送消息时,是否指定消费队列,若没有则随机选择int queueIdInt = requestHeader.getQueueId();// 获取主题配置属性TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());if (queueIdInt < 0) { // 队列ID不符合,则在写队列随机找个queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());}MessageExtBrokerInner msgInner = new MessageExtBrokerInner();msgInner.setTopic(requestHeader.getTopic());msgInner.setQueueId(queueIdInt);// 消息扩展属性Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());// 消息是否进入重试或延迟队列中(重试次数失败)if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig, oriProps)) {return response;}msgInner.setBody(body);msgInner.setFlag(requestHeader.getFlag());String uniqKey = oriProps.get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);if (uniqKey == null || uniqKey.length() <= 0) {uniqKey = MessageClientIDSetter.createUniqID();oriProps.put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, uniqKey);}MessageAccessor.setProperties(msgInner, oriProps);msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(topicConfig.getTopicFilterType(), msgInner.getTags()));msgInner.setBornTimestamp(requestHeader.getBornTimestamp());msgInner.setBornHost(ctx.channel().remoteAddress());msgInner.setStoreHost(this.getStoreHost());msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName);msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));// Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());// 事务标签String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);boolean sendTransactionPrepareMessage = false;if (Boolean.parseBoolean(traFlag)&& !(msgInner.getReconsumeTimes() > 0 && msgInner.getDelayTimeLevel() > 0)) { //For client under version 4.6.1// Broker禁止事务消息存储if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {response.setCode(ResponseCode.NO_PERMISSION);response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()+ "] sending transaction message is forbidden");return response;}sendTransactionPrepareMessage = true;}long beginTimeMillis = this.brokerController.getMessageStore().now();// 消息是否异步存储if (brokerController.getBrokerConfig().isAsyncSendEnable()) {CompletableFuture<PutMessageResult> asyncPutMessageFuture;if (sendTransactionPrepareMessage) { // 事务prepare操作asyncPutMessageFuture = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);} else {asyncPutMessageFuture = this.brokerController.getMessageStore().asyncPutMessage(msgInner);}final int finalQueueIdInt = queueIdInt;final MessageExtBrokerInner finalMsgInner = msgInner;asyncPutMessageFuture.thenAcceptAsync(putMessageResult -> {RemotingCommand responseFuture =handlePutMessageResult(putMessageResult, response, request, finalMsgInner, responseHeader, sendMessageContext,ctx, finalQueueIdInt, beginTimeMillis, mappingContext);if (responseFuture != null) {doResponse(ctx, request, responseFuture);}sendMessageCallback.onComplete(sendMessageContext, response);}, this.brokerController.getPutMessageFutureExecutor());// Returns null to release the send message threadreturn null;}// 消息同步存储else {PutMessageResult putMessageResult = null;// 事务消息存储if (sendTransactionPrepareMessage) {putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);} else {// 同步存储消息putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);}handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt, beginTimeMillis, mappingContext);sendMessageCallback.onComplete(sendMessageContext, response);return response;}
}
三、消息存储流程
1. DefaultMessageStore类
org.apache.rocketmq.store.DefaultMessageStore是消息存储实现类,也是存储模块最重要的一个类,其UML如下。
其关键属性如下代码所示。同步与异步存储的方法:
- 同步消息:单个消息putMessage()、批量消息putMessages()
- 异步消息:单个消息asyncPutMessage()、批量消息asyncPutMessages()
// Commitlog引用次数
public final PerfCounter.Ticks perfs = new PerfCounter.Ticks(LOGGER);// 存储配置属性
private final MessageStoreConfig messageStoreConfig;
// CommitLog(Commitlog文件存储实现类)
private final CommitLog commitLog;
// ConsumeQueue文件存储实现类
private final ConsumeQueueStore consumeQueueStore;
// 刷盘线程
private final FlushConsumeQueueService flushConsumeQueueService;
// 删除过期Commitlog文件服务
private final CleanCommitLogService cleanCommitLogService;
// 删除过期ConsumeQueue文件服务
private final CleanConsumeQueueService cleanConsumeQueueService;
// 矫正逻辑偏移量服务
private final CorrectLogicOffsetService correctLogicOffsetService;
// index文件实现类
private final IndexService indexService;
// MappedFile分配服务
private final AllocateMappedFileService allocateMappedFileService;
// 消息提交到Commitlog时消息转发,构建ConsumeQueue、index文件服务
private ReputMessageService reputMessageService;
// HA服务(主从同步服务)
private HAService haService;
// 存储状态服务
private final StoreStatsService storeStatsService;
// 堆内存缓存
private final TransientStorePool transientStorePool;// Broker状态管理
private final BrokerStatsManager brokerStatsManager;
// 消息达到监听器(消息拉取长轮询模式)
private final MessageArrivingListener messageArrivingListener;
// Broker配置属性
private final BrokerConfig brokerConfig;
// 存储刷盘检查点
private StoreCheckpoint storeCheckpoint;
// 定时消息存储实现类
private TimerMessageStore timerMessageStore;
// 日志打印次数
private AtomicLong printTimes = new AtomicLong(0);
// Commitlog文件转发请求
private final LinkedList<CommitLogDispatcher> dispatcherList;// 延迟消息的延迟级别
private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable =new ConcurrentHashMap<Integer, Long>(32);// 最大延迟级别
private int maxDelayLevel;
2. 存储流程
1):同步与异步存储
如下图所示是同步与异步存储的实现方法调用链,它们之间区别与联系:
联系:a. 同步存储实际是调用异步存储方法,即:DefaultMessageStore#asyncPutMessage;
b. 最终执行存储方法是org.apache.rocketmq.store.CommitLog#asyncPutMessage;
区别:同步存储需要等待存储结果waitForPutResult()
2):CommitLog异步存储消息
org.apache.rocketmq.store.CommitLog#asyncPutMessage是异步执行存储消息。如下代码所示,关键步骤如下:
- step1:mappedFileQueue队列中,获取可写入的Commitlog,即:从commitlog目录下获取当前写入的Commitlog;
- step2:获取当前写入Commitlog的偏移量 = 文件偏移量 + 该文件写入位置;
- step3:是否需要HA(主从Broker同步数据);
- step4:CommitLog.calMsgLength获取该消息的总长度(不定长,总长度存储在前4个字节);
- step5:加锁后(串行写),判定再次Commitlog文件没有或已被写满,则创建新的Commitlog文件;
- step6:Commit操作,即:消息缓存或直接(是否开启堆外内存池)追加到Commitlog文件内存映射buffer,追加当前消息,并没有刷写到磁盘,则返回结果,方法:{@link DefaultMappedFile#appendMessage};
- step7:执行同步或异步刷盘、HA主从同步复制等,方法: {@link CommitLog#handleDiskFlushAndHA}。
/*** 执行存储消息* step1:mappedFileQueue队列中,获取可写入的Commitlog,即:从commitlog目录下获取当前写入的Commitlog;* step2:获取当前写入Commitlog的偏移量 = 文件偏移量 + 该文件写入位置;* step3:是否需要HA(主从Broker同步数据);* step4:CommitLog.calMsgLength获取该消息的总长度(不定长,总长度存储在前4个字节);* step5:加锁后(串行写),判定再次Commitlog文件没有或已被写满,则创建新的Commitlog文件;* step6:Commit操作,即:消息缓存或直接(是否开启堆外内存池)追加到Commitlog文件内存映射buffer,追加当前消息,并没有刷写到磁盘,则返回结果* {@link DefaultMappedFile#appendMessage}* step7:执行同步或异步刷盘、HA主从同步复制等* {@link CommitLog#handleDiskFlushAndHA}* @param msg 消息* @return 存储结果*/
public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {// Set the storage timeif (!defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {msg.setStoreTimestamp(System.currentTimeMillis());}// Set the message body CRC (consider the most appropriate setting on the client)msg.setBodyCRC(UtilAll.crc32(msg.getBody()));// Back to ResultsAppendMessageResult result = null;StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();String topic = msg.getTopic();InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();if (bornSocketAddress.getAddress() instanceof Inet6Address) {msg.setBornHostV6Flag();}InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();if (storeSocketAddress.getAddress() instanceof Inet6Address) {msg.setStoreHostAddressV6Flag();}PutMessageThreadLocal putMessageThreadLocal = this.putMessageThreadLocal.get();updateMaxMessageSize(putMessageThreadLocal);String topicQueueKey = generateKey(putMessageThreadLocal.getKeyBuilder(), msg);long elapsedTimeInLock = 0;// mappedFileQueue队列中,获取可写入的Commitlog,即:从commitlog目录下获取当前写入的CommitlogMappedFile unlockMappedFile = null;MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();// 当前写入Commitlog的偏移量long currOffset;if (mappedFile == null) {currOffset = 0;} else {// 当前写入Commitlog的偏移量 = 文件偏移量 + 该文件写入位置currOffset = mappedFile.getFileFromOffset() + mappedFile.getWrotePosition();}// 是否需要HAint needAckNums = this.defaultMessageStore.getMessageStoreConfig().getInSyncReplicas();boolean needHandleHA = needHandleHA(msg);if (needHandleHA && this.defaultMessageStore.getBrokerConfig().isEnableControllerMode()) {if (this.defaultMessageStore.getHaService().inSyncReplicasNums(currOffset) < this.defaultMessageStore.getMessageStoreConfig().getMinInSyncReplicas()) {return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.IN_SYNC_REPLICAS_NOT_ENOUGH, null));}if (this.defaultMessageStore.getMessageStoreConfig().isAllAckInSyncStateSet()) {// -1 means all ack in SyncStateSetneedAckNums = MixAll.ALL_ACK_IN_SYNC_STATE_SET;}} else if (needHandleHA && this.defaultMessageStore.getBrokerConfig().isEnableSlaveActingMaster()) {int inSyncReplicas = Math.min(this.defaultMessageStore.getAliveReplicaNumInGroup(),this.defaultMessageStore.getHaService().inSyncReplicasNums(currOffset));needAckNums = calcNeedAckNums(inSyncReplicas);if (needAckNums > inSyncReplicas) {// Tell the producer, don't have enough slaves to handle the send requestreturn CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.IN_SYNC_REPLICAS_NOT_ENOUGH, null));}}topicQueueLock.lock(topicQueueKey);try {boolean needAssignOffset = true;if (defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()&& defaultMessageStore.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE) {needAssignOffset = false;}if (needAssignOffset) {defaultMessageStore.assignOffset(msg, getMessageNum(msg));}/*当前消息编码,计算byteBuf(字节缓存)长度, null则正常处理CommitLog.calMsgLength获取该消息的总长度(不定长,总长度存储在前4个字节)*/PutMessageResult encodeResult = putMessageThreadLocal.getEncoder().encode(msg);if (encodeResult != null) {return CompletableFuture.completedFuture(encodeResult);}msg.setEncodedBuff(putMessageThreadLocal.getEncoder().getEncoderBuffer());PutMessageContext putMessageContext = new PutMessageContext(topicQueueKey);putMessageLock.lock(); //spin or ReentrantLock ,depending on store configtry {long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();this.beginTimeInLock = beginLockTimestamp;// Here settings are stored timestamp, in order to ensure an orderly// globalif (!defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {msg.setStoreTimestamp(beginLockTimestamp); // 消息存储时间戳,确保消息存储有序}// Commitlog文件没有或已被写满,则创建新的Commitlog文件if (null == mappedFile || mappedFile.isFull()) {// 创建新的Commitlog文件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 CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPPED_FILE_FAILED, null));}// Commitlog文件写入消息(Commitlog文件内存映射buffer,追加当前消息,并没有刷写到磁盘,则返回结果)result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);switch (result.getStatus()) {case PUT_OK:onCommitLogAppend(msg, result, mappedFile);break;case END_OF_FILE:onCommitLogAppend(msg, result, mappedFile);unlockMappedFile = mappedFile;// Create a new file, re-write the messagemappedFile = this.mappedFileQueue.getLastMappedFile(0);if (null == mappedFile) {// XXX: warn and notify melog.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPPED_FILE_FAILED, result));}result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);if (AppendMessageStatus.PUT_OK.equals(result.getStatus())) {onCommitLogAppend(msg, result, mappedFile);}break;case MESSAGE_SIZE_EXCEEDED:case PROPERTIES_SIZE_EXCEEDED:beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));case UNKNOWN_ERROR:beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));default:beginTimeInLock = 0;return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));}elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;beginTimeInLock = 0;} finally {putMessageLock.unlock();}} finally {topicQueueLock.unlock(topicQueueKey);}// putMessage加锁超时if (elapsedTimeInLock > 500) {log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);}if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {this.defaultMessageStore.unlockMappedFile(unlockMappedFile);}PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);// StatisticsstoreStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).add(result.getMsgNum());storeStatsService.getSinglePutMessageTopicSizeTotal(topic).add(result.getWroteBytes());/*执行同步或异步刷盘、HA主从同步复制等注意:MappedFile.appendMessage只是将消息追加到Commitlog文件内存映射buffer中,并没有刷写到磁盘,则返回结果*/return handleDiskFlushAndHA(putMessageResult, msg, needAckNums, needHandleHA);
}
3):提交消息(Commit)
org.apache.rocketmq.store.logfile.DefaultMappedFile#appendMessage是文件内存映射追加消息方法,目的是把堆外缓存池消息或直接Commit到文件内存映射,其调用链如下。
org.apache.rocketmq.store.logfile.DefaultMappedFile#appendMessagesInner是追加消息到文件内存映射的核心方法,如下代码所示。
/*** Commit操作,即:消息缓存或直接(是否开启堆外内存池)追加到Commitlog文件内存映射buffer,追加当前消息,并没有刷写到磁盘,则返回结果* step1:当前Commitlog的写指针,判断文件是否写满;* step2:slice():创建与原ByteBuffer共享的内存区,拥有独立的position、limit、capacity等指针,并设置position当前写指针* step3:判断是否是批量消息,并追加消息到Commitlog文件内存映射buffer,并没有刷写到磁盘,则返回结果*/
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb,PutMessageContext putMessageContext) {assert messageExt != null;assert cb != null;// 当前Commitlog的写指针int currentPos = WROTE_POSITION_UPDATER.get(this);// 未写满,则追加if (currentPos < this.fileSize) {/*slice():创建与原ByteBuffer共享的内存区,拥有独立的position、limit、capacity等指针并设置position当前写指针*/ByteBuffer byteBuffer = appendMessageBuffer().slice();byteBuffer.position(currentPos);AppendMessageResult result;// 批量消息if (messageExt instanceof MessageExtBatch && !((MessageExtBatch) messageExt).isInnerBatch()) {// traditional batch message// Commitlog文件内存映射buffer,追加当前消息,并没有刷写到磁盘,则返回结果result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,(MessageExtBatch) messageExt, putMessageContext);}// 单个消息else if (messageExt instanceof MessageExtBrokerInner) {// traditional single message or newly introduced inner-batch message// Commitlog文件内存映射buffer,追加当前消息,并没有刷写到磁盘,则返回结果result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos,(MessageExtBrokerInner) messageExt, putMessageContext);} else {return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);}WROTE_POSITION_UPDATER.addAndGet(this, 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);
}
需要注意的是:appendMessageBuffer().slice(),创建与原ByteBuffer共享的内存区,拥有独立的position、limit、capacity等指针,并设置position当前写指针。创建的内存追加消息。
四、参考资料
Rocket Mq消息持久化_飞科-程序人生的博客-CSDN博客
百度安全验证
【RocketMQ】同一个项目中,同一个topic,可以存在多个消费者么? - N!CE波 - 博客园
RocketMQ5.0.0消息发送_爱我所爱0505的博客-CSDN博客
RocketMQ5.0.0路由中心NameServer_爱我所爱0505的博客-CSDN博客
RocketMQ5.0.0消息存储<一>_存储文件及内存映射_爱我所爱0505的博客-CSDN博客