/**
 * 消息存储实现类
 */
public class DefaultMessageStore implements MessageStore {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    // 消息存储配置属性
    private final MessageStoreConfig messageStoreConfig;
    // CommitLog 消息存储文件
    private final CommitLog commitLog;
    // 消息队列 存储缓存表，按消息topic分组
    private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueue>> consumeQueueTable;
    // 消息队列文件 ConsumeQueue 刷盘线程
    private final FlushConsumeQueueService flushConsumeQueueService;
    // 清除 Commitlog 文件服务
    private final CleanCommitLogService cleanCommitLogService;
    // 清除 ConsumeQueue 文件服务
    private final CleanConsumeQueueService cleanConsumeQueueService;
    // 索引文件实现类
    private final IndexService indexService;
    // MappedFile 分配服务
    private final AllocateMappedFileService allocateMappedFileService;
    // Commitlog 消息分发，根据 Commitlog 文件构建 ConsumeQueue、IndexFile 文件
    private final ReputMessageService reputMessageService;
    // 存储 HA 机制
    private final HAService haService;

    private final ScheduleMessageService scheduleMessageService;

    private final StoreStatsService storeStatsService;
    // 消息内存缓存池
    private final TransientStorePool transientStorePool;

    private final RunningFlags runningFlags = new RunningFlags();
    private final SystemClock systemClock = new SystemClock();

    private final ScheduledExecutorService scheduledExecutorService =
        Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreScheduledThread"));
    private final BrokerStatsManager brokerStatsManager;
    // 消息拉取，长轮询模式消息到达监听器
    private final MessageArrivingListener messageArrivingListener;
    // Broker 配置属性
    private final BrokerConfig brokerConfig;

    private volatile boolean shutdown = true;
    // 文件刷盘监测点
    private StoreCheckpoint storeCheckpoint;

    private AtomicLong printTimes = new AtomicLong(0);
    // Commitlog 文件转发请求
    private final LinkedList<CommitLogDispatcher> dispatcherList;

    private RandomAccessFile lockFile;

    private FileLock lock;

    boolean shutDownNormal = false;
}

/**
 * 将 producer 发来的消息 存在 broker 的 commitlog 中
 * @param msg Message instance to store
 * @return
 */
public PutMessageResult putMessage(MessageExtBrokerInner msg) {...}
/**
 * 将 producer 发来的消息 存在 broker 的 commitlog 中
 * @param messageExtBatch Message batch.
 * @return
 */
public PutMessageResult putMessages(MessageExtBatch messageExtBatch) {...}

/**
 * 将 producer 发来的消息 存在 broker 的 commitlog 中
 * @param msg Message instance to store
 * @return
 */
public PutMessageResult putMessage(MessageExtBrokerInner msg) {
    if (this.shutdown) { // 如果已经shutdown 抛出异常
        log.warn("message store has shutdown, so putMessage is forbidden");
        return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    }

    if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
        // 如果 当前broker是 slave，则抛出异常
        long value = this.printTimes.getAndIncrement();
        if ((value % 50000) == 0) {
            log.warn("message store is slave mode, so putMessage is forbidden ");
        }

        return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    }

    if (!this.runningFlags.isWriteable()) {
        // 如果 当前flag中 有不可写 flag，则返回异常
        long value = this.printTimes.getAndIncrement();
        if ((value % 50000) == 0) {
            log.warn("message store is not writeable, so putMessage is forbidden " 
                     + this.runningFlags.getFlagBits());
        }

        return new PutMessageResult(PutMessageStatus.SERVICE_NOT_AVAILABLE, null);
    } else {
        this.printTimes.set(0);
    }

    if (msg.getTopic().length() > Byte.MAX_VALUE) {
        // 如果msg的topic 太长，则抛出消息异常
        log.warn("putMessage message topic length too long " + msg.getTopic().length());
        return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
    }

    if (msg.getPropertiesString() != null && msg.getPropertiesString().length() > Short.MAX_VALUE) {
        // 如果 消息的 propertiesString 扩展属性字符串太长，则抛出异常
        log.warn("putMessage message properties length too long " 
                 + msg.getPropertiesString().length());
        return new PutMessageResult(PutMessageStatus.PROPERTIES_SIZE_EXCEEDED, null);
    }

    if (this.isOSPageCacheBusy()) {
        // 如果 系统PageCache 繁忙， 则 返回异常 系统繁忙
        return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, null);
    }

    long beginTime = this.getSystemClock().now();
    // 调用 commitlog 实际保存 msg
    PutMessageResult result = this.commitLog.putMessage(msg);

    long eclipseTime = this.getSystemClock().now() - beginTime;
    if (eclipseTime > 500) {
        log.warn("putMessage not in lock eclipse time(ms)={}, bodyLength={}",
                 eclipseTime, msg.getBody().length);
    }
    // 统计存储1次消息，耗时 大概处于哪个区间，方便分析性能
    this.storeStatsService.setPutMessageEntireTimeMax(eclipseTime);

    if (null == result || !result.isOk()) {
        // 如果 失败，统计失败次数
        this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
    }

    return result;
}

