Cassandra 存储引擎

Cassandra 存储引擎

1。每个column family都对应一张Memtable，每张Memtable都会被flush到一张磁盘的SSTable
2。Memtable在下面条件下被Flush:
    a.超出一定大小
    b.keys数量超出一定大小(128是缺省大小)
    c.到一定时间(由客户端决定)
3。Memtable被Flush会产生2个文件
    a.数据文件即SSTable，SSTable是根据key排序的key/value对文件
    b.索引文件SSTable Index：
        (key,offset)对（offset指向数据文件SSTable)
        Bloom filter，其主要作用是先判断key是否存在于该SSTable，以避免对磁盘的IO查询
4。当commit log所有column families都Flush到磁盘，该Memtable被删除
5。当数据文件SSTable累计到一定时间，就被合并到一个新的数据文件（同时创建一个新的索引文件)，具体为：
    a合并Keys
    b合并Columns
    c丢弃墓碑(tombstones)


Memtable
memtable是一个内存表，内部通过一个NonBlockingHashMap来维护这样成对的记录，NonBlockingHashMap对ConcurrentHashMap做了些改进
CurrentSize记录当前所有记录占用字节大小
CurrentObjecCount记录当前所有记录条数
对Memtable的主要操作接口就是put(String key,ColumnFamily columnFamily)
这个操作的逻辑如下：
    1。调用NonBlockingHashMap的putIfAbsent()方法，如果该key不存在，只要累加currentSize和currentObjectCount两变量，如果存在，则累加差值
getSortedKeys()方法用于获取NonBlockingHashMap里的key列表
SSTableReader writeSortedContents(List sortedKeys)方法用于将NonBlockingHashMap的记录导出到一个SSTable，并返回对这个SSTable的Reader


Table
Table是最上层的封装接口
Table通过TableMetadata来维护Table的元数据，这些元数据主要Column Family信息，通过ColumnFamilyStore来完成对一个ColumnFamily的存储。下面是主要的接口
void apply(RowMutation mutation,DataOutputBuffer serializedMutation)，主要逻辑如下
1。提交这次RowMutation到CommitLog
2。遍历RowMutation包含的ColumnFamily列表做如下操作：
    a调用该ColumnFamily对应的ColumnFamilyStore对象的apply方法作用到Memtable，
    b如果Memtable到了一个阀制需要Flush时则调用ColumnFamilyStore的swithMemtable方法
    c上面的方法提交一个Flushable任务

Memtable Flush分为两个过程：
1－排序 2－写磁盘
排序是一个消耗CPU的工作，写磁盘是消耗磁盘IO的工作，所以两个任务放到不同的线程池执行(FlushSorter和FlushWrite)，增加并行处理
排序最终结果是一个SSTableReader，然后通过addSSTable()函数被提交的CompactionManager的任务里面。
Flush是否结束通过一个Condition来标识，最后调用onMemtableFlush


CommitLog
Commit Log 跟踪每一次写操作，目的是当系统crash的时候能恢复还没有被写进磁盘的Memtable。每一个Commit Log维护一个头信息CommitLogHeader，头信息包含两个数组，一个bit数据，一个Long数组，以及这两个数组的长度，

当一个ColumnFamily第一次被写入，CommitLogHeader里面对应的bit位被标志为1，当ColumnFamily被Flush到磁盘则对应bit位被标志为0。这个标识用于跟踪那个CommitLogs在Memtalbe Flush后能够被丢弃。同时，当一个ColumnFamily被Flush到磁盘，其对应的Long被更新为到CommitLog文件的偏移量，这样可以帮助加速修复

每个Commit Log到达一个时间后被滚动建立一个新文件，新Commit Log文件会继承老文件里的脏位

这样经过一段时间后就会产生大量的Commit Log文件。当我们Flush一个ColumnFamily我们更新它的活动的Commit Log的bit位


数据流向
数据插入流向
数据被包装成RowMutation对象被MessagingService发送到EndPoint，被RowMutationVerbHandler处理，具体处理很简单，就是调用RowMutation的apply()方法，RowMutation.apply()方法则调用Table.apply(RowMutation row,DataOutputBuffer buffer)方法。该方法处理流程如下：
    a.CommitLog.add(mutation,DataOutputBuffer buffer)方法，把当前的操作记录到Commit Log文件中
    b.遍历RowMutation包含的所有ColumnFamily，做下面操作
        b.1获取对应ColumnFamily的ColumnFamilyStore
        b.2调用ColumnFamilyStore.apply()方法
    c.调用ColumnFamilyStore.switchMemtable()方法

数据查询流向
数据查询最终落到关键函数ColumnFamilyStore.getColumnFamilyInternal(QueryFilter filter,int gcBefore)


SSTableScanner用于迭代SSTable里面的所有Row(IteratingRow)，IteratingRow又能迭代出所有的Column
CompactionIterator是一个Wrap迭代器，将要compact的SSTableReader的SSTableScanner组成一个CollatingIterator。CollatingIterator的外层又Wrap一层ReducingIterator，ReducingIterator用于去除重复记录，CompactionIterator认为key相同的Row为重复记录，需要reduce，具体的reduce逻辑为：
    如果是SuperColumn，就合并，否则根据时间判断用最近的记录

Compact过程
首先把ColumnFamilyStore所有的SSTable分到不同的Bucket里面，然后以Bucket为单位调用doFileCompaction函数做compact