C# API

DolphinDB提供三种方式将数据新增到内存表：

• 通过insert into保存单条数据
• 通过tableInsert函数批量保存多条数据
• 通过tableInsert函数保存数据表

一般不建议通过append!函数保存数据，因为append!函数会返回表的schema，产生不必要的通信量。

下面分别介绍三种方式保存数据的实例，在例子中使用到的数据表有4个列，分别是string, int, timestamp, double类型，列名分别为cstring, cint, ctimestamp, cdouble。

t = table(10000:0,`cstring`cint`ctimestamp`cdouble,[STRING,INT,TIMESTAMP,DOUBLE])
share t as sharedTable

由于内存表是会话隔离的，所以该内存表只有当前会话可见。如果需要在其它会话中访问，需要通过share在会话间共享内存表。

分布式表是DolphinDB推荐在生产环境下使用的数据存储方式，它支持快照级别的事务隔离，保证数据一致性。分布式表支持多副本机制，既提供了数据容错能力，又能作为数据访问的负载均衡。

在C# API中，数据表保存为BasicTable对象。由于BasicTable是列式存储，所以若要在C# API中读取行数据需要先取出需要的列，再取出行。

以下例子中参数BasicTable的有4个列，列名分别为cstring, cint, ctimestamp, cdouble，数据类型分别是STRING, INT, TIMESTAMP, DOUBLE。

public void test_loop_basicTable(BasicTable table1)
{
      BasicStringVector stringv = (BasicStringVector) table1.getColumn("cstring");
      BasicIntVector intv = (BasicIntVector)table1.getColumn("cint");
      BasicTimestampVector timestampv = (BasicTimestampVector)table1.getColumn("ctimestamp");
      BasicDoubleVector doublev = (BasicDoubleVector)table1.getColumn("cdouble");
      for(int ri=0; ri<table1.rows(); ri++){
            Console.WriteLine(stringv.getString(ri));
            Console.WriteLine(intv.getInt(ri));
            DateTime timestamp = timestampv.getTimestamp(ri);
            Console.WriteLine(timestamp);
            Console.WriteLine(doublev.getDouble(ri));
      }
}

DolphinDB C# API 提供 MultithreadedTableWriter 类对象支持多线程的并发写入。目前，MultithreadedTableWriter 支持批量写入数据到内存表、分区表和维度表。

MultithreadedTableWriter 对象及主要方法介绍如下：

MultithreadedTableWriter(string hostName, int port, string userId, string password,string dbName, string tableName, bool useSSL, bool enableHighAvailability = false, string[] pHighAvailabilitySites = null,int batchSize = 1, float throttle = 0.01f, int threadCount = 5, string partitionCol = "", int[] pCompressMethods = null, Mode mode = Mode.M_Append, string[] pModeOption = null, Callback callbackHandler = null);

参数说明：

• hostName 字符串，表示所连接的服务器的地址
• port 整数，表示服务器端口。
• userId / password: 字符串，登录时的用户名和密码。
• dbPath 字符串，表示分布式数据库地址。内存表时该参数为空。请注意，1.30.17及以下版本 API，向内存表写入数据时，该参数需填写内存表表名。
• tableName 字符串，表示分布式表或内存表的表名。请注意，1.30.17及以下版本 API，向内存表写入数据时，该参数需为空。
• useSSL 布尔值，默认值为 False。表示是否启用加密通讯。
• enableHighAvailability 布尔值，默认为 False。若要开启 API 高可用，则需要指定 enableHighAvailability 参数为 True。
• pHighAvailabilitySites 列表类型，表示所有可用节点的 ip:port 构成的 list。
• batchSize 整数，表示批处理的消息的数量，默认值是 1，表示客户端写入数据后就立即发送给服务器。如果该参数大于 1，表示数据量达到 batchSize 时，客户端才会将数据发送给服务器。
• throttle 大于 0 的数，单位为秒。若客户端有数据写入，但数据量不足 batchSize，则等待 throttle的时间再发送数据。
• threadCount 整数，表示创建的工作线程数量，默认为 1，表示单线程。对于维度表，其值必须为1。
• partitionCol 字符串类型，默认为空，仅在 threadCount 大于1时起效。对于分区表，必须指定为分区字段名；如果是流表，必须指定为表的字段名；对于维度表，该参数不起效。
• pCompressMethods 列表类型，用于指定每一列采用的压缩传输方式，为空表示不压缩。每一列可选的压缩方式包括：
  • Vector_Fields.COMPRESS_LZ4: LZ4 压缩
  • Vector_Fields.COMPRESS_DELTA: DELTAOFDELTA 压缩