public class MappedFileQueue {
	// 批量删除的最大文件数
    private static final int DELETE_FILES_BATCH_MAX = 10;

    // 存储目录
    private final String storePath;
    // 单个文件的存储大小
    private final int mappedFileSize;
    // MappedFile 文件列表
    private final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();
    // 创建 MappedFile 服务类
    private final AllocateMappedFileService allocateMappedFileService;
    // 当前刷盘指针，表示该指针之前的数据已经 全部持久化 到了磁盘
    private long flushedWhere = 0;
    // 当前提交指针，内存中 ByteBuffer 当前的写指针，该值大于等于 flushedWhere
    private long committedWhere = 0;

    private volatile long storeTimestamp = 0;
}

/**
 * 根据时间戳 查找MappedFile，找到第一个lastModifiedTimestamp >= timestamp 的文件
 * @param timestamp 时间戳 long
 * @return 满足条件的MappedFile
 */
public MappedFile getMappedFileByTime(final long timestamp) {
    Object[] mfs = this.copyMappedFiles(0);

    if (null == mfs)
        return null;

    for (int i = 0; i < mfs.length; i++) {
        MappedFile mappedFile = (MappedFile) mfs[i];
        if (mappedFile.getLastModifiedTimestamp() >= timestamp) {
            return mappedFile;
        }
    }

    return (MappedFile) mfs[mfs.length - 1];
}

/**
 * Finds a mapped file by offset.
 * 查找哪个 MappedFile 包含 offset，则返回对应的 MappedFile
 * @param offset Offset.
 * @param returnFirstOnNotFound If the mapped file is not found, then return the first one.
 * @return Mapped file or null (when not found and returnFirstOnNotFound is <code>false</code>).
 */
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
    try {
        MappedFile firstMappedFile = this.getFirstMappedFile();
        MappedFile lastMappedFile = this.getLastMappedFile();
        if (firstMappedFile != null && lastMappedFile != null) {
            if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
                // 如果参数 offset 不在 当前MappedFileQueue 范围内，返回 returnFirstOnNotFound
                LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
                    offset,
                    firstMappedFile.getFileFromOffset(),
                    lastMappedFile.getFileFromOffset() + this.mappedFileSize,
                    this.mappedFileSize,
                    this.mappedFiles.size());
            } else {
                // 如果参数 offset 在 当前MappedFileQueue 范围内
                // 拿到 offset 对应的 index
                int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
                MappedFile targetFile = null;
                try {
                    targetFile = this.mappedFiles.get(index);
                } catch (Exception ignored) {
                }

                if (targetFile != null && offset >= targetFile.getFileFromOffset()
                    && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
                    // 如果 offset 在 index 文件的 偏移量范围内，返回 targetFile
                    return targetFile;
                }

                for (MappedFile tmpMappedFile : this.mappedFiles) {
                    // 遍历 所有 mappedFiles，查找满足条件的 mappedFile。
                    // 一般上一步可以找到，这里属于托底
                    if (offset >= tmpMappedFile.getFileFromOffset()
                        && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
                        return tmpMappedFile;
                    }
                }
            }

            if (returnFirstOnNotFound) {
                return firstMappedFile;
            }
        }
    } catch (Exception e) {
        log.error("findMappedFileByOffset Exception", e);
    }

    return null;
}

/**
 * writeBuffer 中的 内容提交到 fileChannel
 * @param commitLeastPages 至少提交的页数
 * @return 返回 已提交位置 committedPosition
 */
