cassandra 学习笔记

1). cassandra任何一个节点都可以被客户端访问。

2). 对cassandra某个节点的访问是通过调用org.apache.cassandra.service.Cassandra的内部类Client的相应接口实现的。

3). 2)中的Cassandra这个类包含了很多内部类和一个接口（Iface）。其中的Client和Processor两个内部类都是对Iface的实现，这保证了他们的内部的所有方法是一一对应的。

4). 当Client这个内部类中的某个方法被调用的时候，该方法会用他内部的send_开头的方法发送消息，并且用recv_开头的方法接收返回的内容，容，返回的内容可能是我们想要的数据，也可能是个异常的消息，如果是异常的消息，则会在客户端生成一个相应的异常并抛出，

5). Client端send_和recv_方法同目标节点的交互是分别通过oprot和iprot的实例完成的，这两个实例是负责输入输出的，具体的功能的实现在libthrift.jar中。

6). 节点和客户端通信的连接是由libthrift.jar中的TThreadPoolServer的实例实现的，这个实例在该节点最初启动的时候被生成，并且该实例内部还保有一个2)中提到的Processor实例。TThreadPoolServer实例给Processor实例提供了输入输出实例iprot和oprot,并且通过调用Processor的processprocess(TProtocol iprot, TProtocol oprot)接口来进一步的向内传递消息。

7). 节点最初启动的初始话过程是在org.apache.cassandra.service.CassandraDaemon的setup()中完成的。

8） 在Processor的processprocess(TProtocol iprot, TProtocol oprot)会解析iprot中传入的客户端的请求，并首先解析出要调用函数的函数名字，然后通过查询processMap_来决定究竟由那个ProcessFunction实例来接收处理消息,相应的ProcessFunction实例的process(int seqid, TProtocol iprot, TProtocol oprot)被激活并开始全权负责消息的处理和反馈。

9). 相应的的ProcessFunction的实例主要负责三件事：i，进一步处理iprot传入的消息 ii，将详细的信息转发给iface的相应方法处理 iii，将得到的反馈通过oprot返回给客户端。这里的iface实例实际上是org.apache.cassandra.service.CassandraServer的一个实例，在Processor的实例创建的时候(节点启动的时候)被装入了Processor实例,但是由于ProcessFunction类是Processor的内部类，所以ProcessFunction的实例也能直接访问。

10)以上可知，最终客户端的信息是交给CassandraServer的相应方法来处理的，而thrift的相关功能只是负责了客户端和节点间的交互（9160端口），而节点之间的交互并没有使用thrift的资源。

源码中对节点的如下称呼应该是等价的: end point , node ,  machine , datacenter , host。

    cassandra节点的启动main()在类org.apache.cassandra.service.CassandraDaemon中，细节在 setup()中。过程中会start一个CassandraServer的实例peerStorageServer。 peerStorageServer在建立的时候，内部会实例化一个 StorageService实例，在该StorageService实例初始化的过程中，该节点的所有功能服务会被配置激活，这些操作是在 StorageService的默认构造器中完成的。

    StorageService的构造器中大致做了如下几件事情：

    1)生成一个storageLoadBalancer_s实例负责负载均衡，现在还没有明白原理。

    2)生成一个endPointSnitch_实例，这个提供了对两个end_point进行比较的一个途径，基本上是判断两个end_point是不是同一个ip等

    3)启动了MessagingService，并且注册了一些handler实例。MessagingService是负责该end_point与其他end_point进行通信的。两个节点间通信的内容被封装在一个Message的实例里面。

       比如，如果节点A想向节点B获得一定的数据，那么A需要通过自己的MessageService向节点B发送一个Message实例，这个实例里面包含了如下信息：这个请求的类型（属于什么stage） ，这个请求要调用的B的哪个handler来处理，以及这个请求的其他具体内容。当节点B接收到节点A发送过来的Message实例后，会将根据这个 Message实例内部指定的handler信息,将该实例转发相应的handler去处理。当然这样做的前提是这个指定的handler已经在B节点注册了，而这个注册过程就是在StorageService启动的时候完成的。

    4)consistencyManager_：还没明白什么意思。

    5)StageManager的配置：

       这个stage的概念来自于“SEDA“( staged event driven architecture)中的”S”,参考。

       大致意思好像是可以将一个工作流程分为若干个阶段(stage)，然后给各个stage动态的分配线程资源来处理stage内定义的逻辑。stage和 stage之间的通信是通过任务队列完成的，当一个stage的逻辑执行完后，如果需要调用下一个stage来继续执行，那么就往下一个stage保有的任务队列中写入必要的任务信息。

       这样的SEDA结构的好处是：在实际的运行当中各个stage的忙闲程度是不一样的，可以通过将比较闲的stage上的线程资源分配给同期比较忙的stage来实现效率上的提高。

       在cassandra中，还是以3)中例子说明，在A节点发往B节点的Message实例中有一个“这个请求的类型”的信息，这个信息保存在 Message实例内header实例的type_上，是一个字符串。这个字符串标明了当MessageService获取了Message实例后究竟由那个 stage来负责执行指定的handler对象。因为handler对象只是定义了对Message的处理逻辑，所以需要stage里面的线程来对其进行执行。

       当前的stage只有4种，依次在StroageService的 /* All stage identifiers */标注下被定义，分别负责不同的任务，“ROW-READ-STAGE”是负责读取的，其他的含义还没有完全搞清楚，大概是负责修改，数据压缩和 http后台页面信息接收的。

       StageManager部分的源码还没哟看到，可能是负责各个stage间线程调度的吧。

    6)设置nodepicker_定义了当前节点对周围节点的查询策略，具体的还不清楚

