Cassandra源代码阅读笔记

1 、目的

满足大数据量、大量随机的读写操作应用场景下的数据存储需求。更是用于实时在线应用。

2 、功能概要

BigTable 存储结构+Dynomo 分布式模型

多副本

高可用

易增量扩展

最终一致性

最小化管理

3 、数据类型描述

Column

字段，包含name ，value, IClock (用于检测和解决冲突，目前使用 timestamp )和isMarkedForDelete

SuperColumn

包含多个字段的字段，字段（IColumn ）内部又是由多个字段组成



ColumnFamily

字段集合，类似于数据库中一行，Column name 到IColumn 的Map

addColumn （输入IColumn ）：若不存在或者时间戳较旧，使用新IColumn

addColumns （输入ColumnFamily ）：依次对每个IColumn 做addColumn

resolve 过程：

字段集合的合并。将每一个ColumnFamily 做addColumns ，然后再更新DeletionTime



Row

key 和ColumnFamily 的键值对



［使用的name->value 的map ，无固定schema, 一行中可以有任意多个column ，且每行的column 可不同］

［每个IColumn 可能本身又是一个map ］

［每一个Column 都有自己的时间戳，用于记录合并时作判断］



Keyspace

一般是应用名称，可认为是逻辑的表，对于每一个Keyspace 可以配置其副本策略、ColumnFamily （即表结构），缓存等

4 、 Partitioning

一致性哈西：输出区间为一个环，每个点管理一个区间，key 找该区间确定点。点的退出或者加入之影响该点相邻区间的点。

针对基本的一致性哈西分布不均匀且不能根据节点能力强弱分配的缺点，Dynamo 是让每个点管理环 中的多个位置，而Cassandra 是让负载轻的节点可以在环上Move 来均衡负载。



用来到环上标识位置的东西叫Token，可以是整型或者字符串等。

Partitioner 类型

RandomPartitioner ：对key 做MD5 再生成BigIntegerToken

OrderPreservingPartitioner ：直接用key 生成StringToken

CollatingOrderPreservingPartitioner ：对key 做getCollationKey 后生成BytesToken ，貌似优势是字符串比较比StringToken 快



生成Token 后，AbstractReplicationStrategy 用Token 取得位置信息



getNaturalEndpoints——按key找对应的节点位置

按key的searchToken，从tokenMetadata中找到环上下一位置的节点Token；

再按节点Token找出InetAddress数组（先找getCachedEndpoints，没有的话重新依据副本策略（参考副本策略）calculateNaturalEndpoints）

5 、副本机制

NWR

表决方式来提升一致性。

N—— 副本个数，N 为配置的ReplicationFactor

W—— 每次保证写入的个数，W 的选取：

输入：naturalTargets ，即N ；hintedTargets ，备选集合总数，包含了故障节点的备份点；ConsistencyLevel ，一致性级别

bootstrapTargets 为hintedTargets 和naturalTargets 的差值，

如果级别为ConsistencyLevel.ANY ，W 等于1

如果级别为ConsistencyLevel.ONE ，W 等于bootstrapTargets ＋1

如果级别为ConsistencyLevel.QUORUM ，W 等于(naturalTargets/2)+bootstrapTargets ＋1 ，即大多数

如果级别为ConsistencyLevel.DCQUORUM 或者DCQUORUMSYNC ，W 等于naturalTargets

如果级别为ConsistencyLevel.ALL ，W 等于bootstrapTargets ＋naturalTargets

R—— 每次读取多少个副本来作为备选集，R 的选取：weakRead 为1 ，strongRead 输入：naturalTargets＝hintedTargets ＝ReplicationFactor ，备选集合总数；ConsistencyLevel ，一致性级别，类似的：

如果级别为ConsistencyLevel.ANY ，R 等于1

如果级别为ConsistencyLevel.ONE ，R 等于1

如果级别为ConsistencyLevel.QUORUM ，R 等于(naturalTargets/2)1 ，即大多数

如果级别为ConsistencyLevel.DCQUORUM 或者DCQUORUMSYNC ，R 等于naturalTargets

如果级别为ConsistencyLevel.ALL ，R 等于naturalTargets

副本策略

LocalStrategy：本地节点

SimpleStrategy： 不考虑机柜因素， 环上依次N个位置的节点（对应于旧版本 RackUnawareStrategy ）

OldNetworkTopologyStrategy ： 在primaryToken 之外，先找一个其他数据中心的，再找一个其他机柜的，然后按环上位置依次找没使用的 （对应于旧版本 RackAwareStrategy ）

NetworkTopologyStrategy：N个副本在数据中心和机柜间均匀部署 （对应于旧版本 DatacenterShardStategy ）



旧版本中的策略类型：

RackUnawareStrategy ：不考虑机柜因素，从Token 位置依次取N 个

RackAwareStrategy ：考虑机柜因素，在primaryToken 之外，先找一个处于不同数据中心的点，然后在不同机柜找

DatacenterShardStategy ：要保证(N-1)%2 个点与primaryToken 不再同一个数据中心



AbstractEndpointSnitch implements IEndpointSnitch

用来维护Endpoint的数据中心和机架数据，并且缓存Token映射到的Endpoints