public int commit(final int commitLeastPages) {
    if (writeBuffer == null) {
        // 不启用 TransientStorePool，就不需要内存拷贝。仅需要 committedPosition = wrotePosition 就行了
        //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
        return this.wrotePosition.get();
    }
    if (this.isAbleToCommit(commitLeastPages)) {
        // 当前的脏页 大于commitLeastPages 才能提交
        if (this.hold()) {
            // 当前 MappedFile 可用
            commit0(commitLeastPages);
            this.release();
        } else {
            log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
        }
    }

    // All dirty data has been committed to FileChannel.
    if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
        // 如果启用了 transientStorePool，并且 已提交位置 已经写满整个物理文件，则 将 writeBuffer 归还 transientStorePool
        this.transientStorePool.returnBuffer(writeBuffer);
        this.writeBuffer = null;
    }

    return this.committedPosition.get();
}

/**
 * 提交 writeBuffer 中的 未提交内容 到 fileChannel
 * @param commitLeastPages 至少提交页数
 */
protected void commit0(final int commitLeastPages) {
    int writePos = this.wrotePosition.get();  // 当前写指针位置
    int lastCommittedPosition = this.committedPosition.get(); // 上次 提交指针 位置

    if (writePos - this.committedPosition.get() > 0) {
        try {
            // 声明一套新指针byteBuffer，指向 writeBuffer[position, limit]。内容相同，指针不同
            ByteBuffer byteBuffer = writeBuffer.slice();
            byteBuffer.position(lastCommittedPosition);
            byteBuffer.limit(writePos);
            // 将 writeBuffer[lastCommittedPosition, writePos] 之间的数据 通过fileChannel 写入 物理文件
            this.fileChannel.position(lastCommittedPosition);
            this.fileChannel.write(byteBuffer);
            // committedPosition 执向当前 写指针位置（writePos）
            this.committedPosition.set(writePos);
        } catch (Throwable e) {
            log.error("Error occurred when commit data to FileChannel.", e);
        }
    }
}

/**
 * 刷盘
 * @param flushLeastPages 刷盘的最小页数
 * @return The current flushed position, 已刷盘的位置
 */
public int flush(final int flushLeastPages) {
    if (this.isAbleToFlush(flushLeastPages)) {
        if (this.hold()) {
            // 当前 MappedFile 可用，记录正在使用
            int value = getReadPosition();

            try {
                //We only append data to fileChannel or mappedByteBuffer, never both.
                if (writeBuffer != null || this.fileChannel.position() != 0) {
                    // writeBuffer不为空，flushedPosition 相当于 上一次 commit指针
                    this.fileChannel.force(false);
                } else {
                    // writeBuffer为空, 数据直接进入mappedByteBuffer，wrotePosition代表mappedByteBuffer的指针
                    this.mappedByteBuffer.force();
                }
            } catch (Throwable e) {
                log.error("Error occurred when force data to disk.", e);
            }

            this.flushedPosition.set(value);
            this.release();  // 记录本次 MappedFile 使用完成
        } else {
            log.warn("in flush, hold failed, flush offset = "
                     + this.flushedPosition.get());
            this.flushedPosition.set(getReadPosition());
        }
    }
    return this.getFlushedPosition();
}

/**
 * writeBuffer == null，说明未启用 transientStorePool，没有提交过程，所以 mappedByteBuffer的
 * 写指针wrotePosition 就是当前最大可读地址
 * writeBuffer ！= null，则启用了 transientStorePool，最大可读指针 就是committedPosition，其他部分还没有提交
 * @return The max position which have valid data
 */
public int getReadPosition() {
    return this.writeBuffer == null ? this.wrotePosition.get() : this.committedPosition.get();
}

/**
 * 从 MappedFile 中 选择一块区域
 * @param pos MappedFile 中 相对于 fileFromOffset 的 位置
 * @return 一套指针，指向 MappedFile中 的 指定位置
 */
public SelectMappedBufferResult selectMappedBuffer(int pos) {
    int readPosition = getReadPosition();
    if (pos < readPosition && pos >= 0) {
        if (this.hold()) {
            ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
            byteBuffer.position(pos);
            int size = readPosition - pos;
            ByteBuffer byteBufferNew = byteBuffer.slice();
            byteBufferNew.limit(size);
            return new SelectMappedBufferResult(this.fileFromOffset + pos, 
                                                byteBufferNew, size, this);
        }
    }

    return null;
}

