流数据高可用

在启用流数据高可用功能前，需要先在多个服务器上部署 DolphinDB 集群（目前高可用功能仅在集群中支持，在单节点模式中不支持)，然后根据流数据教程配置相关参数，再创建高可用流数据表。

高可用相关的配置项主要的有：

• streamingHAMode：指定高可用功能采用的协议，当设置为 Raft 时，表示流数据高可用功能采用了 Raft 协议。

• streamingRaftGroups：配置 Raft 组。格式为:"groupId:node_name1:node_name2:node_name3"，其中 groupId 表示 Raft 组号，为大于等于 2 的整数，node_name 为数据节点或计算节点的节点名字，须配置至少 3 个，中间用冒号分隔。系统支持配置多组，用逗号分隔。建议每组的数据节点分别位于不同的服务器上，以防止一台服务器宕机，组内节点都不可用。

• persistenceDir：流数据表的保存路径。高可用流数据表数据要保存到磁盘上，必须配置 persistenceDir。在集群模式中，应当为每个节点配置不同的 persistenceDir。

• streamingHADir：流数据 Raft 日志文件的存储目录。Raft 日志包括注册表、删除表、append（追加）数据等信息。为保证断电后不丢失数据，DolphinDB 会调用 fsync 把日志刷到磁盘上。一般 HDD 硬盘上 fsync 性能比较差，例如有的 HDD 硬盘刷一次要 10-30 毫秒，为了保证性能，建议这个目录配置在 SSD 硬盘上。如果不配置，系统默认值为 <HomeDir>/log/streamLog。如果在同一个服务器上部署了多个数据节点，每个节点应当配置不同的 streamingHADir。

• streamingHAPurgeInterval：Raft 日志垃圾回收周期。在系统运行时，节点中的 Raft 日志信息若不断增长，会影响节点恢复时状态回放的效率。Raft 采用 Checkpoint（检查点）的方式来截断日志。做 Checkpoint 有一些需要注意的性能点：不要做得太频繁，否则消耗磁盘带宽；也不要做得太不频繁，否则一旦节点重启需要回放大量日志，影响可用性。系统默认值300，单位为秒。

用户需要在集群 cluster.cfg 配置文件中配置上述参数。以下例子配置了 2 个 Raft 组，组 2 包含 3 个数据节点，分别为 DataNode1、DataNode3 和 DataNode5。组 3 包含 3 个数据节点，分别为 DataNode2、DataNode4 和 DataNode6。

streamingHAMode=raft
streamingRaftGroups=2:DataNode1:DataNode3:DataNode5,3:DataNode2:DataNode4:DataNode6
persistenceDir=/ssd/dolphindb/persistenceDir/<ALIAS>
streamingHADir=/ssd/dolphindb/streamLog/<ALIAS>

注意：

• streamingRaftGroups 配置的节点名字，必须和配置文件 cluster.nodes 中配置的数据节点或计算节点名字一致

• persistenceDir 和 streamingHADir 配置的路径必须根据实际服务器的目录情况配置

可使用getStreamingRaftGroups函数以获取与验证用户在集群配置的 Raft 组。

此函数只能获取当前数据节点上配置的 Raft 组。例如在上述例子中配置的 DataNode1上运行getStreamingRaftGroups()，结果如下：

在集群中配置 Raft 组后，每个 Raft 组会自动选出 Leader，组中其他节点为 Follower，由 Leader 为客户端提供流数据订阅服务。当 Leader 出现故障不可用时，系统会自动选举出新的 Leader 来提供流数据服务。获取指定 raft 组的 Leader，可以调用函数getStreamingLeader。函数语法如下：

getStreamingLeader(groupId)

其中参数 groupId 是 Raft 组的编号。

需要注意的是，要获取指定 Raft 组的 Leader，只能在该组配置的数据节点上运行该函数。例如要获取上述例子配置的 Raft 组 2 的 Leader，可以在 DataNode1、DataNode3 或 DataNode5 上运行下列代码：

getStreamingLeader(2)

若要在其他节点上获取组 2 的 Leader，需要先获取配置了该组的任意节点，然后使用rpc函数到该节点获取 Leader。例子如下：

t = exec top 1 id,node from pnodeRun(getStreamingRaftGroups) where id =2
leader = rpc(t.node[0],getStreamingLeader,t.id[0])

创建高可用流数据表使用函数haStreamTable。语法如下：

haStreamTable(raftGroup, table, tableName, cacheLimit, [keyColumn], [retentionMinutes=1440])

参数 raftGroup 是一个大于 1 的整数，表示 Raft 组的 ID，它必须是通过 streamingRaftGroups 配置的 Raft 组中的一个成员。

可选参数 keyColumn 是一个字符串，表示主键。设置了 keyColumn 后，系统能根据 keyColumn 自动过滤重复数据。比如从设备采集的流数据，这个 keyColumn 产生规则可以是设备编号+每个设备的消息唯一号；从股市获取的股票信息，keyColumn 可以是股票编码+每个股票消息的唯一号，这样从多个采集源采集入库或重复入库，都能保证数据不重复。