当某个end point拿到一个key(比如”王老六”)并想取出他的相关信息的时候,这个节点是怎么知道这个key的相关信息是存放在哪些节点中的呢?
以下将用从客户端拿到的”get_clomun”请求为例，进行说明：

      “get_column”的相关信息会在CassandraServer的get_column(String tablename, String key, String columnPath)方法中被封装成一个readCommand实例，该对象简单包含了请求信息，另外也提供了一些别的方法。

      然后该command实例会最终被交给 StorageProxy.readProtocol(command,StorageService.ConsistencyLevel.WEAK)来处理，并期待这个方法能返回一个和key关联的columnFamily的数据，这些数据被装在一个row对象里返回的。

      在上述的StorageProxy.readProtocol(…)这个方法中，首先要做的事情就是根据提交的key(“王老六”)确定他的相关信息内容是存放在那些end point中的。而这个寻找end point的工作由StorageService.instance().getNStorageEndPoint(command.key)来完成。

      实际上，上述的StorageService的getNStorageEndPoint(String key)方法调用了他持有的实例”nodePicker_”的getStorageEndPoints(Token token) 来完成寻找end point这件事情。

      题外话：StorageService这个实例提供了本地节点数据存储和与其他节点交互的服务，这个实例是在节点启动之初被建立的，在建立的时候该实例会初始化一个nodePicker_放在其内部，这个nodePicker主要负责本地节点同其他节点的交互规则，也就是如何选择其他节点的策略。当前的策略有两种:RackAwareStrategy–机架敏感 和 RackUnawareStrategy–非机架敏感，默认策略是RackUnawareStrategy。

      所以默认情况下，nodePicker_就是一个RackUnawareStrategy实例，而他的 getStorageEndPoints(Token token)将被用来查找合适我们给出的key(“王老六”)的end point,这里合适的意思是“存有关于这个key的信息”。而这里提到的token是由key封装得来的，他对key进行了hash，并且继承了 Comparable接口。也就是说，这个token实例是可以和其他同类token比较大小的，而这样做的目的是为了“可以在Token组成的list 上进行二分查询Collection.binarySort，并定位查找目标的在list中的相对位置”–这种查找操作将被经常使用。默认情况下 token是一个BigIntegerToken的实例。

      nodePicker_最终会将查找的工作交给getStorageEndPoints(Token token, Map tokenToEndPointMap)来做，这个方法会返回一个查找到的end_point的数组。以下是对RackUnawareStrategy中的相应方法的说明（因为默认是使用这个方法）:
//此处的tokenToEndPointMap保存了该节点知道的所有其他节点，并且可以根据他们的token查找到

//此处的token是我们要查找的key(“王老五”)生成的token,他跟上面end point的token生成的算法是相同的，也都是BigIntegerToken

