发布与订阅

发布-订阅模型

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)