• mode 写入模式，用于指定 MultithreadedTableWriter 对象写入数据的方式，包括两种：
  • Mode.M_Append：表示以tableInsert 的方式向追加数据。
  • Mode.M_Upsert：表示以upsert!方式更新（或追加）数据。
• pModeOption：字符串数组，表示不同模式下的扩展选项，目前，仅当 mode 指定为 Mode.M_Upsert 时有效，表示由 upsert! 可选参数组成的字符串数组。如： new String[] { "ignoreNull=false", "keyColNames=`volume" }。
• callbackHandler：回调类（Callback），默认为空，表示不使用回调。开启回调后，将继承回调接口 Callback 并重载回调方法 writeCompletion。

若 MultithreadedTableWriter 指定了 callbackHandler，则注意以下事项：

• insert 的第一个参数必须是 STRING 类型，表示这一行数据的 id。
• getUnwrittenData 方法将不可用。

以下是 MultithreadedTableWriter 对象包含的函数方法介绍：

(1) insert

ErrorCodeInfo insert(params Object[] args)

函数说明：

插入单行数据。返回一个ErrorCodeInfo，表示是否插入成功。

参数说明：

• args：是变长参数，代表插入的一行数据。

返回值：返回一个ErrorCodeInfo对象，包含 errorCode 和 errorInfo，分别表示错误代码和错误信息。当 errorCode 不为空时，表示 MTW 写入失败，此时，errorInfo 会显示失败的详细信息。之后的版本中会对错误信息进行详细说明，给出错误信息的代码、错误原因及解决办法。

(2) getUnwrittenData

List<List<IEntity>> getUnwrittenData();

函数说明：

返回一个嵌套列表，表示未写入服务器的数据。

注意：该方法获取到数据资源后， MultithreadedTableWriter 将释放这些数据资源。

(3) insertUnwrittenData

ErrorCodeInfo insertUnwrittenData(List<List<IEntity>> data);

函数说明：

将数据插入数据表。返回值同 insert 方法。与 insert 方法的区别在于，insert 只能插入单行数据，而 insertUnwrittenData 可以同时插入多行数据。

参数说明：

• data：需要再次写入的数据。可以通过方法 getUnwrittenData 获取该对象。

Status getStatus()

函数说明：

获取 MultithreadedTableWriter 对象当前的运行状态。

参数说明：

• status：是MultithreadedTableWriter::Status 类，具有以下属性和方法

属性：

• isExiting：写入线程是否正在退出。
• errorCode：错误码。
• errorInfo：错误信息。
• sentRows：成功发送的总记录数。
• unsentRows：待发送的总记录数。
• sendFailedRows：发送失败的总记录数。
• threadStatus：写入线程状态列表。
  • threadId：线程 Id。
  • sentRows：该线程成功发送的记录数。
  • unsentRows：该线程待发送的记录数。
  • sendFailedRows：该线程发送失败的记录数。

(5) waitForThreadCompletion

waitForThreadCompletion()

函数说明：

调用此方法后，MTW 会进入等待状态，待后台工作线程全部完成后退出等待状态。

MultithreadedTableWriter 常规处理流程如下：

//创建连接，并初始化测试环境
string HOST = "192.168.1.38";
int PORT = 18848;
string USER = "admin";
string PASSWD = "123456";
DBConnection dBConnection = new DBConnection();
dBConnection.connect(HOST, PORT, USER, PASSWD);
Random random = new Random();
string script =
"dbName = 'dfs://valuedb3'" +
"if (exists(dbName))" +
"{" +
      "dropDatabase(dbName);" +
"}" +
"datetest = table(1000:0,`date`symbol`id,[DATE, SYMBOL, LONG]);" +
"db = database(directory= dbName, partitionType= HASH, partitionScheme=[INT, 10]);" +
"pt=db.createPartitionedTable(datetest,'pdatetest','id');";
dBConnection.run(script);