6 、故障和扩展

故障和扩展将导致节点的状态和管理区间发生变化，需要机制来检测和同步这些变化。

路由更新

管理员手工进行节点的加入和移除，Gossip 协议节点间同步路由的变化（list 发送给其他节点，收到list 的节点合并两个list ）

"/Storage/Seeds/Seed" 配置中包含多个seed 信息，以此作为Gossip 通信的起点。

数据迁移

节点加入时，

新节点在bootstrapping 阶段根据自己管理的区间，去相应的位置索取数据。

从路由信息计算出自己要管理区间当前所在位置Multimap<Range, InetAddress> ，然后对每个位置请求Collection<Range>

节点移除时，

根据Gossiper 检测出的节点变化发现节点被移除，根据移除 节点找出需要变更的区间中需要变更为自己管理的区间，为这些区间找到源位置，然后对每个位置请求Collection<Range>

Gossiper ：

维护和同步节点状态和路由信息。

服务启动后首先更新localEndPoint ，seed ，localState 等信息，启动GossipTimerTask 开始定期进行Gossip 通讯

使用GossipDigest （EndPointState 的摘要，包含EndPoint 的版本状态信息）列表生成GossipDigestSynMessage 发送

一旦收到此消息，整理出需要push 的EndPointState 列表（我是新的）和需要poll 的GossipDigest 列表（你是新的），生成GossipDigestAckMessage 回应

一旦收到此消息，根据EndPointState 列表更新本地错误检测和状态维护，根据GossipDigest 列表整理出对应的EndPointState 列表生成GossipDigestAck2Message 回应

一旦收到此消息，根据EndPointState 列表更新本地错误检测和状态维护

7 、 Service

StorageProxy—— 存储服务

mutate

输入：List<RowMutation>

输出：void

工作：

按table 和key 做getNaturalEndpoints 返回满足的InetAddress 列表

进而getHintedEndpointMap 返回Map<InetAddress, InetAddress> ，该映射为每一个地址选择了一个故障备份地址［如果Endpoint 是isAlive ，则备份为自身；若不是isAlive ，则在临近位置找一个尚未使用的地址作备份］

然后对HintedEndpointMap 中的每一个地址处理：如果地址为本地，直接执行；如果地址为外地，直接发送；如果要发到备份地址，标记Hint 标记为原始地址，发到备份地址［即点A 故障，选取点B 作备份，把要写入的数据发给点B ，但是告诉点B 这是点A 的，点B 后期写回给点A ］

[ 无任何一致性保证]

mutateBlocking

输入：List<RowMutation> ， ConsistencyLevel

输出：void

工作：同mutate ，唯一的差别的请求发出后还等待回应

［同时写出N 份，必须在超时前收回W 份响应认为成功，否则报错］



weakReadRemote

输入：List<ReadCommand>

输出：List<Row>

工作：

按key 去findSuitableEndPoint （先找本地，再找相同DataCenter ，最后依次找，找到后判断是否isAlive ）

将Message 做sendRR 到SuitableEndPoint ，得到List<IAsyncResult>

使用IAsyncResult （AsyncResult ）真正去get 数据，若超时则重试（重试Todo ）

对数据反序列化得到ReadResponse

根据一致性检查判断是否需要修复副本（

［写请求，回掉中修改状态和数据，然后再显式get 去拿状态和数据做事］

strongRead

输入：List<ReadCommand>

输出：List<Row>

工作：

按key 去findSuitableEndPoint

按key 去找到所有其他Live 副本集合getLiveNaturalEndpoints

将Message 做sendRR ，对SuitableEndPoint 发ReadMessage ，对其他发DigestOnly的Message ，得到List<QuorumResponseHandler>

使用QuorumResponseHandler 真正去get 回应数据，并使用IResponseResolver（ReadResponseResolver ）判断数据和Digest 是否匹配，否则throws DigestMismatchException 重试一次

如果不是isDigestQuery ，记录ColumnFamily 到rowList 、EndPoint到endPoints ，对于新得到的ColumnFamily ，［使用rowList 中其他结果进行合并得到resolved 结果，再使用resolved 结果找出rowlist 中不同的条目生成写diff 包提交给ReadRepairManager 做回写］

如果是isDigestQuery ，如果rowList 中任意结果的digest （MD5 ）与ReadResponse 不一致，throws DigestMismatchException

［不是份所有R 份都是读结果；只有部分读结果，其余读摘要信息］

［不是读请求都读回来才回应；读>R 且<N 份，读回来R 份且包含了ReadMessage 的回应就认为成功，否则报错］



基于此，实现了两种存储服务接口，一种采用thrift 接口，一种采用Hadoop Avro 接口。

AntiEntropyService—— 同步／错误解决服务

对ColumnFamily 生成哈西，根据TreeRequest/Response 会话与其他节点交换MerkleTrees ，对于发现的不一致发起修复。

EmbeddedCassandraService—— 嵌入式 in-memory 存储服务

启动线程来响应来自thrift 接口的存储请求

StorageLoadBalancer—— 负载均衡服务

Gossiper 协议发ApplicationState 查断看各节点状态，根据状态变化在需要时修改节点负责的Token 。