1.  
2. public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap)
3.     {
4.         //根据key寻址的切入点
5.         int startIndex;
6.         List list = new ArrayList();
7.         int foundCount = 0;
8.         //将key的set装入一个ArrayList中，并完成排序，（一位token是comparable的）
9.          tokens = new ArrayList(tokenToEndPointMap.keySet());
10.         .sort(tokens);
11.         //在排好序的tokens中二叉树搜索token，并返回token所在的位置的index
12.  
13.         //个人认为:根本别指望tokens里面能有我们要找的token，因为一个是end point的token集合,一个是查找key("王老六")生成的 
14.  
15.         //token，生成的时候都使用了hash，并且加入了大随即数，重合的概率很低。
16.  
17.         //这样做的主要目的是为了获得下面这个index，（参见binarySearch的说明可知）。
18.  
19.         //也就是：虽然下面这条语句不大可能从tokens中取出东西，但是生成的index将告诉我们，哪个end point的token离我们给出的
20.  
21.         //key"王老六"的token“最近”。而且，在节点情况固定下来的情况下，用当前这种方法，
22.  
23.         //key"王老六"的token确定的对应的end point也是不会改变的(每次都会取到这个end point，除非end point本身注册的变了)。
24.  
25.         //也就是说：这就成了一种方法，一种能够根据数据key("王老六")定位到一个end point的方法，
26.  
27.         //如果写数据时使用了这种策略找到一个end point 节点然后写进数据，那么手持相应的key在读数据的时候，同样使用这个策略，也
28.  
29.         //相应的能找到当时存入数据的那个节点，并把数据读出来。
30.           int index = .binarySearch(tokens, token);
31.         //基本上这个条件里面的语句肯定会被执行
32.  
33.         //那如果真的用key("王老六")从tokens中找到对应了呢？这个方法里的策略就是，那就直接拿这个被确定的节点来当 “邻居”节点了
34.  
35.         //而key的tokens和end_point的token之所以能混在一起说话，是因为他们都是BigIntegerToken，在同一个数据空间中
36.         if(index < 0)
37.         {
38.  
39.             //以下的运算将把key("王老六")的"邻居"节点给"翻译"出来
40.             index = (index + 1) * (-1);//得到可以插入的位置
41.             if (index >= tokens.size())
42.                 index = 0;
43.         }
44.         int totalNodes = tokens.size();
45.         // Add the node at the index by default
46.  
47.         //把我们找到的"邻居"节点放到一个要返回的list中，当然,对于一个key("王老六")的信息还可能在其他节点上存有副本(replicas)
48.  
49.         //再下面的操作会将这些存有副本的节点也一同取出
50.         list.add(tokenToEndPointMap.get(tokens.get(index)));
51.         foundCount++;
52.         //本来index是从0…一直到size()的一条线，如果我们用如下的方式选取index，就好似将所有的index数值放在了一个圆圈上，然后按
53.  
54.         //从小到大的方向在选取下一个index。startIndex就是一个切入点，本质上Partitioner提供了这个可被"切入"的结构和线索
55.  
56.         //猜想这也是为什么叫Partitioner的原因吧
57.         startIndex = (index + 1)%totalNodes;
58.         // If we found N number of nodes we are good. This loop will just exit. Otherwise just
59.         // loop through the list and add until we have N nodes.
60.         // replicas_对应storage-conf.xml中的最大副本数量
61.    
62.  
63.         //将存有副本的节点也一同取出，就是最初"邻居"节点往后的若干个节点，具体若干几个是在storage-conf.xml的
64.  
65.         //标签中定义的
66.         for (int i = startIndex, count = 1; count < totalNodes && foundCount < replicas_; ++count, i = (i+1)%totalNodes)
67.         {
68.             if( ! list.contains(tokenToEndPointMap.get(tokens.get(i))))
69.             {
70.                 list.add(tokenToEndPointMap.get(tokens.get(i)));
71.                 foundCount++;
72.             }
73.         }
74.  
75.         //统一下节点监听端口的信息
76.         retrofitPorts(list);
77.         return list.toArray(new EndPoint[list.size()]);
78.     }
79.