unsubscribeTable

语法

unsubscribeTable([server], tableName, [actionName], [removeOffset=true], [raftGroup])

详情

在信息订阅端的节点执行，以停止向信息发布者订阅数据。在调用该命令时，会删除流计算执行线程的队列中未处理的消息。

参数

server 是表示服务器的别名或与流数据表所在服务器创建的 xdb 连接的字符串。

tableName 是表示取消订阅的流数据表名称的字符串。取消跨集群订阅 Orca 流表时，格式为fqn@cluster_identifier，例如 demo.orca_table.trades@cluster_BeiJing。

actionName 是表示句柄名称的字符串。它可以包含字母、数字和下划线。如果创建订阅时指定了 actionName，取消订阅时必须指定 actionName。

removeOffset 是一个布尔值，表示是否删除持久化保存的最新一条已经处理订阅数据的偏移量（在 subscribeTable 函数中通过指定 persistOffset 参数为 true 获得）。

raftGroup 是在 subscribeTable 中设定的 raft 组的 ID，用于取消相应的订阅端高可用。

注：

unsubscribeTable 函数如果指定了 raftGroup ，则只能在 leader 上执行。

例子

以下示例演示订阅与取消订阅。例1 在同一个节点上完成发布与订阅；例2 中发布节点与订阅节点不在同一节点。

例1：在本地节点创建共享流数据表 trades，并创建表 trades2 来保存订阅的数据。

t = streamTable(100:0, `date`time`sym`qty`price`exch, [DATE,TIME,SYMBOL,INT,DOUBLE,SYMBOL])
share t as trades
t2 = streamTable(100:0, `date`time`sym`qty`price`exch, [DATE,TIME,SYMBOL,INT,DOUBLE,SYMBOL])
share t2 as trades2
t = NULL
t2 = NULL

订阅表 trades，并向 trades 写入一条数据。trades2 中会同步得到该数据。

subscribeTable(tableName="trades", actionName="sub1", handler=trades2)
insert into trades values(2024.01.01, 09:30:00.000, `AAPL, 100, 120.5, `NASDAQ)
sleep(100)
select count(*) from trades2
// output: 1

取消订阅表 trades 的数据：

unsubscribeTable(tableName="trades", actionName="sub1")

例2：发布节点与订阅节点不在同一节点。该例需要在两个节点上执行，发布节点的发布/订阅端口为 8902；节点不能通过 xdb 连接自身。

在发布节点发布一张表 trades。

t=streamTable(100:0,`date`time`sym`qty`price`exch,[DATE,TIME,SYMBOL,INT,DOUBLE,SYMBOL])
share t as trades
t=NULL

在订阅节点上创建表 trades2 来保存发布节点中的表 trades 的数据。

t=streamTable(100:0,`date`time`sym`qty`price`exch,[DATE,TIME,SYMBOL,INT,DOUBLE,SYMBOL])
share t as trades2
t=NULL
h=xdb("localhost",8902)
subscribeTable(server=h, tableName="trades", actionName="sub1", handler=trades2);
// output: localhost:8902:node1/trades/sub1

在订阅节点上取消订阅表 trades 的数据：

unsubscribeTable(server=h, tableName="trades", actionName="sub1")