/**
 * MappedFile 销毁
 * @param intervalForcibly 拒绝被销毁的最大存活时间
 * @return boolean 是否销毁成功
 */
public boolean destroy(final long intervalForcibly) {
    // 释放资源
    this.shutdown(intervalForcibly);

    if (this.isCleanupOver()) {
        // 如果已经清理完成
        try {
            this.fileChannel.close(); // 关闭channel
            log.info("close file channel " + this.fileName + " OK");

            long beginTime = System.currentTimeMillis();
            boolean result = this.file.delete();  // 删除文件
            // 记录删除时信息
            log.info("delete file[REF:" + this.getRefCount() + "] " + this.fileName
                + (result ? " OK, " : " Failed, ") + "W:" + this.getWrotePosition() + " M:"
                + this.getFlushedPosition() + ", "
                + UtilAll.computeEclipseTimeMilliseconds(beginTime));
        } catch (Exception e) {
            log.warn("close file channel " + this.fileName + " Failed. ", e);
        }

        return true;
    } else {
        log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
            + " Failed. cleanupOver: " + this.cleanupOver);
    }

    return false;
}

/**
 * 将 对象置为 不可用，引用数降低
 * @param intervalForcibly
 */
public void shutdown(final long intervalForcibly) {
    if (this.available) {
        this.available = false;
        this.firstShutdownTimestamp = System.currentTimeMillis();
        this.release(); // 减少refCount，如果时机合适，调用 cleanup() 释放资源
    } else if (this.getRefCount() > 0) {
        if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {
            this.refCount.set(-1000 - this.getRefCount());
            this.release();
        }
    }
}

/**
 * 释放资源
 * @param currentRef 当前引用数，用于记日志 不做判断
 * @return 是否释放成功
 */
@Override
public boolean cleanup(final long currentRef) {
    if (this.isAvailable()) { // 当前MappedFile是否可用
        log.error("this file[REF:" + currentRef + "] " + this.fileName
            + " have not shutdown, stop unmapping.");
        return false;
    }

    if (this.isCleanupOver()) {  // 是否清理完成
        log.error("this file[REF:" + currentRef + "] " + this.fileName
            + " have cleanup, do not do it again.");
        return true;
    }

    // 释放 mappedByteBuffer
    clean(this.mappedByteBuffer);
    // JVM 虚拟内存数减少
    TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(this.fileSize * (-1));
    // JVM 引用文件数减少
    TOTAL_MAPPED_FILES.decrementAndGet();
    log.info("unmap file[REF:" + currentRef + "] " + this.fileName + " OK");
    return true;
}

/**
 * 通过反射清理 MappedByteBuffer
 * @param buffer MappedByteBuffer
 */
public static void clean(final ByteBuffer buffer) {
    if (buffer == null || !buffer.isDirect() || buffer.capacity() == 0)
        return;
    // 嵌套递归获取directByteBuffer的最内部的attachment或者viewedBuffer方法
    // 获取directByteBuffer的Cleaner对象，然后调用cleaner.clean方法，进行释放资源
    invoke(invoke(viewed(buffer), "cleaner"), "clean");
}

/**
 * 递归获取directByteBuffer的最内部的attachment或者viewedBuffer方法，调用
 * @param buffer MappedByteBuffer
 * @return MappedByteBuffer
 */
private static ByteBuffer viewed(ByteBuffer buffer) {
    String methodName = "viewedBuffer";

    Method[] methods = buffer.getClass().getMethods();
    for (int i = 0; i < methods.length; i++) {
        if (methods[i].getName().equals("attachment")) {
            methodName = "attachment";
            break;
        }
    }
    // 如果 有attachment，则调用attachment；如果没有，则调用 viewedBuffer
    ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
    if (viewedBuffer == null)
        return buffer;
    else
        return viewed(viewedBuffer);
}

/**
 * 通过反射去调用 target 的 方法 methodName
 * @param target 目标对象
 * @param methodName 目标对象的方法
 * @param args 目标对象的方法 的参数
 * @return
 */
private static Object invoke(final Object target, final String methodName, 
                             final Class<?>... args) {
    return AccessController.doPrivileged(new PrivilegedAction<Object>() {
        public Object run() {
            try {
                Method method = method(target, methodName, args);
                method.setAccessible(true);
                return method.invoke(target);
            } catch (Exception e) {
                throw new IllegalStateException(e);
            }
        }
    });
}