ErrorCodeInfo ret;
MultithreadedTableWriter.Status writeStatus;
MultithreadedTableWriter writer = new MultithreadedTableWriter(HOST, PORT, USER, PASSWD, "dfs://valuedb3", "pdatetest", false, false, null, 10000, 1, 5, "id", new int[] { Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_DELTA });
try
{
      //插入100行正确数据 （类型和列数都正确），MTW正常运行
      for (int i = 0; i < 100; ++i)
      {
            ret = writer.insert(new DateTime(2022, 3, 23), "AAAAAAAB", i);
            //此处不会执行到
            if (ret.errorCode != "")
            Console.WriteLine(string.Format("insert wrong format data: {0}\n", ret.ToString()));
      }
      Thread.Sleep(2000);

      //插入1行数据，类型不匹配，MTW立刻发现
      //MTW立刻返回错误信息
      ret = writer.insert(new DateTime(2022, 3, 23), random.Next() % 10000, random.Next() % 10000);
      if (ret.errorCode != "")
            Console.WriteLine("insert wrong format data: {0}\n", ret.ToString());
      /*
      insert wrong format data: code = A1 info = Failed to insert data. Cannot convert int to DT_SYMBOL
      */

      //如果发生了连接断开的情况，mtw将会在下一次向服务器写数据时发生失败。
      //先写一行数据，触发error
      ret = writer.insert(new DateTime(2022, 3, 23), "AAAAAAAB", 1);

      Thread.Sleep(1000);

      //再插入9行正确数据，MTW会因为工作线程终止而抛出异常，且该行数据不会被写入MTW
      //这里会直接抛出异常
      for (int i = 0; i < 9; ++i)
      {
            ret = writer.insert(new DateTime(2022, 3, 23), "AAAAAAAB", random.Next() % 10000);
      }

}
catch (Exception e)
{
      Console.WriteLine(e.Message);
      //Thread is exiting.
}
writer.waitForThreadCompletion();
writeStatus = writer.getStatus();
if (writeStatus.errorCode != "")
      //写入发生错误
      Console.WriteLine(string.Format("error in writing:\n {0}", writeStatus.ToString()));
Console.WriteLine(((BasicLong)dBConnection.run("exec count(*) from pt")).getLong());

/*
      error in writing: Cause of write failure
sentRows: 100
unsentRows: 3
sendFailedRows: 7
threadId : 3 sentRows : 20 unsentRows : 0 sendFailedRows : 5
threadId : 4 sentRows : 20 unsentRows : 2 sendFailedRows : 1
threadId : 5 sentRows : 20 unsentRows : 1 sendFailedRows : 0
threadId : 6 sentRows : 20 unsentRows : 0 sendFailedRows : 0
threadId : 7 sentRows : 20 unsentRows : 0 sendFailedRows : 1

100
      */

for (int i = 0; i < 30; ++i)
      Console.Write('-');
Console.WriteLine();

List<List<IEntity>> unwriterdata = new List<List<IEntity>>();
if (writeStatus.sentRows != 110)
{
      Console.WriteLine("error after write complete:" + writeStatus.errorInfo);
      unwriterdata = writer.getUnwrittenData();
      Console.WriteLine("unwriterdata {0}", unwriterdata.Count);

      //重新写入这些数据，原有的MTW因为异常退出已经不能用了，需要创建新的MTW
      MultithreadedTableWriter newWriter = new MultithreadedTableWriter(HOST, PORT, USER, PASSWD, "dfs://valuedb3", "pdatetest", false, false, null, 10000, 1, 5, "id", new int[] { Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_LZ4, Vector_Fields.COMPRESS_DELTA });
      try
      {
            //插入未写入的数据
            if (newWriter.insertUnwrittenData(unwriterdata).errorCode != "")
            {
            //等待写入完成后检查状态
            newWriter.waitForThreadCompletion();
            writeStatus = newWriter.getStatus();
            if (writeStatus.errorCode != "")
            {
                  System.Console.Out.WriteLine("error in write again:" + writeStatus.errorInfo);
            }
            }

      }
      finally
      {
            newWriter.waitForThreadCompletion();
            writeStatus = newWriter.getStatus();
            Console.WriteLine(string.Format("write again result:\n{0}", writeStatus));
      }
}
else
      Console.WriteLine("write complete : \n {0}", writeStatus.ToString());
