package com.alibaba.rocketmq.broker;

public class BrokerController {
	...

	public boolean initizlize() {
		...
		//创建消息持久化服务
		this.messageStore = new DefaultMessageStore(this.meessageStoreConfig, ...);
		//创建了Netty服务端监听消息请求
		this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, ...);
		// 注册事件处理程序
		this.registerProcessor();
	}

	public void start() {
		...
		//开启消息持久化服务
		this.messageStore.start();
		//开启消息请求监听
		this.remotingServer.start();

		this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				// 服务注册 定时向Name Server注册所有的Broker
				BrokerController.this.registerBrokerAll(true, false);
			}
		});
	}
}

	/**
	 * ConsumerManageProcessor
	 */
	ConsumerManageProcessor consumerManageProcessor = new ConsumerManageProcessor(this);
	// Consumer doBalance的时候会发送该请求
	this.remotingServer.registerProcessor(RequestCode.GET_CONSUMER_LIST_BY_GROUP, consumerManageProcessor, this.consumerManageExecutor);

package com.alibaba.rocketmq.broker.processor;

public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {
	...

	@Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {

		switch (request.getCode()) {
		case RequestCode.CONSUMER_SEND_MSG_BACK:
            return this.consumerSendMsgBack(ctx, request);
        default:
        	SendMessageRequestHeader requestHeader = parseRequestHeader(request);
        	if (requestHeader == null) {
                return null;
            }
            mqtraceContext = buildMsgContext(ctx, requestHeader);
            this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
            //调用sendMessage处理消息
            final RemotingCommand response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
            this.executeSendMessageHookAfter(response, mqtraceContext);
            return response;
		}
	}

	private RemotingCommand sendMessage(final ChannelHandlerContext ctx,
            final RemotingCommand request,
            final SendMessageContext mqtraceContext,
            final SendMessageRequestHeader requestHeader) throws RemotingCommandException {
        ...
        //将消息放入DefaultMessageStore，持久化消息
        PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);

        ...
    }
}

package com.alibaba.rocketmq.store;

public class DefaultMessageStore implements MessageStore {
	...

	public PutMessageResult putMessage(MessageExtBrokerInner msg) {
		...

		//将消息持久化到CommitLOg
		PutMessageResult result = this.commitLog.putMessage(msg);

		...
	}
}

package com.alibaba.rocketmq.store;

public class CommitLog {
	...

	public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
		...
		//将消息写入Commit Log
		MapedFile mapedFile = this.mapedFileQueue.getLastMapedFileWithLock();
		...
		//并返回写入结果，包括消息在CommitLog中的偏移量位置offset、消息总字节数bytes、消息MessageId
		AppendMessageResult result = mapedFile.appendMessage(msg, this.appendMessageCallback);
		...

		GroupCommitRequest request = null;

		// Synchronization flush 默认是异步刷盘，将消息从内存中写入文件
        if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {

        	GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
        	request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
        	service.putRequest(request);

        	boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());

        	...
        }
	}
}

package com.alibaba.rocketmq.broker;

public class BrokerController {
	...

	public boolean initialize() {
		boolean result = true;

		this.messageStore = new DefaultMessageStore(this.messageStoreConfig, ...);
		//加载消息文件
		result = result && this.messageStore.load();

		return result;
	}

}

package com.alibaba.rocketmq.store;

public class DefaultMessageStore implements MessageStore {
	...

	public boolean load() {
		boolean result = true;
		...
		//加载CommitLog，CommitLog保存所有消息
		result = result && this.commitLog.load();
		//加载ConsumeQueue，ConsumeQueue保存消息在CommitLog文件中的位置
		result = result && this.loadConsumeQueue();

		return result;
	}

	private boolean loadConsumeQueue() {
		File dirLogic = new File(StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()));
		File[] fileTopicList = dirLogic.listFiles();
			for(File fileTopic : fileTopicList) {
				//ConsumeQueue是根据Topic来分组的，一个Topic保存为一个文件夹
				//Topic文件下保存这它所属的消息位置
				String topic = fileTopic.getName();
				//遍历该Topic下的所有QueueId
				File[] fileQueueIdList = fileTopic.listFiles();
				for(File fileQueueId : fileQueueList) {
					int queueId = Integer.parseInt(fileQueueId.getName());
					//创建ConsumeQueue
					ConsumeQueue logic = new ConsumeQueue(
						topic,
						queueId,
						...);
					this.putConsumeQueue(topic, queueId, logic);
				}
			}
		}
		return true;
	}
}

package com.alibaba.rocketmq.store;

public class CommitLog {
	...

	public boolean load() {
		//将文件加载到mappedFileQueue队列中
		boolean result =this.mappedFileQueue.load();
		return result;
	}
}

package com.alibaba.rocketmq.store;

public class MappedFileQeueue {
	...

	public boolean load() {
		File dir = new File(this.storePath);
		//列出CommitLog下的所有文件
		File[] files = dir.listFiles();
		if(files != null) {
			Arrays.sort(files);
			for(File file : files) {
				//加载文件
				MapedFile mapedFile = new MapedFile(file.getPath);

				mapedFile.setWrotePosition(this.mapedFileSize);
				mapedFile.setCommittedPosition(this.mappedFileSize);
				this.mapedFiles.add(mapedFile);
			}
		}

		return true;
	}
}

package com.alibaba.rocketmq.store;

public class MappedFile extends ReferencesResource {
	...

	public MapedFile(final String fileName, final int fileSize) {
		this.fileName = fileName;
		this.fileSize = fileSize;
		this.file = new File(fileName);
		...
		this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
		//直接将文件拷贝到一份虚拟内存中
		this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
		TotalMapedVitualMemory.addAndGet(fileSize);
		TotalMapedFiles.incrementAndGet();
		...
	}
}