/**
 * TransientStorePool 是一个队列，包含多个可用的 buffer
 */
public class TransientStorePool {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    // avaliableBuffers 个数，可通过在 broker中配置文件中设置 transientStorePoolSize大小，默认为5
    // 创建 poolSize 个堆外内存，并利用 com.sum.jna.Library 类库将该批内存锁定，避免被置换到 交换区，提高存储性能
    private final int poolSize;
    // 每个 ByteBuffer 大小，默认为 mapedFileSizeCommitLog，表明TransientStorePool为 commitLog 文件服务
    private final int fileSize;
    // 存放 avaliableBuffer 的 双端队列
    private final Deque<ByteBuffer> availableBuffers;
    // 存储相关的配置
    private final MessageStoreConfig storeConfig;
}

Class.forName("com.mysql.jdbc.Driver");

Applet : 通过网络读取 class文件

protected Class<?> loadClass(String name, boolean resolve)
    throws ClassNotFoundException
{
    synchronized (getClassLoadingLock(name)) {
        // First, check if the class has already been loaded
        Class<?> c = findLoadedClass(name);
        if (c == null) {
            long t0 = System.nanoTime();
            try {
                if (parent != null) {
                    c = parent.loadClass(name, false);
                } else {
                    c = findBootstrapClassOrNull(name);
                }
            } catch (ClassNotFoundException e) {
                // ClassNotFoundException thrown if class not found
                // from the non-null parent class loader
            }

            if (c == null) {
                // If still not found, then invoke findClass in order
                // to find the class.
                long t1 = System.nanoTime();
                c = findClass(name);

                // this is the defining class loader; record the stats
                sun.misc.PerfCounter.getParentDelegationTime().addTime(t1 - t0);
                sun.misc.PerfCounter.getFindClassTime().addElapsedTimeFrom(t1);
                sun.misc.PerfCounter.getFindClasses().increment();
            }
        }
        if (resolve) {
            resolveClass(c);
        }
        return c;
    }
}

Object obj = new Object();
SoftReference softRef = new SoftReference(obj); // 将软引用指向 Object实例
obj = null; // 结束 obj 对 Object实例的强引用
Object anotherRef = (Object) softRef.get(); // 重新将 anotherRef 强引用指向 Object实例

package org.apache.rocketmq.common.message;

public class Message implements Serializable {
    private static final long serialVersionUID = 8445773977080406428L;

    private String topic; // 消息所属主题topic
    private int flag;  // 消息flag
    private Map<String, String> properties; // 扩展属性
    private byte[] body; // 消息体
    private String transactionId;  // 事务ID
}

package org.apache.rocketmq.common.sysflag;

public class MessageSysFlag {
    public final static int COMPRESSED_FLAG = 0x1;
    public final static int MULTI_TAGS_FLAG = 0x1 << 1;
    public final static int TRANSACTION_NOT_TYPE = 0;
    public final static int TRANSACTION_PREPARED_TYPE = 0x1 << 2;
    public final static int TRANSACTION_COMMIT_TYPE = 0x2 << 2;
    public final static int TRANSACTION_ROLLBACK_TYPE = 0x3 << 2;
}

public interface MQProducer extends MQAdmin {}

package org.apache.rocketmq.client.producer;
public class DefaultMQProducer extends ClientConfig implements MQProducer {}

public class DefaultMQProducer extends ClientConfig implements MQProducer {
    private String producerGroup;
    private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC;
    private volatile int defaultTopicQueueNums = 4; // 一个topic默认队列数量
    private int sendMsgTimeout = 3000; // 发送消息超时ms
    private int compressMsgBodyOverHowmuch = 1024 * 4;  // 消息体超过多大时 压缩
    private int retryTimesWhenSendFailed = 2;  // 同步发送失败重试次数
    private int retryTimesWhenSendAsyncFailed = 2;  // 异步发送失败重试次数
    private boolean retryAnotherBrokerWhenNotStoreOK = false;  // 存储失败时尝试其他Broker
    private int maxMessageSize = 1024 * 1024 * 4; // 4M，最大消息大小
}