//检查最后写入结果
Console.WriteLine(((BasicLong)dBConnection.run("exec count(*) from pt")).getLong());
/*
      unwriterdata 10
write again result:

isExiting: True
sentRows: 10
unsentRows: 0
sendFailedRows: 0
threadId : 8 sentRows : 5 unsentRows : 0 sendFailedRows : 0
threadId : 9 sentRows : 3 unsentRows : 0 sendFailedRows : 0
threadId : 10 sentRows : 1 unsentRows : 0 sendFailedRows : 0
threadId : 11 sentRows : 0 unsentRows : 0 sendFailedRows : 0
threadId : 12 sentRows : 1 unsentRows : 0 sendFailedRows : 0

110
      */

MultithreadedTableWriter 回调的使用

MultithreadedTableWriter 在开启回调后，用户会在回调的方法中获取到一个 BasicTable 类型的回调表，该表由两列构成： 第一列（String类型），存放的是调用 MultithreadedTableWriter.insert 时增加的每一行的 id；第二列（布尔值），表示每一行写入成功与否，true 表示写入成功，false 表示写入失败。

-继承 Callback 接口并重载 writeCompletion 方法用于获取回调数据

示例：

public class CallbackHandler : Callback
{
    public void writeCompletion(ITable callbackTable)
    {
        List<String> failedIdList = new List<string>();
        BasicStringVector idVec = (BasicStringVector)callbackTable.getColumn(0);
        BasicBooleanVector successVec = (BasicBooleanVector)callbackTable.getColumn(1);
        for (int i = 0; i < successVec.rows(); i++)
        {
            if (!successVec.getBoolean(i))
            {
                failedIdList.Add(idVec.getString(i));
            }
        }
    }
}

示例：

MultithreadedTableWriter mtw = new MultithreadedTableWriter(host, port, userName, password, dbName, tbName, useSSL,
        enableHighAvailability, null, 10000, 1, 1, "price", null,MultithreadedTableWriter.Mode.M_Append,null, new CallbackHandler());

-调用 MultithreadedTableWriter 的 insert 方法并在第一列中为每一行写入 id

String theme = "theme1";
for (int id = 0; id < 1000000; id++){
    mtw.insert(theme + id, code, price); //theme+id 为每一行对应的 id，将在回调时返回
}

DolphinDB CSHARP API 提供 AutoFitTableUpsert 类对象来更新并写入 DolphinDB 的表。AutoFitTableUpsert 同 MultithreadedTableWriter 指定 mode 为 Mode.M_Upsert 时更新表数据的功能一样，区别在于 AutoFitTableUpsert 为单线程写入，而 MultithreadedTableWriter 为多线程写入。

-AutoFitTableUpsert的主要方法如下：

-构造方法：

AutoFitTableUpsert(string dbUrl, string tableName, DBConnection connection, bool ignoreNull, string[] pkeyColNames, string[] psortColumns)

参数说明：

• dbUrl 字符串，表示分布式数据库地址。内存表时该参数为空。
• tableName 字符串，表示分布式表或内存表的表名。
• connection DBConnection 对象，用于连接 server 并 upsert 数据。注意：创建用于 AutoFitTableUpsert 的 DBConnection 对象时，asynchronousTask 必须为 false。
• ignoreNull 布尔值，表示 upsert!的一个参数，其含义为若 upsert! 的新数据表中某元素为 NULL 值，是否对目标表中的相应数据进行更新。
• pkeyColNames 字符串数组，用于指定 upsert! 的 keyColNames 参数，即指定 DFS 表（目标表）的键值列。
• psortColumns 字符串数组，用于指定 upsert! 的 sortColumns 参数，设置该参数后，更新的分区内的所有数据会根据指定的列进行排序。排序在每个分区内部进行，不会跨分区排序。

-写入并更新数据的方法：

int upsert(BasicTable table)

函数说明：

将一个 BasicTable 对象更新到目标表中，返回一个 int 类型，表示更新了多少行数据。

AutoFitTableUpsert 使用示例如下：

DBConnection conn = new DBConnection(false, false, false);
conn.connect("192.168.1.116", 18999, "admin", "123456");
String dbName = "dfs://upsertTable";
String tableName = "pt";
String script = "dbName = \"dfs://upsertTable\"\n" +
"if(exists(dbName)){\n" +
"\tdropDatabase(dbName)\t\n" +
"}\n" +
"db  = database(dbName, RANGE,1 10000,,'TSDB')\n" +
"t = table(1000:0, `id`value,[ INT, INT[]])\n" +
"pt = db.createPartitionedTable(t,`pt,`id,,`id)";
conn.run(script);

