说明

这个是AppendMessageCallback的实现类,在前面讲过，参照refer
用于处理追加消息记录(MessageExtBrokerInner或者MessageExtBatch)到mappedFile中，
返回AppendMessageResult

消息体结构

消息结构，参照refer

注意，这里不完全是头部信息，因为body长度后紧跟着body内容
properties长度后紧跟着properties内容

消息类型

这里分为MessageExtBrokerInner以及MessageExtBatch
分别代表单个消息和批量消息
本文主要针对MessageExtBrokerInner讲解
后面MessageExtBatch结合MessageExtBatchEncoder进行讲解

消息id

消息的id是由下述规则生成的

消息id的生成，参照refer

属性

        // File at the end of the minimum fixed length empty
        private static final int END_FILE_MIN_BLANK_LENGTH = 4 + 4;//最后至少要有8个字节空间在结尾
        private final ByteBuffer msgIdMemory;//消息id的buffer，共16位
        // Store the message content
        private final ByteBuffer msgStoreItemMemory;//存放单个Msg的临时byteBuffer
        // The maximum length of the message
        private final int maxMessageSize;//消息长度最大阈值
        // Build Message Key
        private final StringBuilder keyBuilder = new StringBuilder();//构建topic-queueId

        private final StringBuilder msgIdBuilder = new StringBuilder();//构建msgId，处理批量消息时调用

        private final ByteBuffer hostHolder = ByteBuffer.allocate(8);//存放host 前4字节是ip 如127 0 0 1，后4字节是端口号

函数

构造函数

        /**
         * 都会多分配8bytes作为预留空间
         * 参数size代表允许的最大消息长度,调用方传递size默认4M
         */
        DefaultAppendMessageCallback(final int size) {
            this.msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);//16字节，前8字节是hostHolder，后8字节是wroteOffset(long)
            this.msgStoreItemMemory = ByteBuffer.allocate(size + END_FILE_MIN_BLANK_LENGTH);//多分配8字节
            this.maxMessageSize = size;//最大阈值长度
        }

resetByteBuffer

用于将byteBuffer回到初始位置，再设置limit，使得可以重新写

        private void resetByteBuffer(final ByteBuffer byteBuffer, final int limit) {
            byteBuffer.flip();
            byteBuffer.limit(limit);
        }

doAppend实现父类接口函数

针对MessageExtBrokerInner进行讲解

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();

            this.resetByteBuffer(hostHolder, 8);
            //hostHolder放入ip port如127 0 0 1 2181
            //msgIdMemory 放入8字节ip + 8字节ort + 16位offset
            String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(hostHolder), wroteOffset);

            // Record ConsumeQueue information
            keyBuilder.setLength(0);
            keyBuilder.append(msgInner.getTopic());
            keyBuilder.append('-');
            keyBuilder.append(msgInner.getQueueId());
            String key = keyBuilder.toString();//是topic-queueId形式
            Long queueOffset = CommitLog.this.topicQueueTable.get(key);
            if (null == queueOffset) {
                queueOffset = 0L;
                CommitLog.this.topicQueueTable.put(key, queueOffset);//table没有记录就记录一下，对应offset则为0
            }

            // Transaction messages that require special handling
            final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
            switch (tranType) {//todo 这些类型都有什么意义
                // 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;

            //properties长度超过限制，抛异常
            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(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);//把msgStoreItemMemory写入到byteBuffer中
                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(hostHolder, 8);//使得hostHolder可以写
            this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder));//BORNHOST写入hostHolder之后，让hostHolder可读，写入msgStoreItemMemory
            // 11 STORETIMESTAMP
            this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
            // 12 STOREHOSTADDRESS
            this.resetByteBuffer(hostHolder, 8);
            this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder));
            //this.msgBatchMemory.put(msgInner.getStoreHostBytes());
            // 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) {
                //todo 这些暂时不清楚
                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;
        }

主要步骤简介如下

1.生成msgId 8字节ip + 8字节port + 16位offset
2.生成key，从topicQueueTable(CommitLog的属性)找到或者向topicQueueTable添加key，queueOffset 记录
3.处理事务类型（暂时不清楚）
4.验证propertiesLength 以及 msgLen 是否超过阈值
5.如果剩余长度不够写完，则把消息的TOTALSIZE和MAGICCODE 8字节写进去(END_FILE_MIN_BLANK_LENGTH的意义就在此)
6.剩余长度够的话，就按照消息的结构，把各个属性放入msgStoreItemMemory，即临时的消息体byteBuffer
7.向目标byteBuffer写入msgStoreItemMemory
8.生成AppendMessageResult 准备返回，处理事务类型(暂时不清楚)

思考

END_FILE_MIN_BLANK_LENGTH意义

上面讲了，就是最后空间不够的时候，也要把最后一个消息的TOTALSIZE和MAGICCODE写入进去，就占了8字节

resetByteBuffer作用

flip和limit结合，是为了重新写做准备

问题

目前在那时不清楚MessageSysFlag的四个事务类型

吐槽

各种NIO的处理看着有些麻烦

后续

messageExtBatch的doAppend方法 结合MessageExtBatchEncoder进行讲解

refer

http://www.jianshu.com/p/c76b2654a819 之前关于接口AppendMessageCallback的讲解
http://blog.csdn.net/prestigeding/article/details/76652063 图来源
http://blog.csdn.net/meilong_whpu/article/details/76919267 函数讲解