public void start(final boolean startFactory) throws MQClientException {
    switch (this.serviceState) {
        case CREATE_JUST:
            this.serviceState = ServiceState.START_FAILED;
            // step1：检查 producerGroup 是否合法，并改变生产者的instanceName为pid
            this.checkConfig();
            if (!this.defaultMQProducer.getProducerGroup().equals(
                MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                this.defaultMQProducer.changeInstanceNameToPID();
            }
 
            // step2: 创建 MQClientInstance 实例。
            // 整个JVM中只存在一个 MQClientManager实例，维护一个 MQClientInstance 缓存表 factoryTable。也就是同一个 clientId 只会创建一个 MQClientInstance
            // clientId 格式：客户端IP@instanceName（@unitName）
            this.mQClientFactory = MQClientManager.getInstance()
                .getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);

            // step3：向 MQClientInstance 注册，将当前生产者 加入到 MQClientInstance 的管理中。方便后续调用网络请求、进行心跳检测
            boolean registerOK = mQClientFactory.registerProducer(
                this.defaultMQProducer.getProducerGroup(), this);
            if (!registerOK) {
                this.serviceState = ServiceState.CREATE_JUST;
                throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup() + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL), null);
            }

            this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
            // step4；启动 MQClientInstance
            if (startFactory) {
                mQClientFactory.start();
            }
            log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(), this.defaultMQProducer.isSendMessageWithVIPChannel());
            this.serviceState = ServiceState.RUNNING;
            break;
        case RUNNING:
        case START_FAILED:
        case SHUTDOWN_ALREADY:
            throw new MQClientException("The producer service state not OK, maybe started once, " + this.serviceState + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
        default:
            break;
    }

    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
}

public SendResult send(Message msg) {
    return this.defaultMQProducerImpl.send(msg);
}

public SendResult send(Message msg) {
    return send(msg, this.defaultMQProducer.getSendMsgTimeout());
}
public SendResult send(Message msg, long timeout) {
    return this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
}
/**
 * 发送消息方法 的 默认实现
 * @param msg 消息
 * @param communicationMode 同步、异步、单向
 * @param sendCallback 异步发送的回调接口
 * @param timeout 超时时间
 * @return
 * @throws MQClientException
 * @throws RemotingException
 * @throws MQBrokerException
 * @throws InterruptedException
 */
private SendResult sendDefaultImpl(
    Message msg,
    final CommunicationMode communicationMode,
    final SendCallback sendCallback,
    final long timeout
) {
    ...
    // 查找 topic 对应的 路由信息
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    if (topicPublishInfo != null && topicPublishInfo.ok()) {
        // 重试机制，最多重试 timesTotal 次
        for (; times < timesTotal; times++) {
            String lastBrokerName = null == mq ? null : mq.getBrokerName();
            // 选择一个消息队列
            MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
            if (mqSelected != null) {
                mq = mqSelected;
                brokersSent[times] = mq.getBrokerName();
                try {
                    // 实际发送消息
                    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
                    endTimestamp = System.currentTimeMillis();
                    this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
                    switch (communicationMode) {
                        case ASYNC:
                        case ONEWAY:
                            return null;
                        case SYNC:
                            if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                                if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
                                    continue;
                                }
                            }

                            return sendResult;
                        default:
                            break;
                    }
                } catch (...) {
                    ...
                }
            }  // end-if
        } // end-for
    }
    ...
}

/**
 * 查找 topic 对应的路由信息
 * @param topic 主题
 * @return
 */
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
    TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
    // 如果 producer 的 topicPublishInfoTable 中没有缓存 topic，则向 NameServer 查询topic的路由信息
    if (null == topicPublishInfo || !topicPublishInfo.ok()) {
        this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
        // 从 NameServer 拉取 topic 的路由信息，并更新 topicRouteTable 缓存
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
    }

    // 如果producer中缓存了 topic 路由信息，则直接返回
    if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
        return topicPublishInfo;
    } else {
        this.mQClientFactory.updateTopicRouteInfoFromNameServer(
            topic, true, this.defaultMQProducer);
        topicPublishInfo = this.topicPublishInfoTable.get(topic);
        return topicPublishInfo;
    }
}

public class TopicPublishInfo {
    private boolean orderTopic = false;  // 消息是否顺序消息
    private boolean haveTopicRouterInfo = false;  // 是否包含 topic 路由信息
    // 该 topic 下的消息队列
    private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();
    // 发送到哪个queue，[1, Integer.MAX_VALUE]
    private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
    private TopicRouteData topicRouteData;  // topic 的 路由信息
}

