package com.alibaba.rocketmq.store;

public class CommitLog {
	...

	public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
		//设置消息存储时间
		msg.setStoreTimestamp(System.currentTimeMillis());

		AppendMessageResult result = null;

		String topic = msg.getTopic();
		int queueId = msg.getQueueId();
		...
		//获取文件队列中最后一个文件，将新消息添加到队尾。
		//这里使用同步锁，获取到的是最后一个文件。
		MapedFile mapedFile = this.mapedFileQueue.getLastMapedFileWithLock();
 		//使用同步锁添加消息
		synchronized(this) {
			if(null == mapedFile || mapedFile.isFull()) {
				//如果MapedFile已满则在文件队列新建一个MapedFile，重新获取最后一个文件。
				mapedFile = this.mapedFileQueue.getLastMapedFile();
			}
			if(null == mapedFile) {
				//如果新建mapedFile失败，则直接返回错误
				return new PutMessageResult(PutMessageResult.CREATE_MAPEDFILE_FAILED, null);
			}
			//将消息添加到文件中，这里消息还只是暂存在虚拟缓存中，只有调用了刷盘操作，消息才真正写入磁盘中
			result = mapedFile.appendMessage(msg, this.appendMessageCallback);
			...

			//将消息在Commit Log文件中的位置，大小等信息写入Consume Queue
			DispatchRequest dispatchRequest = new DispatchRequest(//
            	topic,// 1 主题
                queueId,// 2 队列ID
                result.getWroteOffset(),// 3 写入位置
                result.getWroteBytes(),// 4 字节数
                tagsCode,// 5 Tag 标识
                msg.getStoreTimestamp(),// 6 存储时间
                result.getLogicsOffset(),// 7 逻辑位置
                msg.getKeys(),// 8 消息KEY
                /...);

            this.defaultMessageStore.putDispatchRequest(dispatchRequest);

			...
		}// 结束同步锁

		//构造消息添加成功返回结果
		PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);

		//消息成功添加到文件中后，调用刷盘操作，将消息从内存写入磁盘文件中。

		GroupCommitRequest request = null;

		// 默认是异步刷屏操作
		if(FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
			// 同步刷屏操作
		} else {
			// 异步刷盘操作
			this.flushCommitLogService.wakeup();
		}

		return putMessageResult;
	}

	abstract class FlushCommitLogService extends ServiceThread {

	}

	class FlushRealTimeService extends FlushCommitLogService {
		...

		public void run() {
			while(!this.isStoped()) {
				boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();

				//定时刷盘时间间隔
				int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
				...
				if(flushCommitLogTimed) {
					Thread.sleep(interval);
				} else {
					this.waitForRunning(interval);
				}
				//写入文件
				CommitLog.this.mapedFileQueue.commit(flushPhysicQueueLeastPages);
				...
			}
		}
	}
}

package com.alibaba.rocketmq.store;

public class MapedFileQueue {
	...

	public boolean commit(final int flushLeastPages) {
		boolean result = true;
		MapedFile mapedFile = this.findMapedFileByOffset(this.committedWhere, true);
		if(mapedFile != null) {
			...
			//写入文件
			mapedFile.commit(flushLeastPages);
			...
		}

		return result;
	}
}

package com.alibaba.rocketmq.store;

public class MapedFile extends ReferenceResource {
	...

	public int commit(final int flushLeastPages) {
		...
		//将ByteBuffer写入磁盘
		this.mappedByteBuffer.force();
		...

		return this.getCommittedPosition();
	}
}

package com.alibaba.rocketmq.store;

public class DefaultMessageStore implements MessageStore {
	...

	/**
     * 分发消息服务，将消息位置写入Consume Queue
     */
	class DispathMessageService extends ServiceThread {
		private volatile List<DispatchRequest> requestsWrite;
        private volatile List<DispatchRequest> requestsRead;

        ...

        public void putRequest(fianl DispatchRequest dispatchRequest) {
        	...
        	synchronized(this) {
        		//加入写请求队列
        		this.requestsWrite.add(dispatchRequest);
        		...
        	}
        	...
        }

        private void swapRequests() {
        	//交换读写请求队列，写请求队列负责添加新的请求，读请求队列负责将处理请求
        	List<DispatchRequest> tmp = this.requestsWrite;
        	this.requestsWrite = this.requestsRead;
        	this.requestsRead = tmp;
        }

        private void doDispatch() {
        	if(!this.requestsRead.isEmpty()) {
        		//处理可读队列的请求
        		for(DispatchRequest req : this.requestsRead) {
        			...
        			//将消息添加到ConsumeQueue
        			DefaultMessageStore.this.putMessagePositionInfo(
        				req.getTopic(), req.getQueueId, ...);
        			...
        		}
        		...
        		this.requestsRead.clear();
        	}
        }

        public void run() {
        	while(!this.isStoped()) {
        		this.waitForRunning(0);
        		this.doDispatch();//轮询执行分发请求，将消息写入ConsumeQueue
        	}
        }
	}

	/*
	 * 将消息写入ConsuemeQueue文件
	 *
	 */
	public void putMessagePositionInfo(String topic, int queueId, long offset, int size, ...) {
		//1.查找该Topic的ConsumeQueue是否存在，不存在则新建
		ConsumeQueue cq = this.findConsumeQueue(topic, queueId);
		//2.将ConsumeQueue写入文件
		cq.putMessagePositionInfoWrapper(offset, size, tagsCode, storeTimestamp, logicOffset);
	}
}

package com.alibaba.rocketmq.store;

/**
 * 消费队列
 */
public class ConsumeQueue {
	...

	public void putMessagePositionInfoWrapper(long offset, int size, long tagsCode, ...) {
		final int MaxRetries = 5;//ConsumeQueue如果写入失败，则重新尝试次数
		boolean canWrite = this.defaultMessageStore.getRunningFlags().isWriteable();//判断当前存储层是否可写
		//最多尝试5次写文件
		for(int i = 0; i < MaxRetries && canWrite; i++) {
			//写入文件
			boolean result = this.putMessagePositionInfo(offset, size, tagsCode, logicOffset);
			if(result) {
				//写入成功
				...
				return;
			} else {
				//写入失败

				try {
					Thread.sleep(1000);//1秒后重试
				} catch(InterruptedException e) {
					log.warn("", e);
				}
			}
		}

		//如果5次都写入失败，则停止存储层对外写服务，但仍然还可以读。
		this.defaultMessageStore.getRunningsFlags().makeLogicsQueueError();
	}

	/**
     * 将ConsumeQueue写入文件，文件存储一个20字节的信息
     */
	private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode, final long cqOffset) {
		...

		this.byteBufferIndex.flip();
		this.byteBufferIndex.putLong(offset);
		this.byteBufferIndex.putInt(size);
		this.byteBufferIndex.putLong(tagsCode);

		//获取最后一个MapedFile
		MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile(expectLogicOffset);
		if(mapedFile != null) {
			...
			//写入文件
			return mapedFile.appendMessage(this.byteBufferIndex.array());
		}

		return false;
	}
}