订阅异构流表

DolphinDB server 自 1.30.17 及 2.00.5 版本开始,支持通过 replay 函数将多个结构不同的流数据表,回放(序列化)到一个流表里,这个流表被称为异构流表。

在 C++ API 中,需要在回调函数中通过 symbol 值来区分该条消息具体来自哪个表。

Note: 在订阅异构流表时,如果表中某一列是 ArrayVector,那么在回调函数的参数 Message 中,该列对应的元素也是一个 ArrayVector,长度为 1。如果订阅的是普通流表,表中某一列是 ArrayVector,则在回调参数 Message 中,该列对应的元素是一个 Vector。

创建反序列化器

如果要通过 C++ API 来订阅异构流表,则需要定义一个异构流表反序列化器,也就是 StreamDeserializer 类的一个对象。创建该类对象的几种方式如下:

  • 通过symbol与表的对应关系来创建,其中,表由一个pair<string, string>来表示,前面的字符串表示库名,后面的字符串表示表名,如果是内存表, 库名为空字符串。第二个参数pconn目前可以忽略。
    StreamDeserializer(const unordered_map<string, pair<string, string>> &sym2tableName, DBConnection *pconn = nullptr);
  • 通过symbol与表的Schema的对应关系来创建。
    StreamDeserializer(const unordered_map<string, DictionarySP> &sym2schema);
  • 通过symbol与表中每一列的类型的关系来创建。

示例代码

本例使用2.00.10版本的 DolphinDB Server

DolphinDB 脚本,创建两个结构不同的表,将两个表的内容回放到一个异构流表中:

st2 = streamTable(100:0, `timestampv`sym`blob`price1,[TIMESTAMP,SYMBOL,BLOB,DOUBLE])
share st2 as SDoutTables

n = 5;
table1 = table(100:0, `datetimev`timestampv`sym`price1`price2, [DATETIME, TIMESTAMP, SYMBOL, DOUBLE, DOUBLE]);
table2 = table(100:0, `datetimev`timestampv`sym`price1, [DATETIME, TIMESTAMP, SYMBOL, DOUBLE]);
share table1 as table1_SDPT
share table2 as table2_SDPT
tableInsert(table1_SDPT, 2012.01.01T01:21:23 + 1..n, 2018.12.01T01:21:23.000 + 1..n, take(`a`b`c,n), rand(100,n)+rand(1.0, n), rand(100,n)+rand(1.0, n));
tableInsert(table2_SDPT, 2012.01.01T01:21:23 + 1..n, 2018.12.01T01:21:23.000 + 1..n, take(`a`b`c,n), rand(100,n)+rand(1.0, n));
d = dict(['msg1','msg2'], [table1_SDPT, table2_SDPT]);
replay(inputTables=d, outputTables=`SDoutTables, dateColumn=`timestampv, timeColumn=`timestampv)

C++ 代码 1(用第一个构造函数来创建反序列化器):

int main(int argc, const char **argv)
{
    std::unordered_map<std::string, std::pair<std::string, std::string>> sym2table{{"msg1", {"", "table1_SDPT"}}, {"msg2", {"", "table2_SDPT"}}};
    StreamDeserializerSP sdsp = new StreamDeserializer(sym2table);
    auto onehandler = [&](Message msg)
    {
        std::cout << "symbol: " << msg.getSymbol() << " content: " << msg->getString() << std::endl;
    };
    ThreadedClient threadedClient(0);
    threadedClient.subscribe("127.0.0.1", 8848, onehandler, "SDoutTables", "test_SD", 0, true, nullptr, false, false, "admin", "123456", sdsp);
    Util::sleep(1000);
    threadedClient.unsubscribe("127.0.0.1", 8848, "SDoutTables", "test_SD");
}

C++ 代码 2(用第二个构造函数来创建反序列化器):

int main(int argc, const char **argv)
{
    DBConnection conn;
    conn.connect("127.0.0.1", 8848, "admin", "123456");
    DictionarySP t1s = conn.run("schema(table1_SDPT)");
    DictionarySP t2s = conn.run("schema(table2_SDPT)");
    unordered_map<string, DictionarySP> sym2schema;
    sym2schema["msg1"] = t1s;
    sym2schema["msg2"] = t2s;
    StreamDeserializerSP sdsp = new StreamDeserializer(sym2schema);
    auto onehandler = [&](Message msg)
    {
        std::cout << "symbol: " << msg.getSymbol() << " content: " << msg->getString() << std::endl;
    };
    ThreadedClient threadedClient(0);
    threadedClient.subscribe("127.0.0.1", 8848, onehandler, "SDoutTables", "test_SD", 0, true, nullptr, false, false, "admin", "123456", sdsp);
    Util::sleep(1000);
    threadedClient.unsubscribe("127.0.0.1", 8848, "SDoutTables", "test_SD");
}

C++ 代码 3(用第三个构造函数来创建反序列化器):

int main(int argc, const char **argv)
{
    unordered_map<string, std::vector<DATA_TYPE>> sym2cols{{"msg1", {DT_DATETIME, DT_TIMESTAMP, DT_SYMBOL, DT_DOUBLE, DT_DOUBLE}}, {"msg2", {DT_DATETIME, DT_TIMESTAMP, DT_SYMBOL, DT_DOUBLE}}};
    StreamDeserializerSP sdsp = new StreamDeserializer(sym2cols);
    auto onehandler = [&](Message msg)
    {
        std::cout << "symbol: " << msg.getSymbol() << " content: " << msg->getString() << std::endl;
    };
    ThreadedClient threadedClient(0);
    threadedClient.subscribe("127.0.0.1", 8848, onehandler, "SDoutTables", "test_SD", 0, true, nullptr, false, false, "admin", "123456", sdsp);
    Util::sleep(1000);
    threadedClient.unsubscribe("127.0.0.1", 8848, "SDoutTables", "test_SD");
}