public class TopicRouteData extends RemotingSerializable {
    private String orderTopicConf;  //
    private List<QueueData> queueDatas;  // topic 队列元数据
    private List<BrokerData> brokerDatas;  // topic 分布的 broker 元数据
    // broker 上 过滤服务器地址列表
    private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
}

/**
 * 从路由信息中选择一个消息队列去发送消息
 * @param tpInfo 路由信息
 * @param lastBrokerName 上次发送消息出错的 broker，优先排除
 * @return
 */
public MessageQueue selectOneMessageQueue(
           final TopicPublishInfo tpInfo, final String lastBrokerName) {
    if (this.sendLatencyFaultEnable) {
        // 如果启用了故障延迟机制 (sendLatencyFaultEnable = true)
        // 1、【在健康的Broker中】遍历 messageQueueList，找一个 lastBrokerName 之外的 消息队列
        tpInfo.getMessageQueueList()
        // 2、尝试从 【被 规避的Broker中（被关进小黑屋的Broker）】选择一个可用的Broker。如果没找到，返回 null
        final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
        // 3、如果前两步都没找到，找一个托底的 MessageQueue
        return tpInfo.selectOneMessageQueue();
    }

    // 如果没有启用故障延迟机制(sendLatencyFaultEnable = false)
    return tpInfo.selectOneMessageQueue(lastBrokerName);
}

/**
 * 不启用故障延迟机制，获取消息队列
 * @param lastBrokerName 上一次消息发送失败的 brokerName
 * @return MessageQueue
 */
public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
    if (lastBrokerName == null) {
        return selectOneMessageQueue();
    } else {
        int index = this.sendWhichQueue.getAndIncrement();
        for (int i = 0; i < this.messageQueueList.size(); i++) {
            int pos = Math.abs(index++) % this.messageQueueList.size();
            if (pos < 0)
                pos = 0;
            MessageQueue mq = this.messageQueueList.get(pos);
            // 规避上次 发送失败的 MessageQueue 所在 Broker
            if (!mq.getBrokerName().equals(lastBrokerName)) {
                return mq;
            }
        }
        // 上述机制找不到可用的MessageQueue时，找一个托底的 MessageQueue返回
        return selectOneMessageQueue();
    }
}
public MessageQueue selectOneMessageQueue() {
    int index = this.sendWhichQueue.getAndIncrement();
    int pos = Math.abs(index) % this.messageQueueList.size();
    if (pos < 0)
        pos = 0;
    return this.messageQueueList.get(pos);
}

/**
 * 从路由信息中选择一个消息队列去发送消息
 * @param tpInfo 路由信息
 * @param lastBrokerName 上次发送消息出错的 broker，优先排除
 * @return
 */
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
    if (this.sendLatencyFaultEnable) {
        // 如果启用了故障延迟机制 (sendLatencyFaultEnable = true)
        try {
            // 1、【在健康的Broker中】遍历 messageQueueList，找一个 lastBrokerName 之外的 消息队列
            int index = tpInfo.getSendWhichQueue().getAndIncrement();
            for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                // 根据 入狱时间（startTimestamp） 判断是否小黑屋期满，期满 则 重新可用
                if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                    if (null == lastBrokerName 
                        || !mq.getBrokerName().equals(lastBrokerName)) // bug
                        return mq;
                }
            }

            // 2、尝试从 【被 规避的Broker中（被关进小黑屋的Broker）】选择一个可用的Broker
            final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
            int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
            if (writeQueueNums > 0) {
                final MessageQueue mq = tpInfo.selectOneMessageQueue();
                if (notBestBroker != null) {
                    mq.setBrokerName(notBestBroker);
                    mq.setQueueId(tpInfo.getSendWhichQueue()
                                  .getAndIncrement() % writeQueueNums);
                }
                return mq;
            } else {
                latencyFaultTolerance.remove(notBestBroker);
            }
        } catch (Exception e) {
            log.error("Error occurred when selecting message queue", e);
        }

        // 3、如果前两步都没找到，找一个托底的 MessageQueue
        return tpInfo.selectOneMessageQueue();
    }

    // 如果没有启用故障延迟机制(sendLatencyFaultEnable = false)
    return tpInfo.selectOneMessageQueue(lastBrokerName);
}

