public List<Event> readEvents(int numEvents) throws IOException {
if (!committed) {
if (!currentFile.isPresent()) {//为空,假设Optional包括非null的引用(引用存在),返回true
throw new IllegalStateException("File should not roll when " +
"commit is outstanding.");
}
logger.info("Last read was never committed - resetting mark position.");
currentFile.get().getDeserializer().reset();
} else {//已经committed成功
// Check if new files have arrived since last call
//Returns true if this holder contains a (non-null) instance
if (!currentFile.isPresent()) {//为空,获取下一个文件,初次调用
currentFile = getNextFile();
}
// Return empty list if no new files
if (!currentFile.isPresent()) {//为空,已经没有可读的文件了
return Collections.emptyList();
}
//其他的说明是currentFile眼下还在读
}
EventDeserializer des = currentFile.get().getDeserializer();
List<Event> events = des.readEvents(numEvents);//加入event的body
/* It's possible that the last read took us just up to a file boundary.
* If so, try to roll to the next file, if there is one. */
if (events.isEmpty()) {
retireCurrentFile(); //改名字
currentFile = getNextFile();//换下一个文件
if (!currentFile.isPresent()) {
return Collections.emptyList();
}
events = currentFile.get().getDeserializer().readEvents(numEvents);//继续读,加入event的body
}
if (annotateFileName) {
String filename = currentFile.get().getFile().getAbsolutePath();
for (Event event : events) {
event.getHeaders().put(fileNameHeader, filename);//加入header
}
}
committed = false;
lastFileRead = currentFile;
return events;
}
/** Commit the last lines which were read. */
@Override
public void commit() throws IOException {
if (!committed && currentFile.isPresent()) {
currentFile.get().getDeserializer().mark();
committed = true;
}
}
这种方法说明满足两个条件就能够:一、向trackerFile写入读到的记录位置,mark()方法会将syncPosition写入trackerFile,而ResettableFileInputStream中的position用来暂存位置添加的,待到何时会syncPosition=position,这样是为了防止出现异常时用于恢复丢失的数据;二、将committed =
true。两个条件:一个是committed=false,这个运行完readEvents最后会置为false;二、currentFile“非空”,代表有正在读的文件。假设committed在readEvents中開始时为false,说明:一、event提交到channel时出现了问题,没有运行reader.commit;二、currentFile已经“为空”,说明没有能够读的文件。这两点也体如今readEvents開始部分,committed=false时,假设没有可读文件就会抛出异常File
should not roll when commit is outstanding.";假设是在提交到channel时出问题会通过currentFile.get().getDeserializer().reset()又一次撤回到上次正确提交channel的位置,这样能够使得不丢失数据。