BasicIntVector v1 = new BasicIntVector(3);
v1.setInt(0, 1);
v1.setInt(1, 100);
v1.setInt(2, 9999);

BasicArrayVector ba = new BasicArrayVector(DATA_TYPE.DT_INT_ARRAY);
ba.append(v1);
ba.append(v1);
ba.append(v1);

List<String> colNames = new List<string>();
colNames.Add("id");
colNames.Add("value");
List<IVector> cols = new List<IVector>();
cols.Add(v1);
cols.Add(ba);
BasicTable bt = new BasicTable(colNames, cols);
String[] keyColName = new String[] { "id" };
AutoFitTableUpsert aftu = new AutoFitTableUpsert(dbName, tableName, conn, false, keyColName, null);
aftu.upsert(bt);
BasicTable res = (BasicTable)conn.run("select * from pt;");
System.Console.Out.WriteLine(res.getString());

三种方法对应的 subscribe 接口如下：

1. 通过 ThreadedClient 方式订阅的接口：

subscribe(string host, int port, string tableName, string actionName, MessageHandler handler, long offset, bool reconnect, IVector filter, int batchSize, float throttle = 0.01f, StreamDeserializer deserializer = null, string user = "", string password = "")

• host 是发布端节点的 IP 地址。
• port 是发布端节点的端口号。
• tableName 是发布表的名称。
• actionName 是订阅任务的名称。
• handler 是用户自定义的回调函数，用于处理每次流入的数据。
• offset 是整数，表示订阅任务开始后的第一条消息所在的位置。消息是流数据表中的行。如果没有指定 offset，或它为负数或超过了流数据表的记录行数，订阅将会从流数据表的当前行开始。offset 与流数据表创建时的第一行对应。如果某些行因为内存限制被删除，在决定订阅开始的位置时，这些行仍然考虑在内。
• reconnect 是布尔值，表示订阅中断后，是否会自动重订阅。
• filter 是一个向量，表示过滤条件。流数据表过滤列在 filter 中的数据才会发布到订阅端，不在 filter 中的数据不会发布。
• batchSize 是一个整数，表示批处理的消息的数量。如果它是正数，直到消息的数量达到 batchSize 时，handler 才会处理进来的消息。如果它没有指定或者是非正数，消息到达之后，handler 就会马上处理消息。
• throttle 是一个浮点数，表示 handler 处理到达的消息之前等待的时间，以秒为单位。默认值为 1。如果没有指定 batchSize，throttle 将不会起作用。
• deserializer 是订阅的异构流表对应的反序列化器。
• user 是一个字符串，表示 API 所连接服务器的登录用户名。
• password 是一个字符串，表示 API 所连接服务器的登录密码。

1. 通过 ThreadPooledClient 方式订阅的接口：

subscribe(string host, int port, string tableName, string actionName, MessageHandler handler, long offset, bool reconnect, IVector filter, StreamDeserializer deserializer = null, string user = "", string password = "")

1. 通过 PollingClient 方式订阅的接口：

subscribe(string host, int port, string tableName, string actionName, long offset, bool reconnect, IVector filter, StreamDeserializer deserializer = null, string user = "", string password = "")

下面分别介绍如何通过3种方法订阅流数据。

• 通过客户机上的应用程序定期去流数据表查询是否有新增数据，推荐使用 PollingClient。

PollingClient client = new PollingClient(subscribeHost, subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, offset);

while (true)
{
      List<IMessage> msgs = poller1.poll(1000);

      if (msgs.Count > 0)
      {
            foreach(IMessage msg in msgs)
            System.Console.Out.WriteLine(string.Format("receive: {0}, {1}, {2}", msg.getEntity(0).getString(), msg.getEntity(1).getString(), msg.getEntity(2).getString()));
      }
      /*
      Successfully subscribed table 192.168.1.38:18848:local8848/Trades/csharpStreamingApi
      receive: 1, 2022.05.26T10:39:22.105, 1.5
      */
}

• 使用 MessageHandler 回调的方式获取新数据。

首先需要调用者定义handler。需要实现 dolphindb.streaming.MessageHandler接 口。