/**
 * 消息发送API 核心入口
 * @param msg 待发送消息
 * @param mq MessageQueue【topic，brokerName，queueId】
 * @param communicationMode 发送模式， 同步，异步，Oneway
 * @param sendCallback 异步发送 回调函数
 * @param topicPublishInfo  路由信息
 * @param timeout  消息发送超时时间
 * @return SendResult 发送结果
 * @throws MQClientException
 * @throws RemotingException
 * @throws MQBrokerException
 * @throws InterruptedException
 */
private SendResult sendKernelImpl(
            final Message msg,
            final MessageQueue mq,
            final CommunicationMode communicationMode,
            final SendCallback sendCallback,
            final TopicPublishInfo topicPublishInfo,
            final long timeout)

// 单向发送
this.remotingClient.invokeOneway(addr, request, timeoutMillis);
// 同步发送
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
// 异步发送
this.remotingClient.invokeAsync(addr, request, timeoutMillis, new InvokeCallback(){...});

public static NamesrvController main0(String[] args) {
    try {
        NamesrvController controller = createNamesrvController(args);
        start(controller);
        String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
        log.info(tip);
        System.out.printf("%s%n", tip);
        return controller;
    } catch (Throwable e) {
        e.printStackTrace();
        System.exit(-1);
    }
    return null;
}

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    //PackageConflictDetect.detectFastjson();
    // 将 args 转换成 CommandLine 对象
    Options options = ServerUtil.buildCommandlineOptions(new Options());
    commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
    if (null == commandLine) {
        System.exit(-1);
        return null;
    }
    // 读取 CommandLine 中 -c namesrv.conf 中的配置, 加载 到 NamesrvConfig 和 NettyServerConfig 中
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();
    nettyServerConfig.setListenPort(9876);
    if (commandLine.hasOption('c')) {
        String file = commandLine.getOptionValue('c');
        if (file != null) {
            InputStream in = new BufferedInputStream(new FileInputStream(file));
            properties = new Properties();
            properties.load(in);
            MixAll.properties2Object(properties, namesrvConfig);
            MixAll.properties2Object(properties, nettyServerConfig);

            namesrvConfig.setConfigStorePath(file);

            System.out.printf("load config properties file OK, %s%n", file);
            in.close();
        }
    }
    // 如果有 -p 参数，则 向console打印 namesrvConfig 和 nettyServerConfig 所有属性，退出
    if (commandLine.hasOption('p')) {
        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
        MixAll.printObjectProperties(console, namesrvConfig);
        MixAll.printObjectProperties(console, nettyServerConfig);
        System.exit(0);
    }
    // 将 commandLine 中的其他参数 加载到 namesrvConfig 中
    MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
    // 如果没有 设置 rocketmqHome，则 退出
    if (null == namesrvConfig.getRocketmqHome()) {
        System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
        System.exit(-2);
    }
    // 从 logback_namesrv.xml 中 加载 日志配置信息
    LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
    JoranConfigurator configurator = new JoranConfigurator();
    configurator.setContext(lc);
    lc.reset();
    configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");

    log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    // 向 日志文件.log 打印 namesrvConfig 和 nettyServerConfig
    MixAll.printObjectProperties(log, namesrvConfig);
    MixAll.printObjectProperties(log, nettyServerConfig);
    // 使用 namesrvConfig 和 nettyServerConfig 初始化 NamesrvController
    // 同时初始化 Configuration
    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

    // remember all configs to prevent discard
    // 将 properties 全部记录到 Configuration 中
    controller.getConfiguration().registerConfig(properties);

    return controller;
}

public NamesrvController(
    NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
    this.namesrvConfig = namesrvConfig;
    this.nettyServerConfig = nettyServerConfig;
    this.kvConfigManager = new KVConfigManager(this);
    this.routeInfoManager = new RouteInfoManager();
    this.brokerHousekeepingService = new BrokerHousekeepingService(this);
    this.configuration = new Configuration(
        log,
        this.namesrvConfig, this.nettyServerConfig
    );
    this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
}

org.apache.rocketmq.namesrv.NamesrvController;

public static NamesrvController start(final NamesrvController controller) throws Exception {
    if (null == controller) {
        throw new IllegalArgumentException("NamesrvController is null");
    }
    // 初始化
    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }
    // 注册钩子，JVM 关闭时，关闭controller
    Runtime.getRuntime().addShutdownHook(
        new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));
    // 启动 controller
    controller.start();
    return controller;
}