消息存储主要做的事情:首先将消息放入,然后进行消息追加,进行统计,然后进行刷盘操作,最后进行HA主从同步。此时的消息放入是在CommitLog中会进行转发到ConsumerQueue和IndexFile中。当然在这个过程中,会对消息文件进行人工干预,进行消息的修复和恢复。同时为了防止消息重复消费,会执行ReputMessageService操作。
接着会进入到消息消费端,此时的消息消费者会在消息消费者启动之后,进行消息的核心实现中,进行消息的消费,此时的消费者会进行消息消费,进入到pullMessageProcessor中,此时会根据消息消费记录好偏移量信息,同时在每天的凌晨4点会启动定时任务将消费的消息文件进行删除。
消息存储的代码在store模块中,而store中,我们需要关注的一个类:DefaultMessageStore默认消息存储,它实现了MessageStore消息存储。可以看到接口信息:
/**
* 此类定义要实现的关联接口,从而允许第三方供应商使用自定义消息存储
*/
public interface MessageStore {
/**
* 加载以前存储的消息
*/
boolean load();
/**
* 启动此消息存储
*/
void start() throws Exception;
/**
* 关闭消息存储
*/
void shutdown();
/**
* 销毁消息存储,通常,调用之后所有持久化文件将被移除
*/
void destroy();
/**
* 将消息以异步方式存储到存储器中,处理器可以处理下一个请求,而不是在结果完成后等待结果,以异步方式通知客户端
* @param msg MessageInstance to store
* @return a CompletableFuture for the result of store operation
*/
default CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
return CompletableFuture.completedFuture(putMessage(msg));
}
/**
* 将批量消息以异步的方式存储到存储器中
* @param messageExtBatch the message batch
* @return a CompletableFuture for the result of store operation
*/
default CompletableFuture<PutMessageResult> asyncPutMessages(final MessageExtBatch messageExtBatch) {
return CompletableFuture.completedFuture(putMessages(messageExtBatch));
}
/**
* 将消息存储到存储器中
* @param msg Message instance to store
* @return result of store operation.
*/
PutMessageResult putMessage(final MessageExtBrokerInner msg);
/**
* 将批消息存储到存储器中
*/
PutMessageResult putMessages(final MessageExtBatch messageExtBatch);
/**
* 从给定的偏移量开始最多查询queueId所属主题的maxMsgNumsmessages。 将使用提供的消息过滤器进一步筛选结果消息。
*/
GetMessageResult getMessage(final String group, final String topic, final int queueId,
final long offset, final int maxMsgNums, final MessageFilter messageFilter);
/**
* Get maximum offset of the topic queue.
* 获取主题队列的最大偏移量
*/
long getMaxOffsetInQueue(final String topic, final int queueId);
/**
* Get the minimum offset of the topic queue.
* 获取主题队列的最小偏移量
*/
long getMinOffsetInQueue(final String topic, final int queueId);
/**
* Get the offset of the message in the commit log, which is also known as physical offset.
* 获取commitLog中消息的偏移量,也称为物理偏移量。
*/
long getCommitLogOffsetInQueue(final String topic, final int queueId, final long consumeQueueOffset);
/**
* Look up the physical offset of the message whose store timestamp is as specified.
* 查找其存储时间戳如指定的消息的物理偏移量
*/
long getOffsetInQueueByTime(final String topic, final int queueId, final long timestamp);
/**
* Look up the message by given commit log offset.
* 查找消息通过commitLog的偏移量
*/
MessageExt lookMessageByOffset(final long commitLogOffset);
/**
* Get one message from the specified commit log offset.
* 获取一个消息从特定commitLog的偏移中获取
*/
SelectMappedBufferResult selectOneMessageByOffset(final long commitLogOffset);
/**
* Get one message from the specified commit log offset.
* 获取一个消息从特定commitLog的offset中获取,带消息大小
*/
SelectMappedBufferResult selectOneMessageByOffset(final long commitLogOffset, final int msgSize);
/**
* 获取这个存储器的运行信息
*/
String getRunningDataInfo();
/**
* 消息存储运行信息,包含各种统计信息
*/
HashMap<String, String> getRuntimeInfo();
/**
* Get the maximum commit log offset.
* 获取最大提交commitLog的offset
*/
long getMaxPhyOffset();
/**
* Get the minimum commit log offset.
*获取最小提交commitLog的offset
*/
long getMinPhyOffset();
/**
* Get the store time of the earliest message in the given queue.
* 获取最早消息在给定队列中的存储时间
*/
long getEarliestMessageTime(final String topic, final int queueId);
/**
* Get the store time of the earliest message in this store.
* 获取最早消息在给定队列中的存储时间
* @return timestamp of the earliest message in this store.
*/
long getEarliestMessageTime();
/**
* 获取特定消息的存储时间
*/
long getMessageStoreTimeStamp(final String topic, final int queueId, final long consumeQueueOffset);
/**
* 获取特定队列消息的总数
*/
long getMessageTotalInQueue(final String topic, final int queueId);
/**
* 获取commitLog数据
*/
SelectMappedBufferResult getCommitLogData(final long offset);
/**
* Append data to commit log.
* 追加数据到commitLog中
*/
boolean appendToCommitLog(final long startOffset, final byte[] data);
/**
* Execute file deletion manually.
* 手动执行文件删除。
*/
void executeDeleteFilesManually();
/**
* Query messages by given key.
* 查询消息 通过给定的key
*/
QueryMessageResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
final long end);
/**
* 更新HA master的地址
* @param newAddr new address.
*/
void updateHaMasterAddress(final String newAddr);
/**
* Return how much the slave falls behind.
* 返回slave落后多少
* @return number of bytes that slave falls behind.
*/
long slaveFallBehindMuch();
/**
* Return the current timestamp of the store.
* 返回存储的当前时间戳
* @return current time in milliseconds since 1970-01-01.
*/
long now();
/**
* Clean unused topics.
* 清理无用的topics
* @param topics all valid topics.
* @return number of the topics deleted.
*/
int cleanUnusedTopic(final Set<String> topics);
/**
* Clean expired consume queues.
* 清理过期的消费者队列
*/
void cleanExpiredConsumerQueue();
/**
* Check if the given message has been swapped out of the memory.
* 检查给定的消息是否已被换出内存
*/
boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset);
/**
* 获取已存储在提交日志中但尚未调度到使用队列的字节数。
* @return number of the bytes to dispatch.
*/
long dispatchBehindBytes();
/**
* Flush the message store to persist all data.
* 刷新消息存储去持久化所有数据
* @return maximum offset flushed to persistent storage device.
*/
long flush();
/**
* Reset written offset.
* 重新写入偏移量
*/
boolean resetWriteOffset(long phyOffset);
/**
* Get confirm offset.
* 获取确认的偏移量
* @return confirm offset.
*/
long getConfirmOffset();
/**
* Set confirm offset.
* 设置确认偏移量
* @param phyOffset confirm offset to set.
*/
void setConfirmOffset(long phyOffset);
/**
* Check if the operation system page cache is busy or not.
* 检查操作系统页面缓存是否繁忙
* @return true if the OS page cache is busy; false otherwise.
*/
boolean isOSPageCacheBusy();
/**
* Get lock time in milliseconds of the store by far.
* 获得锁定时间(以毫秒为单位)。
* @return lock time in milliseconds.
*/
long lockTimeMills();
/**
* Check if the transient store pool is deficient.
* 检查临时存储池是否不足
*/
boolean isTransientStorePoolDeficient();
/**
* Get the dispatcher list.
* 获取分发列表
* @return list of the dispatcher.
*/
LinkedList<CommitLogDispatcher> getDispatcherList();
/**
* Get consume queue of the topic/queue.
* 获取消费队列的topic/queue
*/
ConsumeQueue getConsumeQueue(String topic, int queueId);
/**
* Get BrokerStatsManager of the messageStore.
* 获取broker统计管理信息关于消息存储的
* @return BrokerStatsManager.
*/
BrokerStatsManager getBrokerStatsManager();
/**
* handle定时消息服务
* @param brokerRole
*/
void handleScheduleMessageService(BrokerRole brokerRole);
}
/**
* 消息存储配置信息
*/
public class MessageStoreConfig {
//The root directory in which the log data is kept
//存储路径的根目录
@ImportantField
private String storePathRootDir = System.getProperty("user.home") + File.separator + "store";
//The directory in which the commitlog is kept
//commitLog的存储路径
@ImportantField
private String storePathCommitLog = System.getProperty("user.home") + File.separator + "store"
+ File.separator + "commitlog";
// CommitLog file size,default is 1G
//commitLog 文件大小,默认1G
private int mappedFileSizeCommitLog = 1024 * 1024 * 1024;
// ConsumeQueue file size,default is 30W
//消费者队列文件大小,默认30W
private int mappedFileSizeConsumeQueue = 300000 * ConsumeQueue.CQ_STORE_UNIT_SIZE;
// enable consume queue ext
// 是否使用消费者扩展队列 默认false
private boolean enableConsumeQueueExt = false;
// ConsumeQueue extend file size, 48M
//消费者队列 扩展文件大小 48M
private int mappedFileSizeConsumeQueueExt = 48 * 1024 * 1024;
// Bit count of filter bit map.
// this will be set by pipe of calculate filter bit map.
//位图大小 消费者扩展队列 64
private int bitMapLengthConsumeQueueExt = 64;
// CommitLog flush interval
// flush data to disk
//commitLog 刷新间隔 500 刷新数据到disk
@ImportantField
private int flushIntervalCommitLog = 500;
// Only used if TransientStorePool enabled
// flush data to FileChannel
//刷新数据到fileChannel
@ImportantField
private int commitIntervalCommitLog = 200;
/**
* introduced since 4.0.x. Determine whether to use mutex reentrantLock when putting message.<br/>
* By default it is set to false indicating using spin lock when putting message.
* 从4.0.x开始引入
* 确定是否在放置消息时使用互斥的reentrantLock。
* 默认情况下,将其设置为false表示在放置消息时使用自旋锁。
*/
private boolean useReentrantLockWhenPutMessage = false;
// Whether schedule flush,default is real-time
//是否定时刷新,默认是实时的
@ImportantField
private boolean flushCommitLogTimed = false;
//消费队列刷新间隔
// ConsumeQueue flush interval
private int flushIntervalConsumeQueue = 1000;
// Resource reclaim interval
//资源重新声明 间隔 10000
private int cleanResourceInterval = 10000;
//commitLog 移除间隔 100
// CommitLog removal interval
private int deleteCommitLogFilesInterval = 100;
//消费者队列移除 间隔100
// ConsumeQueue removal interval
private int deleteConsumeQueueFilesInterval = 100;
//强制销毁 mapedFile 间隔 1000*120
private int destroyMapedFileIntervalForcibly = 1000 * 120;
//重新删除 挂起文件 间隔 1000*120
private int redeleteHangedFileInterval = 1000 * 120;
//当删除的时 默认时间凌晨4点
// When to delete,default is at 4 am
@ImportantField
private String deleteWhen = "04";
private int diskMaxUsedSpaceRatio = 75;
//保存文件的时间72小时 也即3天 3天后进行删除
// The number of hours to keep a log file before deleting it (in hours)
@ImportantField
private int fileReservedTime = 72;
// Flow control for ConsumeQueue
// 放置消息索引最大高水位 600000
private int putMsgIndexHightWater = 600000;
//默认最大的消息 4M
// The maximum size of message,default is 4M
private int maxMessageSize = 1024 * 1024 * 4;
// Whether check the CRC32 of the records consumed.
// This ensures no on-the-wire or on-disk corruption to the messages occurred.
// This check adds some overhead,so it may be disabled in cases seeking extreme performance.
//是否检查消耗的记录的CRC32
//这样可确保不会对消息进行任何在线或磁盘损坏。
//此检查会增加一些开销,因此在寻求极端性能的情况下可能会被禁用。
private boolean checkCRCOnRecover = true;
//刷新CommitLog时要刷新多少页 4k
// How many pages are to be flushed when flush CommitLog
private int flushCommitLogLeastPages = 4;
// How many pages are to be committed when commit data to file
//将数据提交到文件时要提交多少页 4页
private int commitCommitLogLeastPages = 4;
// Flush page size when the disk in warming state
//磁盘处于预热状态时刷新页面大小 1024/4*16
private int flushLeastPagesWhenWarmMapedFile = 1024 / 4 * 16;
// How many pages are to be flushed when flush ConsumeQueue
//刷新ConsumeQueue时要刷新多少页 2页
private int flushConsumeQueueLeastPages = 2;
private int flushCommitLogThoroughInterval = 1000 * 10;
private int commitCommitLogThoroughInterval = 200;
private int flushConsumeQueueThoroughInterval = 1000 * 60;
//消息在内存中的最大传输字节
@ImportantField
private int maxTransferBytesOnMessageInMemory = 1024 * 256;
//内存中消息的最大传输计数
@ImportantField
private int maxTransferCountOnMessageInMemory = 32;
//磁盘中消息的最大传输字节数
@ImportantField
private int maxTransferBytesOnMessageInDisk = 1024 * 64;
//磁盘中消息的最大传输计数
@ImportantField
private int maxTransferCountOnMessageInDisk = 8;
@ImportantField
private int accessMessageInMemoryMaxRatio = 40;
@ImportantField
private boolean messageIndexEnable = true;
private int maxHashSlotNum = 5000000;
private int maxIndexNum = 5000000 * 4;
private int maxMsgsNumBatch = 64;
@ImportantField
private boolean messageIndexSafe = false;
private int haListenPort = 10912;
private int haSendHeartbeatInterval = 1000 * 5;
private int haHousekeepingInterval = 1000 * 20;
private int haTransferBatchSize = 1024 * 32;
@ImportantField
private String haMasterAddress = null;
private int haSlaveFallbehindMax = 1024 * 1024 * 256;
//broker的角色 异步master
@ImportantField
private BrokerRole brokerRole = BrokerRole.ASYNC_MASTER;
//刷盘的方式 异步
@ImportantField
private FlushDiskType flushDiskType = FlushDiskType.ASYNC_FLUSH;
//异步刷新的超时时间 1000*5
private int syncFlushTimeout = 1000 * 5;
//消息延迟的等级 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
private long flushDelayOffsetInterval = 1000 * 10;
//强制清理文件 true
@ImportantField
private boolean cleanFileForciblyEnable = true;
private boolean warmMapedFileEnable = false;
private boolean offsetCheckInSlave = false;
private boolean debugLockEnable = false;
private boolean duplicationEnable = false;
private boolean diskFallRecorded = true;
private long osPageCacheBusyTimeOutMills = 1000;
private int defaultQueryMaxNum = 32;
//临时存储池启用 也即堆外内存存储 默认不开启
@ImportantField
private boolean transientStorePoolEnable = false;
private int transientStorePoolSize = 5;
private boolean fastFailIfNoBufferInStorePool = false;
private boolean enableDLegerCommitLog = false;
private String dLegerGroup;
private String dLegerPeers;
private String dLegerSelfId;
可以看到存储相关的配置信息,但是这样配置信息似乎是不完全和配置信息匹配的的,比如消息主题的长度是不得超过127,消息属性长度不能超过32767,因为外置了配置信息或者在代码中可以看到写好了判断的条件。
前面虽然说了发送消息,但消息中存在的属性信息有哪些呢?
/**
* 消息
*/
public class Message implements Serializable {
//主题
private String topic;
//标识
private int flag;
//配置
private Map<String, String> properties;
//消息体
private byte[] body;
//事务id
private String transactionId;
}
/**
* 消息的扩展
*/
public class MessageExt extends Message {
//broker名称
private String brokerName;
//队列id
private int queueId;
//存储大小
private int storeSize;
//队列偏移量
private long queueOffset;
//系统标识
private int sysFlag;
//产生的时间戳
private long bornTimestamp;
//产生的host
private SocketAddress bornHost;
//存储时间戳
private long storeTimestamp;
//存储host
private SocketAddress storeHost;
//消息id
private String msgId;
//提交日志偏移量
private long commitLogOffset;
private int bodyCRC;
//重新消费事件
private int reconsumeTimes;
//准备事务偏移量
private long preparedTransactionOffset;
}
public class MessageExtBrokerInner extends MessageExt {
//配置字符、tags编码
private String propertiesString;
private long tagsCode;
}
也即这些都是消息的属性信息。
//放入消息
@Override
public PutMessageResult putMessage(MessageExtBrokerInner msg) {
//检查存储状态:是否关闭、是否可写、是否是salve、是否存储OSPageCacheBusy
//检查完之后,返回ok状态码
PutMessageStatus checkStoreStatus = this.checkStoreStatus();
//判断状态码
if (checkStoreStatus != PutMessageStatus.PUT_OK) {
return new PutMessageResult(checkStoreStatus, null);
}
//检查消息状态:配置字符长度、消息长度,检查完之后,返回ok的状态码
PutMessageStatus msgCheckStatus = this.checkMessage(msg);
if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
return new PutMessageResult(msgCheckStatus, null);
}
//获取系统时间
long beginTime = this.getSystemClock().now();
//在提交日志中放入消息
PutMessageResult result = this.commitLog.putMessage(msg);
//统计时间
long elapsedTime = this.getSystemClock().now() - beginTime;
if (elapsedTime > 500) {
log.warn("not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
}
this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);
if (null == result || !result.isOk()) {
this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
}
return result;
}
//存放message
public PutMessageResult putMessage(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)
//设置消息体,使用crc32
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
// Back to Results
AppendMessageResult result = null;
//获取消息存储统计服务
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
//获取topic、queueId、事务值
String topic = msg.getTopic();
int queueId = msg.getQueueId();
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
//事务类型
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 = ScheduleMessageService.SCHEDULE_TOPIC;
queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// Backup real topic, queueId
// backup真实的topic、queueId
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
msg.setTopic(topic);
msg.setQueueId(queueId);
}
}
//获取产生的host
InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
if (bornSocketAddress.getAddress() instanceof Inet6Address) {
msg.setBornHostV6Flag();
}
//获取存储host
InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
if (storeSocketAddress.getAddress() instanceof Inet6Address) {
msg.setStoreHostAddressV6Flag();
}
long elapsedTimeInLock = 0;
MappedFile unlockMappedFile = null;
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
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);
}
elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
beginTimeInLock = 0;
} finally {
putMessageLock.unlock();
}
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);
// Statistics
//统计信息
storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
//解决硬盘刷新操作
handleDiskFlush(result, putMessageResult, msg);
//进行HA同步
handleHA(result, putMessageResult, msg);
return putMessageResult;
}
注意:在4.7.0中使用的是asyncPutMessage.
//追加消息
public AppendMessageResult appendMessage(final MessageExtBrokerInner msg, final AppendMessageCallback cb) {
//追加消息 内部
return appendMessagesInner(msg, cb);
}
//追加消息
public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
assert messageExt != null;
assert cb != null;
//获取当前的写位置 可以理解成写指针
int currentPos = this.wrotePosition.get();
//如果当前的写位置<文件大小,则进行只加操作
if (currentPos < this.fileSize) {
//slice操作:通过slice创建的新缓冲区只能操作原始缓冲区中数组剩余的数据,
// 即索引为调用slice方法时原始缓冲区的position到limit索引之间的数据,
// 超出这个范围的数据通过slice创建的新缓冲区无法操作到
//通过slice操作拿到剩余的元素空间,将当前位置设置为byteBuffer的当前位置,对消息扩展进行判断,如果属于消息扩展brokerInner
//则进行追加操作,同时分为单个、批量的,否则为错误消息直接返回
//操作完,进行位置的添加操作,同时拿到操作时间,返回
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);
} else if (messageExt instanceof MessageExtBatch) {
//执行消息批量追加操作
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
} 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);
}
//返回追加结果
public AppendMessageResult(AppendMessageStatus status) {
this(status, 0, 0, "", 0, 0, 0);
}
//追加消息结果信息 构造函数
public AppendMessageResult(AppendMessageStatus status, long wroteOffset, int wroteBytes, String msgId,
long storeTimestamp, long logicsOffset, long pagecacheRT) {
this.status = status;
this.wroteOffset = wroteOffset;
this.wroteBytes = wroteBytes;
this.msgId = msgId;
this.storeTimestamp = storeTimestamp;
this.logicsOffset = logicsOffset;
this.pagecacheRT = pagecacheRT;
}
//进行append操作
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBrokerInner msgInner) {
// STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
// PHY OFFSET
//物理偏移量 写入偏移量
long wroteOffset = fileFromOffset + byteBuffer.position();
//拿到标识信息
int sysflag = msgInner.getSysFlag();
//产生host的长度、存储host长度、申请容量bornHostHolder、申请容量storeHostHolder
int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
//重置byteBuffer
this.resetByteBuffer(storeHostHolder, storeHostLength);
String msgId;
//创建消息id
if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
} else {
msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
}
// Record ConsumeQueue information
//记录消息队列信息
keyBuilder.setLength(0);
keyBuilder.append(msgInner.getTopic());
keyBuilder.append('-');
keyBuilder.append(msgInner.getQueueId());
String key = keyBuilder.toString();
//队列偏移量
Long queueOffset = CommitLog.this.topicQueueTable.get(key);
if (null == queueOffset) {
queueOffset = 0L;
CommitLog.this.topicQueueTable.put(key, queueOffset);
}
// 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;
}
/**
* Serialize message
* 序列化消息,进行编码
*/
final byte[] propertiesData =
msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
//配置长度,也即消息主题的长度不能大于127
if (propertiesLength > Short.MAX_VALUE) {
log.warn("putMessage message properties length too long. length={}", propertiesData.length);
return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
}
//主题数据
final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
final int topicLength = topicData.length;
final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
//消息长度
final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
// Exceeds the maximum message
if (msgLen > this.maxMessageSize) {
CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
+ ", maxMessageSize: " + this.maxMessageSize);
return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
}
// Determines whether there is sufficient free space
//是否有充足的容量
if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
// 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, maxBlank);
return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
}
// Initialization of storage space
//初始化存储空间,放入消息的相关属性信息、配置
this.resetByteBuffer(msgStoreItemMemory, msgLen);
// 1 TOTALSIZE
this.msgStoreItemMemory.putInt(msgLen);
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
// 3 BODYCRC
this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
// 4 QUEUEID
this.msgStoreItemMemory.putInt(msgInner.getQueueId());
// 5 FLAG
this.msgStoreItemMemory.putInt(msgInner.getFlag());
// 6 QUEUEOFFSET
this.msgStoreItemMemory.putLong(queueOffset);
// 7 PHYSICALOFFSET
this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
// 8 SYSFLAG
this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
// 9 BORNTIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
// 10 BORNHOST
this.resetByteBuffer(bornHostHolder, bornHostLength);
this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
// 11 STORETIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
// 12 STOREHOSTADDRESS
this.resetByteBuffer(storeHostHolder, storeHostLength);
this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
// 13 RECONSUMETIMES
this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
// 14 Prepared Transaction Offset
this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
// 15 BODY
this.msgStoreItemMemory.putInt(bodyLength);
if (bodyLength > 0)
this.msgStoreItemMemory.put(msgInner.getBody());
// 16 TOPIC
this.msgStoreItemMemory.put((byte) topicLength);
this.msgStoreItemMemory.put(topicData);
// 17 PROPERTIES
this.msgStoreItemMemory.putShort((short) propertiesLength);
if (propertiesLength > 0)
this.msgStoreItemMemory.put(propertiesData);
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
// Write messages to the queue buffer
byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);
//追加消息结果
AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
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);
break;
default:
break;
}
return result;
}
//进行追加操作
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
final MessageExtBatch messageExtBatch) {
//对byteBuffer进行标记,获取物理偏移量,记录消费队列信息(长度、消息主题、消息的队列id)
byteBuffer.mark();
//physical offset
long wroteOffset = fileFromOffset + byteBuffer.position();
// Record ConsumeQueue information
keyBuilder.setLength(0);
keyBuilder.append(messageExtBatch.getTopic());
keyBuilder.append('-');
keyBuilder.append(messageExtBatch.getQueueId());
String key = keyBuilder.toString();
//拿到队列的偏移量
Long queueOffset = CommitLog.this.topicQueueTable.get(key);
if (null == queueOffset) {
queueOffset = 0L;
CommitLog.this.topicQueueTable.put(key, queueOffset);
}
//记录开始的偏移量
long beginQueueOffset = queueOffset;
int totalMsgLen = 0;
int msgNum = 0;
msgIdBuilder.setLength(0);
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
ByteBuffer messagesByteBuff = messageExtBatch.getEncodedBuff();
int sysFlag = messageExtBatch.getSysFlag();
//存储host长度
int storeHostLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
this.resetByteBuffer(storeHostHolder, storeHostLength);
ByteBuffer storeHostBytes = messageExtBatch.getStoreHostBytes(storeHostHolder);
//对消息byteBuff进行标记
messagesByteBuff.mark();
//hasRemaining判断在当前位置postion和限制limit之间是否有元素
//也即postion<limit是否为true
while (messagesByteBuff.hasRemaining()) {
// 1 TOTALSIZE
//拿到当前的位置,当前的长度、消息体的长度
final int msgPos = messagesByteBuff.position();
final int msgLen = messagesByteBuff.getInt();
final int bodyLen = msgLen - 40; //only for log, just estimate it
// Exceeds the maximum message
//如果消息长度大于消息最大长度,进行提示并返回
if (msgLen > this.maxMessageSize) {
CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLen
+ ", maxMessageSize: " + this.maxMessageSize);
return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
}
//否者叠加长度
totalMsgLen += msgLen;
//查看是否有充足的空间
// Determines whether there is sufficient free space
if ((totalMsgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
//重置byteBuffer、放入最大长度、将commitLog设置为空白魔法code
//进行reset操作,此时会恢复到mark的位置
//标记是一个索引,通过Buffer中的mark()方法指定Buffer中一个特定的position,
//之后可以通过调用reset()方法恢复到这个position。
this.resetByteBuffer(this.msgStoreItemMemory, 8);
// 1 TOTALSIZE
this.msgStoreItemMemory.putInt(maxBlank);
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
// 3 The remaining space may be any value
//ignore previous read
messagesByteBuff.reset();
// Here the length of the specially set maxBlank
byteBuffer.reset(); //ignore the previous appended messages
byteBuffer.put(this.msgStoreItemMemory.array(), 0, 8);
return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgIdBuilder.toString(), messageExtBatch.getStoreTimestamp(),
beginQueueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
}
//move to add queue offset and commitlog offset
//移动以添加队列偏移量和commitlog偏移量
messagesByteBuff.position(msgPos + 20);
messagesByteBuff.putLong(queueOffset);
messagesByteBuff.putLong(wroteOffset + totalMsgLen - msgLen);
//重绕缓存区,此时会将原来的位置进行标记清除,位置postion值归为0,同时limit不变
//常用于重新读取缓冲区中数据时使用
storeHostBytes.rewind();
String msgId;
if ((sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
msgId = MessageDecoder.createMessageId(this.msgIdMemory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
} else {
msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
}
//进行append操作
if (msgIdBuilder.length() > 0) {
msgIdBuilder.append(',').append(msgId);
} else {
msgIdBuilder.append(msgId);
}
//对队列偏移量、msgNum、当前位置进行叠加
queueOffset++;
msgNum++;
messagesByteBuff.position(msgPos + msgLen);
}
messagesByteBuff.position(0);
messagesByteBuff.limit(totalMsgLen);
byteBuffer.put(messagesByteBuff);
messageExtBatch.setEncodedBuff(null);
AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, totalMsgLen, msgIdBuilder.toString(),
messageExtBatch.getStoreTimestamp(), beginQueueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
result.setMsgNum(msgNum);
CommitLog.this.topicQueueTable.put(key, queueOffset);
return result;
}
private void resetByteBuffer(final ByteBuffer byteBuffer, final int limit) {
byteBuffer.flip();
byteBuffer.limit(limit);
}
}
接着还会执行统计、刷盘、HA同步操作
// Statistics
//统计信息
storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
//解决硬盘刷新操作
handleDiskFlush(result, putMessageResult, msg);
//进行HA同步
handleHA(result, putMessageResult, msg);
//执行刷盘操作:刷盘操作分为两种:同步刷盘和异步刷盘
public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
// Synchronization flush
//同步刷新操作
if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
//如果处于等待存储消息的状态,则进行存储
if (messageExt.isWaitStoreMsgOK()) {
//拿到消息的偏移量和消息
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
//放入刷盘消息的请求
service.putRequest(request);
CompletableFuture<PutMessageStatus> flushOkFuture = request.future();
PutMessageStatus flushStatus = null;
try {
//获取刷盘状态
flushStatus = flushOkFuture.get(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),
TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
//flushOK=false;
}
//如果刷盘状态ok,则进行wakeup操作,如果不ok,则返回超时状态
if (flushStatus != PutMessageStatus.PUT_OK) {
log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags()
+ " client address: " + messageExt.getBornHostString());
putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
}
} else {
service.wakeup();
}
}
// Asynchronous flush
//否者进行异步刷盘
else {
if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
flushCommitLogService.wakeup();
} else {
commitLogService.wakeup();
}
}
}
//执行HA主从同步操作
public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
//broker的角色,是否是master,如果是则执行HA操作
if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
//拿到HA服务
HAService service = this.defaultMessageStore.getHaService();
//查看是否是等待存储消息状态,如果是则查看slave是否处于等待状态
if (messageExt.isWaitStoreMsgOK()) {
// Determine whether to wait
if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
service.getWaitNotifyObject().wakeupAll();
PutMessageStatus replicaStatus = null;
try {
replicaStatus = request.future().get(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),
TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
}
if (replicaStatus != PutMessageStatus.PUT_OK) {
log.error("do sync transfer other node, wait return, but failed, topic: " + messageExt.getTopic() + " tags: "
+ messageExt.getTags() + " client address: " + messageExt.getBornHostNameString());
putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
}
}
// Slave problem
//否者告诉生产者,slave不可用
else {
// Tell the producer, slave not available
putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
}
}
}
}
消息存储的入口是DefaultMessageStore。同时可以看到其实现了MessageStore。同时通过MessageStoreConfig可以看到配置信息。同时可以看到putMessage操作作为消息存储的入口,putMessage->appendMessage->appendMessagesInner->doAppend,进行统计操作Statistics(getSinglePutMessageTopicTimesTotal、getSinglePutMessageTopicSizeTotal),进行刷盘操作handleDiskFlush,handleHA进行主从同步