PollingClient
PollingClient 在订阅数据时,会返回一个消息队列,用户可以从该消息队列中获取和处理数据。本小节将从构造函数、订阅和取消订阅对 PollingClient 进行介绍,并展示一个使用示例。
构造方法
PollingClient(int listeningPort = 0)
其中,listeningPort
表示单线程客户端的订阅端口号。
Note: 2.00.9 及之后版本的 server 发布端通过订阅端的请求连接推送数据,不再需要接收端指定端口,参数
listeningPort 填 0 或者不填即可。
订阅
MessageQueueSP subscribe(
string host,
int port,
string tableName,
string actionName,
int64_t offset = -1,
bool resub = true,
const VectorSP &filter = nullptr,
bool msgAsTable = false,
bool allowExists = false,
string userName="",
string password="",
const StreamDeserializerSP &blobDeserializer = nullptr,
const std::vector<std::string>& backupSites = std::vector<std::string>(),
int resubTimeout = 100,
bool subOnce = false
)
参数
host
:表示发布端节点的主机名。port
:表示发布端节点的端口号。tableName
:表示发布端上共享流数据表的名称。actionName
:表示订阅任务的名称。传入值可以包含字母、数字和下划线。offset
:表示订阅任务开始后的第一条消息所在的位置。消息是流数据表中的行。如果没有指定 offset、或其为负数、或超过了流数据表的记录行数,则订阅将会从流数据表的当前行开始。offset 与流数据表创建时的第一行对应。如果某些行因为内存限制被删除,在决定订阅开始的位置时,这些行仍然考虑在内。resub
:表示订阅中断后,API 是否会自动重订阅。filter
:一个向量,表示过滤条件。流数据表过滤列在 filter 中的数据才会发布到订阅端,不在 filter 中的数据不会发布。msgAsTable
:只有设置了 batchSize 参数,该参数才会生效。若其设置为 true,则订阅的数据会转换为 Table;若其设置为 false,则订阅的数据会转换成 AnyVector。allowExists
:若该参数设置为 true,则支持对同一个订阅流表使用多个 handler 处理数据;若其设置为 False,则不支持同一个订阅流表使用多个 handler 处理数据。userName
:表示 API 所连接服务器的登录用户名。password
:表示 API 所连接服务器的登录密码。blobDeserializer
:表示订阅的异构流表对应的反序列化器。backupSites
: 字符串列表,可选,表示备用的发布端节点列表,由节点 ip 和端口号组成,例如 [“192.168.0.1:8848“, “192.168.0.2:8849“]。- 指定backupSites,表示启动主备节点切换。如果发生节点切换(例如连接断开),会在可用节点列表中不断轮询订阅,直至订阅成功。
- 若配置该参数,用户需保证主节点(由 host 和 port 参数指定)和备用节点上存在相同结构、相同数据的同名流数据表。否则,可能出现订阅的数据不符合预期。
- 若订阅的是高可用流表,重连时将从 backupSites 指定的节点列表中选择节点。
- 取消订阅需使用主节点 ip 和端口号进行取消。
resubTimeout
:一个整数,表示重订阅距离断线最小时间间隔,默认值为 100 毫秒。subOnce
:布尔值,False表示在节点切换时,不去尝试之前成功连接过的节点。默认值为 False。
返回类型
该函数返回指向消息队列的指针。
取消订阅
void unsubscribe(
string host,
int port,
string tableName,
string actionName
)
Note: 参数值需要与订阅时填入参数值相同。
使用示例
DolphinDB 脚本:新建一个Stream表,然后共享该表,名为shared_stream_table
。
rt = streamTable(`XOM`GS`AAPL as id, 102.1 33.4 73.6 as x)
share rt as shared_stream_table
C++代码:
本例使用2.00.10版本的DolphinDB server,在订阅一秒后取消订阅
int main(int argc, const char **argv)
{
PollingClient client(0);
MessageQueueSP queue = client.subscribe("127.0.0.1", 8848, "shared_stream_table", "action3", 0);
ThreadSP t = new Thread(new Executor([&client](){
Util::sleep(1000);
client.unsubscribe("127.0.0.1", 8848, "shared_stream_table", "action3");
}));
t->start();
Message msg;
while (queue->poll(msg, 1000) && !msg.isNull()) {
std::cout << msg->getString() << std::endl;
}
}