public class MyHandler : MessageHandler
{
      public void doEvent(IMessage msg)
      {
            System.Console.Out.WriteLine(string.Format("receive: {0}, {1}, {2}", msg.getEntity(0).getString(), msg.getEntity(1).getString(), msg.getEntity(2).getString()));
      }

      public void batchHandler(List<IMessage> msgs)
      {
            throw new NotImplementedException();
      }
}

在启动订阅时，把 handler 实例作为参数传入订阅函数。包括单线程回调或多线程回调两种方式：

1. 单线程回调 ThreadedClient

ThreadedClient client = new ThreadedClient(subscribeHost, subscribePort);
client.subscribe(serverIP, serverPort, tableName, new MyHandler());
Thread.Sleep(10000);
//如果需要停止订阅，可以使用close函数。
client.close();

1. 多线程回调(ThreadPollingClient)：handler 模式客户端(线程池处理任务)

ThreadPooledClient client = new ThreadPooledClient(subscribeHost, subscribePort);
client.subscribe(serverIP, serverPort, tableName, new MyHandler());
//如果需要停止订阅，可以使用close函数。
Thread.Sleep(10000);
client.close();

reconnect参数是一个布尔值，表示订阅意外中断后，是否会自动重新订阅。默认值为false。如果reconnect=true，有以下三种情况：

• 如果发布端与订阅端处于正常状态，但是网络中断，那么订阅端会在网络正常时，自动从中断位置重新订阅。
• 如果发布端崩溃，订阅端会在发布端重启后不断尝试重新订阅。
  • 如果发布端对流数据表启动了持久化，发布端重启后会首先读取硬盘上的数据，直到发布端读取到订阅中断位置的数据，订阅端才能成功重新订阅。
  • 如果发布端没有对流数据表启用持久化，那么订阅端将自动重新订阅失败。
• 如果订阅端崩溃，订阅端重启后不会自动重新订阅，需要重新执行subscribe函数。

以下例子在订阅时，设置 reconnect 为 true：

PollingClient client = new PollingClient(subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, offset, true);

filter参数是一个向量。该参数需要发布端配合setStreamTableFilterColumn函数一起使用。使用setStreamTableFilterColumn指定流数据表的过滤列，流数据表过滤列在filter中的数据才会发布到订阅端，不在filter中的数据不会发布。

以下例子将一个包含元素1和2的整数类型向量作为subscribe的filter参数：

BasicIntVector filter = new BasicIntVector(2);
filter.setInt(0, 1);
filter.setInt(1, 2);

PollingClient client = new PollingClient(subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, actionName, offset, filter);

DolphinDB server 自 1.30.17 及 2.00.5 版本开始，支持通过replay函数将多个结构不同的流数据表，回放（序列化）到一个流数据表里，这个流数据表被称为异构流数据表。Python API 自 1.30.19 版本开始，新增 StreamDeserializer 类，用于构造异构流数据表反序列化器，以实现对异构流数据表的订阅和反序列化操作。

C# API 通过 StreamDeserializer 类来构造异构流数据表反序列化器，语法如下：

1. 通过指定表的schema进行构造，包含以下两种方式，指定表的schema信息或指定表的各列类型 ：

指定表的schema信息：

StreamDeserializer(Dictionary<string, BasicDictionary> filters)

指定表的各列类型：

StreamDeserializer(Dictionary<string, List<DATA_TYPE>> filters)

1. 通过指定表进行构造：

StreamDeserializer(Dictionary<string, Tuple<string, string>> tableNames, DBConnection conn = null)

订阅示例：

//假设异构流数据表回放时inputTables如下：
//d = dict(['msg1', 'msg2'], [table1, table2]); \
//replay(inputTables = d, outputTables = `outTables, dateColumn = `timestampv, timeColumn = `timestampv)";
//异构流数据表解析器的创建方法如下：

