对于Flume来说主要有两个Channel：Memory，File；对于线上环境主要以FileChannel为主，因此这里主要讨论它的实现：

在FileChannel里主要由一个WAL的log和一个内存队列组成：

FileChannel的Queue主要又以下几个部分组成：

private final EventQueueBackingStore backingStore;

private final InflightEventWrapper inflightTakes;

private final InflightEventWrapper inflightPuts;

其中backingStore代表了queue在持久化存在，使用了内存映射文件的方式；每次对queue的读写操作都记录在backingStore的overwritemap（update in place）中，当进行checkpoint的时候合并到elementsBuffer并持久化到磁盘；所有未提交的正在读写数据都分别保存在inflight结构中，当checkpoint时一并进行持久化，为回滚时使用；

在inflight中存储了transactionid->fileid以及transactionid->eventptr的映射，具体存储在backingStore里的则是eventptr（fileid,offset）;

Checkpoint file的文件结构如下：

File Header：1029 bytes

Eventptr;

在File header里前8个字节存储了版本号，接下来24个字节是sequeuece no.(类似rdbms的scn)，接下来4个字节存储了checkpoint的状态；

作为WAL的Log主要存储了（transactionid,sequenceNo,Event）,每次读写都先在log里写入event，对于写操作会拿到eventptr放入queue中；而commit和rollback操作在log中的记录形式是(transactionid,sequenceNo，OP={commit,rollback})；

这两个结构主要是体现在FileBackedTransaction中如下：

FileBackedTransaction extends BasicTransactionSemantics

......

LinkedBlockingDeque takeList;

LinkedBlockingDeque putList;

long transactionID;

Log log;

FlumeEventQueue queue: EventQueueBackingStoreFile

其中queue = log.getFlumeEventQueue();

首先看put/take path以及commit：

1. doPut(Event event)->

queue.addWithoutCommit(ptr, transactionID)

log.put(transactionID, event)->

synchronized LogFile.Writer.put(ByteBuffer buffer)

putList.offer(ptr)

2. doTake() ->

FlumeEventPointer ptr = queue.removeHead(transactionID);

takeList.offer(ptr),

log.take(transactionID, ptr); ->

synchronized LogFile.Writer.take(ByteBuffer buffer)

Event event = log.get(ptr);

3. doCommit()->

if(puts > 0) {

log.commitPut(transactionID);

synchronized (queue) {

while(!putList.isEmpty()) {

queue.addTail(putList.removeFirst())

queue.completeTransaction(transactionID);

}

}

else if (takes > 0) {

log.commitTake(transactionID);->

logFileWriter.commit(buffer);

logFileWriter.sync();

queue.completeTransaction(transactionID);

queueRemaining.release(takes);

}

}

从上面的代码可以看出，对于每一个put/take都会记录一条oplog到log里,当commit的时候会对log进行sync到磁盘持久化，同时会把event指针存放到queue上；这里的log就类似于mysql里的binlog(binlog_format=statement)，而这里的queue存放的是指向event的指针；

简例：FileChannel如下，对FileChannel put了2个消息，a,b；则在log,queue里的存储状态如下，Log里存储了（transactionid,sequenceNo，Event），queue则存储了eventptr；

Queue：ptr->a,ptr->b

WAL log:(1,1,put a),(1,2,put b),(1,3,commit)

当实例crash时，通过log来恢复queue的状态，类似rdbms一样，replay是很耗时的操作，因此会定期对queue进行checkpoint：

Log在初始化的时候会启动一个调度线程workerExecutor,由调度线程定期（checkpoint interval）调度一个backgroupWorkder来进行非强制性checkpoint；

Log.writeCheckpoint(Boolean force): tryLockExclusive->

synchronized queue.checkpoint->

backingStore.beginCheckpoint();//检查是否checkpoint正在进行;同时进行标记checkpoint开始，并同步MMAP file;

inflightPuts.serializeAndWrite();//

inflightTakes.serializeAndWrite();//将inflightputs/takes序列化并写到相应文件

backingStore.checkpoint();->

setLogWriteOrderID(WriteOrderOracle.next());

writeCheckpointMetaData();

//copy from overwriteMap to elementsBuffer(MMAP)

//标记checkpoint结束，并同步文件

简例：接上例，在a,b提交后，这时进行了一次checkpoint（存储在磁盘上的checkpoint则是2个指针ptr->a,ptr->b），此时scn=4；之后，又完成了一个take transaction ,ptr to a 也同时被删除；如果这时Flume crash，queue从checkpoint中重建，并且取得checkpoint scn=4,则replay这之后的log进行crash recovery；在恢复后，立刻执行一次checkpoint.

queue:ptr->b

WAL log:(1,1,put a),(1,2,put b),(1,3,commit),(2,5,take a),(2,6,commit)

[@more@]