一个 Raft 组可以包含多个高可用流数据表。这里要注意以下几点：

• 高可用流数据表在日志中持久化了表结构信息，重启后不需要重新建表。普通流数据表在节点重启后，需要重新创建流数据表。

• 客户端只需订阅 Raft 组中任意一个数据节点上的高可用流数据表，并启用订阅的自动重连功能。Leader 上的高可用流数据表会向客户端发布数据。如果 Raft 组中的 Leader 宕机，系统会选举出新的 Leader 继续发布数据，客户端会自动切换订阅到新的 Leader 上的高可用流数据表。

• 在启动高可用流数据功能后，普通流数据表和高可用流数据表可共存于同一个数据节点上。普通流数据表与高可用流数据表也可互相兼容，但从普通流数据表转化为同名高可用流数据表时，要注意清除之前普通流数据表的数据和日志（使用dropStreamTable命令进行删除）。与普通流数据表一样，查看高可用流数据表的元数据信息可使用getPersistenceMeta函数。

下面的例子在 Raft 组2上创建了一个高可用流数据表 haDevSt，其中 cacheLimit 是100000，keyColumn 是 id。

tmp = table(1:0, ["id", "source_address", "source_port", "destination_address", "destination_port"], [STRING,INT,INT,INT,INT])
haStreamTable(2, tmp, "haDevSt", 100000, "id")

删除高可用流数据表可调用命令dropStreamTable。语法如下：

dropStreamTable(tableName)

用户需要在取消所有订阅客户端后才能删除流数据表。

生产者向流数据表写入数据，只能向 Leader 节点写入，向 Follower 节点写数据会抛出 NotLeader 异常。目前 JAVA API、C++ API 和 Python API 已支持流数据生产者的高可用，能自动捕获这个异常，并获取异常中携带的新 Leader 的节点信息，然后向新 Leader 继续写入数据。在 API 中，要求在 connect 函数中指定 highAvailabilitySites 参数。highAvailabilitySites 是一个 String 类型的 Array，每个 String 都是 host:port 格式。这样在 Leader 切换时，就能自动切换到 highAvailabilitySites 中的另一个节点，即新的 Leader。JAVA API 代码举例如下：

String[] haSites = new String[]{"192.168.1.11:19151","192.168.1.12:19153","192.168.1.13:19155"};
conn.connect(host,port,"admin", "123456", initScript, true,haSites);

Python API 代码举例如下：

conn.connect(host, port, "admin","123456", initScript, True, ["192.168.1.2:9921", "192.168.1.3:9921","192.168.1.4:9921"])

客户端只需订阅 Raft 组中任意一个数据节点上的高可用流数据表，并启用订阅的自动重连功能（即把subscribeTable的 reconnect 参数设置为 true）。Leader 上的高可用流数据表会向订阅端发布数据。如果 Raft 组中的 Leader 宕机，系统会选举出新的 Leader 继续发布数据，客户端会自动切换订阅到新 Leader 上的高可用流数据表。

同时针对高可用流数据表，subscribeTable提供了持久化保存最新一条已经处理过的订阅数据的偏移量的功能，以避免在重订阅时部分未处理的数据丢失，通过设置参数 persistOffset = true 可开启该功能。偏移量可以通过 getTopicProcessedOffset 函数获取。

下面的例子在 Raft 组2上创建了一个高可用流数据表 haDevSt。然后在集群中的某个节点上执行subscribeTable，向 Leader 节点订阅 haDevSt。subscribeTable的参数 reconnect 和 persistOffset 设为 true。

t1=table(1:0, ["id","source_address","source_port","destination_address","destination_port"], [STRING,INT,INT,INT,INT] )
haStreamTable(2,t1,"haDevSt",100000,"id")
​
t=select id,node from pnodeRun(getStreamingRaftGroups) where id =2
def msgProc(msg){ //todo        
}
subscribeTable(server=t.node[0], tableName="haDevSt",  handler=msgProc{}, reconnect=true, persistOffset=true)

当流数据表启用高可用时，API 客户端连接必须启用重连功能。目前 JAVA API、C++ API 和 Python API 已支持流数据消费者的高可用。其实现原理如下：

• 客户端订阅时，保存 raftGroup 中所有节点的 site（节点信息）。

• Leader 变为 Follower 时，Broker 会主动关闭这个组中所有的流数据表对应的订阅，所以客户端就会发起重连。

• 客户端重连时如果连接到了 Follower 上，会收到 Not Leader Exception 的异常，这个异常中还携带了目前新 Leader 的信息（如果 Leader 尚未被选出，Follower 就返回自身节点信息）。客户端收到这个异常之后，等待一段时间后再尝试连接新的 Leader。

• 客户端重连一个节点多次都失败时，则开始尝试连接其他配置的节点。