GroupCommitService

run

public void run() {CommitLog.log.info(this.getServiceName() + " service started");while (!this.isStopped()) { // 未stop 一直循环try {this.waitForRunning(10);  // 每隔10ms允许1次this.doCommit();} catch (Exception e) {CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);}}// Under normal circumstances shutdown, wait for the arrival of the// request, and then flushtry {Thread.sleep(10);} catch (InterruptedException e) {CommitLog.log.warn("GroupCommitService Exception, ", e);}synchronized (this) {this.swapRequests(); // 停止前，再处理1次}this.doCommit();CommitLog.log.info(this.getServiceName() + " service end");
}

waitForRunning

protected void waitForRunning(long interval) {if (hasNotified.compareAndSet(true, false)) { // 默认false wakeup后设置truethis.onWaitEnd();  // return;}//entry to wait  阻塞waitPoint.reset(); // CountDownLatch2 setState(state)  重置statetry {waitPoint.await(interval, TimeUnit.MILLISECONDS);} catch (InterruptedException e) {log.error("Interrupted", e);} finally {hasNotified.set(false);this.onWaitEnd();}
}

onWaitEnd

protected void onWaitEnd() {this.swapRequests();
}

swapRequests

其实就是交换

private void swapRequests() {// 新消息放requestsWrite List<GroupCommitRequest> tmp = this.requestsWrite;this.requestsWrite = this.requestsRead; // requestsRead这个时候是空的// 刷盘用this.requestsRead = tmp;  
}

doCommit

private void doCommit() {synchronized (this.requestsRead) {if (!this.requestsRead.isEmpty()) { // 有数据for (GroupCommitRequest req : this.requestsRead) {// There may be a message in the next file, so a maximum of// two times the flush 消息有可能存在2个文件boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset(); //是否刷盘成功for (int i = 0; i < 2 && !flushOK; i++) {CommitLog.this.mappedFileQueue.flush(0);  // 刷盘  根据flushedWhere的位置通过findMappedFileByOffset找MappedFile   flushedWhere = mappedFile.getFileFromOffset() + offsetflushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();}//  doCommit中每对一个request处理完成后，会调用wakeupCustomerreq.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT); // 通知consumer}long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();if (storeTimestamp > 0) { // 设置存盘时间CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);}this.requestsRead.clear(); // 刷盘循环处理完，清空requestsRead列表} else {// Because of individual messages is set to not sync flush, it// will come to this processCommitLog.this.mappedFileQueue.flush(0);}}
}

流程图