zoukankan      html  css  js  c++  java
  • RocketMQ源码解析之Broker消息存储(消息存储)

    接上篇RocketMQ源码解析之Broker消息存储(消息接收)

    DefaultMessageStore#asyncPutMessage
    /**
     * 进行存储
     *
     * @param msg MessageInstance to store
     * @return
     */
    @Override
    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
        // broker状态校验部分,检测存储器状态:是否关闭、broker是否为slave、是否不可写、操作系统页写入操作是否频繁
        PutMessageStatus checkStoreStatus = this.checkStoreStatus();
        if (checkStoreStatus != PutMessageStatus.PUT_OK) {
            return CompletableFuture.completedFuture(new PutMessageResult(checkStoreStatus, null));
        }
    
        // 消息校验部分,检测msg的长度是否合法:topic长度是否超过127、properties大小是否超过32767
        PutMessageStatus msgCheckStatus = this.checkMessage(msg);
        if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
            return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null));
        }
        // 获取系统时间
        long beginTime = this.getSystemClock().now();
        // 往commitlog中存储消息
        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().incrementAndGet();
            }
        });
    
        return putResultFuture;
    }
    
    • 第10行,broker状态校验,检测存储器状态:是否关闭、broker是否为slave、是否不可写、操作系统页写入操作是否频繁

    • 第15行,消息校验部分,检测msg的长度是否合法:topic长度是否超过127、properties大小是否超过32767

    • 第23行,往commitlog中存储消息,详细解析见CommitLog#asyncPutMessage

    • 第26行,elapsedTime记录消息存储的时间

    CommitLog#asyncPutMessage
    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;
    
        // Statistics 统计存储服务,记录消息推送次数和消息大小
        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
    
        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 延时消息,延迟级别大于0
            // 延迟消息会由ScheduleMessageService的start方法去创建每个延迟级别对应的定时任务
            if (msg.getDelayTimeLevel() > 0) {
                // 如果延时等级大于最大延时等级, 就设置成最大延时等级
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }
                // 设置延时队列,每个延迟消息的主题都被暂时修改为SCHEDULE_TOPIC_XXXX
                topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
                // 并且根据延迟级别,延迟消息更改了新的队列id
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    
                // Backup real topic, queueId 备份真正的topic和队列id
                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()));
                // 设置延迟消息的topic和队列id
                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }
    
        long elapsedTimeInLock = 0;
        MappedFile unlockMappedFile = null;
        // 获取最后一个文件,就是commitlog 目录下的那个文件
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
        // 获取写入锁,根据配置来选择使用ReentrantLock还是CAS,默认使用CAS
        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        try {
            // 开始在锁里的时间
            long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
            // 获取锁的时间戳,在isOSPageCacheBusy()会根据这个变量判断页写操作是否繁忙
            this.beginTimeInLock = beginLockTimestamp;
    
            // Here settings are stored timestamp, in order to ensure an orderly
            // global
            // 重新设置了写入的时间戳,以确保全局有序
            msg.setStoreTimestamp(beginLockTimestamp);
    
            // 判断mappedFile 文件是否为null 或者写满了
            if (null == mappedFile || mappedFile.isFull()) {
                // 从mappedFileQueue中获取最后一个MappedFile
                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_MAPEDFILE_FAILED, null));
            }
    
            // 往MappedFile中追加消息
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            switch (result.getStatus()) {   // 判断写入状态
                case PUT_OK:
                    break;
                case END_OF_FILE:       // 放不下追加的消息,会重新获取一个MappedFile,然后追加
                    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 CompletableFuture.completedFuture(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 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();
        }
    
        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());
    
        // 刷盘 + HA
        // 前面将数据写入到ByteBuffer缓冲区了,后面就是刷盘操作
        CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
        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);
                if (replicaStatus == PutMessageStatus.FLUSH_SLAVE_TIMEOUT) {
                    log.error("do sync transfer other node, wait return, but failed, topic: {} tags: {} client address: {}",
                            msg.getTopic(), msg.getTags(), msg.getBornHostNameString());
                }
            }
            return putMessageResult;
        });
    }
    
    • 第3行和第6行,分别设置存储时间和crc循环冗余校验
    • 第11行,storeStatsService用于记录消息推送次数和消息大小,第119行和第120行会分别记录消息推送次数和消息的大小
    • 第16行,获取事务状态,
    • 第19行~第41行是针对延迟任务
    • 第46行,获取最后一个文件,commitlog目录下的文件
    • 第48行和第104行分别是在MappedFile追加消息的过程中加锁和释放锁,可以根据配置来选择使用ReentrantLock还是CAS,默认使用CAS
    • 第51行,记录开始的时间,,在DefaultMessageStore#isOSPageCacheBusy函数中会利用这个变量来判断OS页写操作是否繁忙
    • 第72行,往MappedFile中追加消息,追加的详细解析见MappedFile#appendMessagesInner
    • 第73行~第98行是针对追加消息的结果分情况处理,如果是END_OF_FILE,说明文件剩余空间不足,放不下需要追加的消息,就会重新获取一个MappedFile进行追加
    • 第124行~第125行,具体刷盘过程和HA,CommitLog#submitFlushRequest  详情见RocketMQ源码解析之Broker消息存储(刷盘机制)CommitLog#submitReplicaRequest(后续介绍HA时再详细分析)
    MappedFile#appendMessagesInner 在MappedFile追加消息
    /**
     * 追加消息
     *
     * @param messageExt
     * @param cb
     * @return
     */
    public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
        assert messageExt != null;
        assert cb != null;
    
        /**
         * 当前MappedFile的写入位置
         */
        int currentPos = this.wrotePosition.get();
        /**
         * 当前写入位置不超过文件最大值,小于1G继续写入,文件还有剩余空间
         */
        if (currentPos < this.fileSize) {
            /**
             * 仅当transientStorePoolEnable 为true,刷盘策略为异步刷盘(FlushDiskType为ASYNC_FLUSH),
             * 并且broker为主节点时,才启用堆外分配内存。此时:writeBuffer不为null
             *
             * Buffer与同步和异步刷盘相关
             * writeBuffer/mappedByteBuffer 的 position始终为0,而limit则始终等于capacity
             * slice创建一个新的buffer, 是根据position和limit来生成byteBuffer
             */
            ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
            byteBuffer.position(currentPos);    // Buffer定位到当前位置,设置写的起始位置
            AppendMessageResult result;
            /**
             * 根据消息类型,是批量消息还是单个消息,进入相应的处理
             * doAppend是具体的消息写入实现,将消息写入内存
             */
            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);
            }
            /**
             * 修改这个MappedFile写入的位置
             */
            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);
    }
    
    • 第15行,获取当前MappedFile的写入位置
    • 第19行,判断当前写入位置是否超过文件的最大值,小于1G,文件还有剩余空间就继续写入
    • 第29行,Buffer定位到当前位置,设置为写的起始位置
    • 第35行—第41行,根据消息的类型是批量消息还是单个消息,进行相应的而处理,以单个消息为例,详细解析见DefaultAppendMessageCallback#doAppend
    • 第45行和第46行分别是更新MappedFile的写入位置以及最后一次存储时间戳
    DefaultAppendMessageCallback#doAppend 将消息写入MappedFile
    /**
     * 消息写入
     *
     * @param fileFromOffset 起始位置:一个commitLog文件(对应一个MappedFile)在整个文件系列中的偏移量
     *                       文件名代表这个偏移量
     * @param byteBuffer     NIO 字节容器
     * @param maxBlank       最大可写字节数
     * @param msgInner       消息内部封装实体
     * @return
     */
    public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
                                        final MessageExtBrokerInner msgInner) {
        // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>
    
        // PHY OFFSET 这是总的偏移量,在整个commitLog中一个总的
        long wroteOffset = fileFromOffset + byteBuffer.position();
    
        int sysflag = msgInner.getSysFlag();
    
        int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
        int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
        // 分别是broker存储的地址和消息的物理地址
        ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
        ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
        // 重置storeHostHolder
        this.resetByteBuffer(storeHostHolder, storeHostLength);
        String msgId;   // 消息ID
        /**
         * 根据broker存储的地址和消息的物理绝对位置创建msgId
         */
        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 记录消息队列的信息
        /**
         * 根据topic -queryId获取该队列的偏移地址(待写入地址),如果没有就新增一个键值对,当前偏移量为0
         */
        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
        /**
         * @4start
         * 对事务消息(PREPARE,ROLLBACK 准备或者回滚类型的消息)需要特殊处理,其偏移量都为0,不进入consumer队列
         */
        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;
        }// @4 end
    
        /**
         * Serialize message
         */
        final byte[] propertiesData =
                msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
    
        final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
    
        // 消息的附加属性长度不超过 32767
        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 消息长度超过允许的最大长度,则返回MESSAGE_SIZE_EXCEEDED,应该是4M,批量也是?
        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
        /**
         * 如果该MappedFile中可剩余空间小于当前消息长度,则返回AppendMessageStatus.END_OF_FILE
         * broker会重新创建一个新的commitLog来存储该消息
         * 从这里可以看出,每个commitLog文件至少剩余8个字节:
         *  (1)高4字节存储当前文件剩余空间,
         *  (2)低4字节存储魔数CommitLog.BLANK_MAGIC_CODE
         */
        if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) { // 没有足够的可用空间的话需要新建个commitLog
            // 新建一个commitLog文件来存储消息
            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中,然后创建AppendMessageResult对象
            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  magic
        this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
        // 3 BODYCRC body的crc
        this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
        // 4 QUEUEID 队列Id
        this.msgStoreItemMemory.putInt(msgInner.getQueueId());
        // 5 FLAG
        this.msgStoreItemMemory.putInt(msgInner.getFlag());
        // 6 QUEUEOFFSET 在队列中的偏移量
        this.msgStoreItemMemory.putLong(queueOffset);
        // 7 PHYSICALOFFSET 记录在整个大的commitLog中的位置
        this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
        // 8 SYSFLAG 记录系统flag
        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);
    
        // 将消息存储到ByteBuffer中,然后创建AppendMessageResult对象
        AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
                msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
    
        switch (tranType) {
            /**
             * 对事务消息(PREPARE,ROLLBACK类型的消息)需要特殊处理,其偏移量都为0,不进入consumer队列
             */
            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 记录更新下队列的一个偏移量,就是加1
                CommitLog.this.topicQueueTable.put(key, ++queueOffset);
                break;
            default:
                break;
        }
        return result;
    }
    
    • 第16行,获取在commitLog中的绝对位置
    • 第20行至第26行,分别是broker存储的地址和消息的物理地址
    • 第31行—第35行,根据broker存储地址和消息的绝对物理位置创建msgId
    • 第41行—第51行,根据topic和queryId获取该队列的偏移地址,创建一个key,在利用该key,从topicQueueTable中获取对应队列的偏移地址(topicQueueTable是一个HashMap变量,保存topic-queueid, 偏移量之间的关系),如果该变量中没有对应的key,就该键值对添加到Map变量中
    • 第58行—第70行,是针对事务消息进行处理
    • 第75行—第177行,是对消息进行序列化,并计算消息存储所需的长度,如果可用空间能容纳当前消息,就封装消息后放入到对应的队列缓存byteBuffer中,并创建AppendMessageResult对象。没有足够的空间就创建一个新的commitLog来存储该消息
    • 第189行,消息提交成功的话,会更新topicQueueTable中对应的偏移量
  • 相关阅读:
    linux 开发GTK需要的初始配置
    Kernel panic not syncing: cannot execute a PAEenabled kernel on PAEless CPU
    在NetBeans IDE 6.9.1上搭建Android SDK环境(WIN和Linux平台)
    Gtk+ Tutorials & Resources
    新手上路学习配置C,C++,GTK等开发环境
    又是忙碌的一天
    Graphics Programming on linux using C用c语言开发图形界面
    如何在Netbeans下配置Android开发环境 a platform target has to be selected
    linux终端中最漂亮的几款字体介绍及安装
    自已接触过的数据访问方式总结
  • 原文地址:https://www.cnblogs.com/fyusac/p/15309258.html
Copyright © 2011-2022 走看看