cassandra读源码---Streaming

时间:2021-04-23 20:22:12

前言

cassandra的很多过程需要网络传输模块,需要在各个节点直接发送文件。包括加入节点,删除节点引起的不同节点的负责ring环的key值发生了变化,导致sstable需要在各个节点中移动。

整体过程

两个节点会创建一个相似的对称的StreamSession。主要是下面四个阶段。一个stream会话包含了多个文件,每个节点既有发送task,也会有接收task.涉及到的可参看的java知识有

1. 如何进行数据文件的串行化,使得在网络中传输。然后反串行化,重新写入到SSTable中

2. 一次会话中的读写需要分离吗?

3. 传输文件需要占用网络带宽,和节点的I/O资源,如何控制stream的速率?

4. 一次会话中包含多个文件的传输,某个文件失败了,如果处理?

5. 多个table之间的streaming应该是分离的,这之间如何管理

cassandra读源码---Streaming

详细解释

cassandra读源码---Streaming

1. 连接初始化

a. 节点会创建一个新的StreamSesssion。init,然后start。会创建一个ConnectionHandler,去创建两个connections,一个incoming,一个outgoing。

然后发送一条StreamInit 消息。

 /**
* Bind this session to report to specific {@link StreamResultFuture} and
* perform pre-streaming initialization.
*
* @param streamResult result to report to
*/
public void init(StreamResultFuture streamResult)
{
this.streamResult = streamResult;
StreamHook.instance.reportStreamFuture(this, streamResult);
} public void start()
{
if (requests.isEmpty() && transfers.isEmpty())
{
logger.info("[Stream #{}] Session does not have any tasks.", planId());
closeSession(State.COMPLETE);
return;
} try
{
logger.info("[Stream #{}] Starting streaming to {}{}", planId(),
peer,
peer.equals(connecting) ? "" : " through " + connecting);
handler.initiate();
onInitializationComplete();
}
catch (Exception e)
{
JVMStabilityInspector.inspectThrowable(e);
onError(e);
}
}

(b)一旦收到了StreamInit message,follower就会创建自己的StreamSession,如果不存在的的话就创建,

然后将它附到自己的ConnectionHandler的socket。

(c)当incoming和outgoing connections都建立起来了,StreamSession 调用onInitializationComplete 方法去

进行下一个阶段。Streaming 准备阶段。

2. Streaming 准备阶段

(a)当调用onInitializatioinComplete()方法时,这个方法发送一个PrepareMessage包含将要发送的文件/sections,

(包装在StreamTransferTask,每个cf分离的)以及需要对方返回的task。如果没有从对方那边收到任何东西,就直接进入Streaming 阶段,否则等待对方prepareMessage。

(b)一旦收到PrepareMessage,接收者会记录下将会接收到的files/sections。并返回一个摘要给发送者。发送完消息后,接收者就进入到Streaming 阶段了。

(c)当发送者收到接收者PrepareMessage,记录下接收到的files/sections。然后进入到Streaming 阶段。

/**
* Call back when connection initialization is complete to start the prepare phase.
*/
public void onInitializationComplete()
{
// send prepare message
state(State.PREPARING);
PrepareMessage prepare = new PrepareMessage();
prepare.requests.addAll(requests);
for (StreamTransferTask task : transfers.values())
prepare.summaries.add(task.getSummary());
handler.sendMessage(prepare); // if we don't need to prepare for receiving stream, start sending files immediately
if (requests.isEmpty())
startStreamingFiles();
}

3. Streaming发送阶段

(a)这个阶段的是有每一个调用startStreamingFiles()方法的节点开启的(发送者,但是注意一个StreamSession的每一端都有可能是某些文件的发送者),然后发送一个FileMassage为每个StreamTransferTask中的每一个文件。每一条FileMessage包含一个FileMessageHeader,代表哪个文件要进来,然后开始streaming那个文件的内容。(StreamWriter 在FlieMessage.serialize())当一个文件完全被发送了,会调用那个文件的fileSent()方法。当一个StreamTransferTask所有的文件都完成了,调用StreamTransforTask.complete()。任务标记为完成。

(b)接收这端,一个SSTable会为进来的文件写SSTable(StreamReader 在FileMessage.deserialize())一旦FileMessage完全被接收了,文件会被标记为完成(received())。当所有的文件都接收了,sstables被加到了文件系统中(2nd index也已经建立了,StreamReceiveTask.complete()),任务被标记为完成(taskCompleted())

(c) 如果在某个特定的文件的streaming过程中,在接收过程中发生了I/O错误。(FileMessage.deserialize)。节点会重新retry这个文件,(retry次数取决于DatabaseDescriptor.getMaxStreamingRetries())通过发送一个RetryMessage给发送者。一旦接收到RetryMessage,发送者会为那个文件创建一个新的FileMessage

(d)当一个会话中所有的transfer和receive task都完成了,就进行到完成阶段

4. 完成阶段

当一个节点完成了所有的transfer 和receive task,就进入到了completion 阶段(maybeCompleted())。如果它已经从另外一边接收到一个CompleteMessage,这个会话就关闭了,否则这个节点切换到WAIT_COMPLETE 状态,并且发送一个CompleteMessage给另外一端。