{//指定schema的方式
      BasicDictionary outSharedTables1Schema = (BasicDictionary)conn.run("table1.schema()");
      BasicDictionary outSharedTables2Schema = (BasicDictionary)conn.run("table2.schema()");
      Dictionary<string, BasicDictionary> schemas = new Dictionary<string, BasicDictionary>();
      schemas["msg1"] = outSharedTables1Schema;
      schemas["msg2"] = outSharedTables2Schema;
      StreamDeserializer streamFilter = new StreamDeserializer(schemas);
}
{//指定表的各列类型
	Dictionary<string, List<DATA_TYPE>> colTypes = new Dictionary<string, List<DATA_TYPE>>();
	List<DATA_TYPE> table1ColTypes = new List<DATA_TYPE> { DATA_TYPE.DT_DATETIME, DATA_TYPE.DT_TIMESTAMP, DATA_TYPE.DT_SYMBOL, DATA_TYPE.DT_DOUBLE, DATA_TYPE.DT_DOUBLE };
	colTypes["msg1"] = table1ColTypes;
	List<DATA_TYPE> table2ColTypes = new List<DATA_TYPE> { DATA_TYPE.DT_DATETIME, DATA_TYPE.DT_TIMESTAMP, DATA_TYPE.DT_SYMBOL, DATA_TYPE.DT_DOUBLE };
	colTypes["msg2"] = table2ColTypes;
	StreamDeserializer streamFilter = new StreamDeserializer(colTypes);
}
{//指定表的方式
      Dictionary<string, Tuple<string, string>> tables = new Dictionary<string, Tuple<string, string>>();
      tables["msg1"] = new Tuple<string, string>("", "table1");
      tables["msg2"] = new Tuple<string, string>("", "table2");
      //conn是可选参数，如果不传入，在订阅的时候会自动使用订阅的conn进行构造
      StreamDeserializer streamFilter = new StreamDeserializer(tables, conn);
}

下面分别介绍如何通过 ThreadedClient, ThreadPooledClient 和 PollingClient 三种方式订阅异构流表：

1. 通过 ThreadedClient 订阅异构流数据表：通过两种方式完成订阅时对异构流数据表的解析操作。

• 通过指定 subscribe 函数的 deserialize 参数，实现在订阅时直接解析异构流表：

ThreadedClient client = new ThreadedClient(listenport);
client.subscribe(hostName, port, tableName, actionName, handler, 0, true, null, -1, (float)0.01, streamFilter);

• 异构流表（streamFilter）也可以写入客户自定义的 Handler 中，在回调时被解析：

public class Handler6 : MessageHandler
      {
      private StreamDeserializer deserializer_;
      private List<BasicMessage> msg1 = new List<BasicMessage>();
      private List<BasicMessage> msg2 = new List<BasicMessage>();

      public Handler6(StreamDeserializer deserializer)
      {
            deserializer_ = deserializer;
      }

      public void batchHandler(List<IMessage> msgs)
      {
            throw new NotImplementedException();
      }

      public void doEvent(IMessage msg)
      {
            try
            {
                  BasicMessage message = deserializer_.parse(msg);
                  if (message.getSym() == "msg1")
                  {
                  msg1.Add(message);
                  }
                  else if (message.getSym() == "msg2")
                  {
                  msg2.Add(message);
                  }
            }
            catch (Exception e)
            {
                  System.Console.Out.WriteLine(e.StackTrace);
            }
      }

      public List<BasicMessage> getMsg1()
      {
            return msg1;
      }

      public List<BasicMessage> getMsg2()
      {
            return msg2;
      }
      };

Handler6 handler = new Handler6(streamFilter);
ThreadedClient client = new ThreadedClient(listenport);
client.subscribe(SERVER, PORT, tableName, actionName, handler, 0, true);

1. 通过 ThreadPooledClient 订阅异构流数据表的方法和 ThreadedClient 一致。

• 指定 subscribe 函数的 deserialize 参数：

ThreadPooledClient client = new ThreadPooledClient(listenport);
client.subscribe(hostName, port, tableName, actionName, handler, 0, true, null, streamFilter);

• 异构流数据表（streamFilter）也可以写入客户自定义的 Handler 中，在回调时被解析：

Handler6 handler = new Handler6(streamFilter);
ThreadPooledClient client = new ThreadPooledClient(listenport);
client.subscribe(hostName, port, tableName, actionName, handler, 0, true);

由于 PollingClient 没有回调函数，只能通过为 subscirbe 的 deserialize 参数传入 streamFilter 的方式进行解析：

PollingClient client = new PollingClient(listenport);
TopicPoller poller = client.subscribe(hostName, port, tableName, actionName, 0, true, null, streamFilter);

每一个订阅都有一个订阅主题topic作为唯一标识。如果订阅时topic已经存在，那么会订阅失败。这时需要通过unsubscribeTable函数取消订阅才能再次订阅。

client.unsubscribe(serverIP, serverPort, tableName,actionName);
```Title