发布与订阅
发布-订阅模型
DolphinDB 采用了经典的发布-订阅(Pub-Sub)通信模型,通过消息队列实现流数据的发布与订阅,将流数据生产者(发布者)与消费者(订阅者)解耦。这种模式促进了系统内不同组件之间的通信,提高可伸缩性,改善发送者的响应能力。
发布数据:发布端在每个节点上维护一个发布队列。当新的流数据注入到该节点的流数据发布表时,DolphinDB 会将这些数据发送到相应的消息发布队列,再由发布线程将数据发布到各个订阅端的消费队列。
订阅数据:每个订阅线程对应一个消费队列。订阅成功提交后,每当有新数据写入流数据发布表时,DolphinDB 会主动通知所有订阅方,消费线程从消费队列中获取数据进行增量处理。
订阅消费
DolphinDB 的数据节点和计算节点可以充当消费者的角色,通过 subscribeTable 函数可以订阅流数据表,以下代码是一个简单的流数据实时清洗转换写入数据库的例子:
step1:创建存储数据库和分区表
drop database if exists "dfs://minuteBar" create database "dfs://minuteBar" partitioned by VALUE(2020.01.01..2021.01.01) engine='OLAP' create table "dfs://minuteBar"."minuteBar"( securityid SYMBOL tradetime TIMESTAMP price DOUBLE volume INT amount DOUBLE ) partitioned by tradetime
step2:创建发布流数据表
colNames = ["code", "tradetime", "price", "volume"] colTypes = [SYMBOL, TIMESTAMP, DOUBLE, INT] share(table=streamTable(1:0, colNames, colTypes), sharedName="pubTable")
step3:定义数据清洗和转换的回调函数
def dataETL(msg){ result = select *, price*volume as amount from msg where price>=10.6 loadTable("dfs://minuteBar", "minuteBar").append!(result) }
step4:订阅发布流数据表
subscribeTable(tableName="pubTable", actionName="dataETL", offset=-1, handler=dataETL, msgAsTable=true, batchSize=2000, throttle=0.01)
-
offset=-1
表示从流数据表的当前最新行开始消费 -
handler=dataETL
表示处理订阅的回调函数为自定义函数dataETL
-
msgAsTable=true
表示回调函数处理的 msg 对象数据格式为表 -
batchSize=2000
,throttle=0.01
表示上游流表一旦有数据更新,满足以下任一条件触发回调函数执行:(1)待处理数据行数大于等于 2000 条;(2) 时间间隔大于等于 0.01 秒。设置该参数的目的是为了批量处理数据以提高数据处理吞吐量
返回如下信息说明订阅成功:
localhost:8200:local8200/pubTable/dataETL
step5:模拟单条数据写入
tableInsert(pubTable, "000001SZ", 2023.12.15T09:30:00.000, 10.5, 200) tableInsert(pubTable, "000001SZ", 2023.12.15T09:31:00.000, 10.6, 1000) tableInsert(pubTable, "000001SZ", 2023.12.15T09:32:00.000, 10.7, 600) tableInsert(pubTable, "000001SZ", 2023.12.15T09:33:00.000, 10.8, 800) tableInsert(pubTable, "000001SZ", 2023.12.15T09:34:00.000, 10.7, 500) tableInsert(pubTable, "000001SZ", 2023.12.15T09:35:00.000, 10.6, 1200)
step6:查询分区表内数据
res = select * from loadTable("dfs://minuteBar", "minuteBar") where date(tradetime)=2023.12.15
返回结果 res:
从返回结果可以看到,price 小于 10.6 的记录被清洗掉,且增加了 amount 列(price*volume):
step7:取消订阅
unsubscribeTable(tableName="pubTable", actionName="dataETL")
step8:删除发布流数据表
注意:删除流数据表前,必须先把其所有订阅取消掉。
dropStreamTable(tableName="pubTable")
step9:删除测试数据库
drop database if exists "dfs://minuteBar"
自动重新订阅
DolphinDB 的流数据订阅提供了自动重连的功能。当 subscribe
函数的 reconnect 参数设为 true
时,消费者会记录流数据的 offset,连接中断时订阅端会自动从 offset 开始重新订阅。上述例子中的订阅可以这样设置自动重新订阅功能:
subscribeTable(tableName="pubTable", actionName="dataETL", offset=-1, handler=dataETL, msgAsTable=true, batchSize=2000, throttle=0.01, reconnect=true)
过滤发布
发布端可以设置流数据表的过滤列,消费者在执行 subscribe
函数订阅该流数据表的时候,可以指定过滤条件
,设置过滤发布的步骤如下:
-
首先需要在发布端执行 setStreamTableFilterColumn 函数指定流数据表的过滤列
-
然后指定
subscribe
函数的 filter 参数 -
filter 是一个向量。流数据表过滤列只有在 filter 中的数据才会发布到订阅端,不在 filter 中的数据不会发布。目前仅支持对一个列的数据进行过滤
-
在发布端进行数据过滤可以减少网络流量
例如,发布端上的流数据表 st 只向该消费者发布 code 为 000001SZ 和 000002SZ 的数据:
step1:创建发布流数据表并设置过滤列
colNames = ["code", "tradetime", "price", "volume"] colTypes = [SYMBOL, TIMESTAMP, DOUBLE, INT] share(table=streamTable(1:0, colNames, colTypes), sharedName="pubTable") go setStreamTableFilterColumn(streamTable=pubTable, columnName="code")
step2:创建存储过滤后数据的共享普通内存表
colNames = ["code", "tradetime", "price", "volume"] colTypes = [SYMBOL, TIMESTAMP, DOUBLE, INT] share(table=streamTable(1:0, colNames, colTypes), sharedName=`result)
step3:订阅发布流数据表并设置过滤条件
filter=symbol(`000001SZ`000002SZ) subscribeTable(tableName="pubTable", actionName="dataFilter", offset=-1, handler=result, msgAsTable=true, batchSize=2000, throttle=0.01, reconnect=true, filter=filter)
step4:模拟单条数据写入
tableInsert(pubTable, "000001SZ", 2023.12.15T09:30:00.000, 10.5, 200) tableInsert(pubTable, "000002SZ", 2023.12.15T09:31:00.000, 10.6, 1000) tableInsert(pubTable, "000003SZ", 2023.12.15T09:32:00.000, 10.7, 600) tableInsert(pubTable, "000004SZ", 2023.12.15T09:33:00.000, 10.8, 800) tableInsert(pubTable, "000005SZ", 2023.12.15T09:34:00.000, 10.7, 500) tableInsert(pubTable, "000006SZ", 2023.12.15T09:35:00.000, 10.6, 1200)
查看发布端的流数据表 pubTable 和订阅端的过滤后的数据表 result:
step5:取消订阅
unsubscribeTable(tableName="pubTable", actionName="dataFilter")
step6:删除发布流数据表
注意:删除流数据表前,必须先把其所有订阅取消掉。
dropStreamTable(tableName="pubTable")
step7:删除共享的普通内存表
undef(obj="result", objType=SHARED)