ThreadPooledClient

Note:

该类已不推荐使用，原因如下：

  • 如果回调处理时间大于数据间隔，请检查回调函数是否有耗时操作（例如写文件、网络传输等）。

  • 如果无法及时处理数据，建议优先尝试下列方案：

    • 建议将您的数据处理过程调整到回调函数之外进行，例如您可以在回调中将数据保存为自己的数据类型，然后创建多个业务线程进行处理。

    • 使用 PollingClient，创建多个线程进行读取与处理。

ThreadPooledClient 支持用户创建指定数量的多个线程。每当发布端的流数据到达订阅端时，如果存在空闲的线程，则从空闲的线程中选择一个来调用回调函数。故当数据到达的间隔时间小于回调函数的处理时间时，ThreadPooledClient 比 ThreadedClient 有优势。本小节将从构造函数、订阅和取消订阅对 ThreadPooledClient 进行介绍，并展示一个使用示例。

构造函数

ThreadPooledClient(int listeningPort = 0, int threadCount = 3)
  • listeningPort：表示多线程客户端节点的订阅端口号。
  • threadCount：表示线程池创建的线程数量。
Note:
2.00.9 及之后版本的 server 发布端通过订阅端的请求连接推送数据，不再需要接收端指定端口，参数 listeningPort 填 0 或者不填即可

ThreadPooledClient 的新构造函数如下：

ThreadPooledClient(StreamingClientConfig config, int threadCount = 3);

threadCount 表示线程池创建的线程数量。

StreamingClientConfig 为流订阅 API 的配置，当前定义如下：

enum class TransportationProtocol {
    TCP, UDP,
};

enum class SubscribeState {
    Connected, // 连接成功，不代表有数据传来
    Disconnected, // 连接断开
    Resubscribing, // 正在重试
};
struct SubscribeInfo {
    std::string hostName;
    int port;
    std::string tableName;
    std::string actionName;
};
using SubscribeCallbackT = std::function<bool(const SubscribeState state, const SubscribeInfo &info)>;

struct StreamingClientConfig {
    TransportationProtocol protocol{TransportationProtocol::TCP};
    SubscribeCallbackT callback;
    int netTimeout{30000};
};

protocol 为传输协议 ：

  • 为 TCP （默认值）时，与 3.00.2 及之前版本功能相同。

  • 为 UDP 时，通过 Aeron 库与 DolphinDB Server 进行 UDP 组播通信。

Note:

使用 UDP 协议时：

  • 当前 UDP 通信方式仅支持订阅接口1，且不支持 resub, filter, backupSites, resubscribeInterval, subOnce 等参数。

  • C++ API 仅支持在 Linux 下编译和运行。

  • C++ API 在编译时依赖于 Aeron 库，但对 Aeron 库的版本没有限制。

  • C++ API 以嵌入式 Aeron Media Driver 的方式运行，客户无需单独启动 Aeron 线程。

  • 当 C++ API 使用 UDP 进行流订阅时，会在/dev/shm目录下创建一个以 dolphindb_udp_<pid> 命名的文件夹供 Aeron 使用，订阅结束后会自动删除。如客户程序异常退出，请手动删除相关文件夹。

  • Server 须为 Linux 3.00.0 以上版本。

  • Server 每次发布的信息大小上限为 2MB。如果订阅未收到数据，请检查是否超过了该限制。如有需要，可以调整配置项 maxMsgNumPerBlock 以解决此问题。

callback 为状态发生变化时的回调函数，请参考以下示例：

int main(){
    auto handler = [](const Message &msg) {
        std::cout << "this is a message" << std::endl;
    };
    auto stateCallback = [](const SubscribeState state, const SubscribeInfo &info) {
        std::cout << "state: " << static_cast<int>(state)
                  << ", table: " << info.tableName << std::endl;
        return true;
    };
    StreamingClientConfig config {
        .callback = stateCallback,
    };
    ThreadPooledClient client(config);
    client.subscribe("localhost", 8848, handler, "shared_stream_table", "action1", -1, true, nullptr, false, false, "admin", "123456");
    std::this_thread::sleep_for(100s);
    client.unsubscribe("localhost", 8848, "shared_stream_table", "action1");
    return 0;
}
Note:
当前回调函数的返回值（bool）尚未生效，请返回 true。后续将支持通过返回 false 来中断订阅。

netTimeout 控制流订阅的网络异常检测和建连超时，单位为毫秒，默认值为 30000。若设置为 0，保活检测使用 30 秒的默认值，且不额外设置 TCP 建连超时。

订阅

vector<ThreadSP> ThreadPooledClient::subscribe(
    string host,
    int port,
    const MessageHandler &handler,
    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 resubscribeInterval = 100,
    bool subOnce = false,
    int resubscribeTimeout = 0
)

参数说明

参考：ThreadedClient 参数。

返回类型

返回一个指针向量，每个指针指向循环调用handler的线程。这些线程在此topic被取消订阅后会退出。

取消订阅

bool unsubscribe(
    string host,
    int port,
    string tableName,
    string actionName
)

参数值需要与订阅时的填入参数值相同。

取消订阅成功时返回 true，否则返回 false。

使用示例

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

#include <iostream>
#include "Streaming.h"
using namespace dolphindb;

int main(int argc, const char **argv)
{
    auto handler = [](Message msg){
        std::cout << msg->getString() << std::endl;
    };
    ThreadPooledClient client(0, 10);
    auto t = client.subscribe("127.0.0.1", 8848, handler, "shared_stream_table", "action2", 0);
    sleep(10);
    client.unsubscribe("127.0.0.1", 8848, "shared_stream_table", "action2");
    return 0;
}

示例：将 TCP 流订阅的网络超时配置为 10 秒。

#include "Streaming.h"

using namespace dolphindb;

StreamingClientConfig config;
config.netTimeout = 10'000; // 单位：毫秒

ThreadPooledClient client(config, 3);