Java API

Java API 通过 TCP/IP 协议连接到 DolphinDB 服务器。连接正在运行的端口号为 8848 的本地 DolphinDB 服务器：

import com.xxdb;
DBConnection conn = new DBConnection();
boolean success = conn.connect("localhost", 8848);

声明 connection 变量的时候，有三个可选参数：asynchronousTask（支持异步），compress（支持开启压缩），useSSL（支持 SSL）。以上三个参数默认值为 false。目前只支持 linux, 稳定版>=1.10.17，最新版>=1.20.6。

下面例子是，建立支持 SSL 而非支持异步的 connection，要求数据进行压缩。服务器端应该添加参数 enableHTTPS=true(单节点部署，需要添加到 dolphindb.cfg;集群部署需要添加到 cluster.cfg)。

DBConnection conn = new DBConnection(false, true, true);

下面建立不支持 SSL，但支持异步的 connection。异步情况下，只能执行 DolphinDB 脚本和函数，且不再有返回值。该功能适用于异步写入数据。

DBConnection conn = new DBConnection(true, false);

输入用户名和密码建立连接：

boolean success = conn.connect("localhost", 8848, "admin", "123456");

若未使用用户名及密码连接成功，则脚本在 Guest 权限下运行。后续运行中若需要提升权限，可以使用 conn.login('admin','123456',true) 登录获取权限。

若需要开启 API 高可用，则须设置 highAvailability=true。2.00.11.0 版本之前，开启高可用模式即开启负载均衡模式；自 2.00.11.0 版本起，connect 方法新增参数 enableLoadBalance，用户可手动关闭高可用模式下的负载均衡功能，且该功能默认为关闭。

以下为同时开启高可用模式和负载均衡功能的情况：

1.30.22.1 及之前版本的 API 将随机选择一个可用节点进行连接；用户也可以通过 highAvailabilitySites 指定可连接的节点组，此时 API 将从 highAvailabilitySites 中随机选择可用节点进行连接。

1.30.22.2 版本起，API 将优先选择低负载节点，判断标准为：内存占用小于 80%、连接数小于 90% 且节点负载小于 80%。即在开启高可用后，API 将优先随机选择一个低负载节点进行连接，若没有低负载节点，则将随机连接一个可用节点。若用户通过 highAvailabilitySites 指定了可连接的节点组，此时 API 将仍优先从 highAvailabilitySites 中随机连接一个低负载节点，若无，则随机选择一个 highAvailabilitySites 中的可用节点。

注意：若 API 断开重连，将按照上述规则重新连接节点。

示例如下：

• 仅开启高可用模式，此时会开启负载均衡功能。

  sites=["192.168.1.2:24120", "192.168.1.3:24120", "192.168.1.4:24120"]
  boolean success = conn.connect("192.168.1.2", 24120, "admin", "123456", highAvailability=true, highAvailabilitySites=sites);

• 开启高可用模式，同时手动开启负载均衡功能。

  boolean success = conn.connect("192.168.1.2", 24120, "admin", "123456", enableHighAvailability=true, highAvailabilitySites=sites, enableLoadBalance);

  若开启高可用模式、同时不开启负载均衡功能：API 将优先从 highAvailabilitySites 中随机选择一个可用节点进行连接。若未设置 highAvailabilitySites，则将随机选择一个集群中的节点。

  注意：不支持仅开启负载均衡功能的情况。

  当需要在应用程序里定义和使用自定义函数时，可以使用 initialScript 参数传入函数定义脚本。这样做的好处是：一、无需每次运行run函数的时候重复定义这些函数。二、API 提供自动重连机制，断线之后重连时会产生新的会话。如果 initialScript 参数不为空，API 会在新的会话中自动执行初始化脚本重新注册函数。在一些网络不是很稳定但是应用程序需要持续运行的场景里，这个参数会非常有用。

  boolean success = conn.connect("localhost", 8848, "admin", "123456", "");

Java API 同时提供 SimpleDBConnectionPool 连接池和 ExclusiveDBConnectionPool 连接池。

• SimpleDBConnectionPool 连接池符合标准连接池的定义，提供连接池的初始化与关闭，连接的获取与释放等常用方法。在构造完连接池后，用户可以通过 getConnection 方法获取一个连接进行使用；在使用完毕后通过 connection.close() 释放连接，使其重回连接池并更新为闲置状态。此后该连接可以再次被获取使用。
• ExclusiveDBConnectionPool 连接池的机制则与任务有关。在构造完连接池后，用户需要调用连接池的 execute 方法来执行 BasicDBTask 任务，后续亦需要使用 BasicDBTask 的相关方法以执行脚本、查询执行状态与结果等，如通过 getResult 方法获取该任务的执行结果。

Java API 自 2.00.11.1 版本起提供 SimpleDBConnectionPool 连接池，以此对连接进行管理和重用。

在使用时，先通过 SimpleDBConnectionPoolConfig 设置连接池的具体参数，然后在构造连接池时，将 SimpleDBConnectionPoolConfig 作为 SimpleDBConnectionPool 的配置参数。SimpleDBConnectionPool 将根据传递的参数进行解析、初始化连接等操作。连接池创建成功后，用户可以通过 getConnection 方法获取一个连接进行使用。使用完毕后，用户可通过 DBConnection.close() 释放连接。连接重回连接池后属于闲置状态，之后可以再次被获取使用。

SimpleDBConnectionPoolConfig 说明

SimpleDBConnectionPoolConfig 仅可通过 setXxx 方法来配置参数，示例如下：

SimpleDBConnectionPoolConfig config = new SimpleDBConnectionPoolConfig();
        config.setHostName("1sss");
        config.setPort(PORT);
        config.setUserId("admin");
        config.setPassword("123456");
        config.setInitialPoolSize(5);
        config.setEnableHighAvailability(false);

目前 SimpleDBConnectionPoolConfig 支持的参数如下：

• hostName ：IP，默认为 localhost。
• port ：端口，默认为 8848。
• userId：用户名，默认为””。
• password：密码，默认为””。用户名和密码仅填写其中一个，创建连接不登录；用户名和密码填写正确，创建连接且登录；用户名和密码填写错误，创建连接失败
• initialPoolSize：连接池的初始连接数，默认为 5。
• initialScript：表示初始化的脚本，默认为空。
• compress：表示是否在下载时对数据进行压缩，默认值为 false。该模式适用于大数据量的查询。压缩数据后再传输，这可以节省网络带宽，但会增加 DolphinDB、API 的计算量。
• useSSL：表示是否使用 SSL 连接，默认值为 false。注意：若要开启 SSL 连接，服务器端的配置文件（如果是单节点为 dolphindb.cfg，如果是集群为 cluster.cfg）须同时配置功能参数 enableHTTPS=true。
• usePython：表示是否开启 Python 解析，默认值为 false。
• loadBalance：表示是否开启负载均衡，默认为 false，为 true 时注意：
  • 如果未指定highAvailabilitySites，Java API 会对 server 集群的所有节点采取轮询策略的负载均衡。
  • 如果指定了highAvailabilitySites ，Java API 将对 highAvailabilitySites 数组中的连接节点执行轮询策略的负载均衡。
• enableHighAvailability：表示是否开启高可用，默认为 false。
• highAvailabilitySites：表示开启高可用情况下指定填入的主机名和端口号数组，默认为 null。

SimpleDBConnectionPool 类的相关方法

注意：该处的 DBConnection.close() 方法区别于 DBConnection 类中关闭当前会话的功能，此处仅用于将使用 getConnection 获取的连接释放到连接池。

连接池使用示例

// 设置连接池参数
SimpleDBConnectionPoolConfig config = new SimpleDBConnectionPoolConfig();
        config.setHostName("1sss");
        config.setPort(PORT);
        config.setUserId("admin");
        config.setPassword("123456");
        config.setInitialPoolSize(5);
        config.setEnableHighAvailability(false);
 
// 初始化连接池       
SimpleDBConnectionPool pool = new SimpleDBConnectionPool(config);

// 从连接池中获取一个连接
DBConnection conn = pool.getConnection();
conn.run("..."); // 执行脚本

// 释放当前连接
conn.close();

// 获取当前使用中的连接数
int activeConns = pool.getActiveConnectionsCount();

// 获取当前空闲的连接数
int idleConns = pool.getIdleConnectionsCount();

// 关闭连接池
pool.close();

Java API 提供 ExclusiveDBConnectionPool 连接池用以执行任务。用户可以通过 execute 方法执行任务，然后使用 BasicDBTask 的 getResult 方法获取该任务的执行结果。

以下为 ExclusiveDBConnectionPool 方法的参数介绍：

• host：表示主机名。
• port： 表示端口。
• uid： 可选参数，表示登录的用户名，默认值为空 。
• pwd：可选参数，表示登录的密码，默认值为空。
• count：正整数，表示连接池中的连接数量。当提交的任务数量大于连接池的连接数时，部分任务会进行等待，直到出现空闲的连接再进行分配。
• loadBalance：表示在初始化连接时是否开启负载均衡。
  • 如果不开启，连接池中的所有连接都按照 host、port 建立连接。
  • 如果开启，在连接池初始化时，API 会获取集群中所有的节点信息，然后将连接池中的连接均匀分配给这些节点。
• enableHighAvailability：可选参数，表示是否针对连接池中的连接开启高可用，默认为 false，表示不开启。
• highAvailabilitySites：可选参数，表示所有可用节点的地址和端口，格式为 ip:port。须开启 enableHighAvailability 后才可使用，默认为 null。该参数处理逻辑如下：
  • 如果为空，使用 3.00.0.1 之前版本，则仅支持连接通过 host、port 指定的节点；使用 3.00.0.1 及之后版本，会将集群中所有节点做为高可用节点。
  • 如果不为空，此时会将 highAvailabilitySites 指定的节点做为高可用节点。若当前节点断开，会按照 highAvailabilitySites 中填写的节点顺序进行切换。
• initialScript：可选参数，表示初始化的脚本，默认为空。
• compress：可选参数，表示是否在下载时压缩数据，默认值为 false，表示不压缩。该模式适用于大数据量的查询。压缩数据后再传输可以节省网络带宽，但也会增加 DolphinDB 服务器和 API 客户端的计算量。
• useSSL：可选参数，表示是否使用 SSL 连接，默认值为 false，表示不使用。注意：若要开启 SSL 连接，服务器端的配置文件（单节点：dolphindb.cfg，集群：cluster.cfg）须同时配置功能参数 enableHTTPS=true。
• usePython：可选参数，表示是否开启 Python 解析，默认值为 false，表示不开启。

BasicDBTask 包装了需要执行的脚本和参数。

建立一个数量为 10 的连接池。

ExclusiveDBConnectionPool pool = new ExclusiveDBConnectionPool(hostName, port, userName, passWord, 10, false, false);

创建一个任务。

BasicDBTask task = new BasicDBTask("1..10");
pool.execute(task);

检查任务是否执行成功。如果执行成功，获取相应结果；如果失败，获取异常信息。

BasicIntVector data = null;
if (task.isSuccessful()) {
    data = (BasicIntVector)task.getResult();
} else {
    throw new Exception(task.getErrorMsg());
}
System.out.print(data.getString());

输出

[1,2,3,4,5,6,7,8,9,10]

创建多个任务，在 ExclusiveDBConnectionPool 上并行调用。

List<DBTask> tasks = new ArrayList<>();
for (int i = 0; i < 10; ++i){
    //调用函数 log。
    tasks.add(new BasicDBTask("log", Arrays.asList(data.get(i))));
}
pool.execute(tasks);

检查任务是否都执行成功。如果执行成功，获取相应结果；如果失败，获取异常信息。

List<Entity> result = new ArrayList<>();
for (int i = 0; i < 10; ++i)
{
    if (tasks.get(i).isSuccessful())
    {
        result.add(tasks.get(i).getResult());
    }
    else
    {
        throw new Exception(tasks.get(i).getErrorMsg());
    }
    System.out.println(result.get(i).getString());
}

输出：

0
0.693147
1.098612
1.386294
1.609438
1.791759
1.94591
2.079442
2.197225
2.302585

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

DolphinDB Java API 提供 MultithreadedTableWriter 类对象用于批量异步追加数据，并在客户端维护了一个数据缓冲队列。当服务器端忙于网络 I/O 时，客户端写线程仍然可以将数据持续写入缓冲队列（该队列由客户端维护）。写入队列后即可返回，从而避免了写线程的忙等。目前，MultithreadedTableWriter 支持批量写入数据到内存表、分区表和维度表。

注意，对于异步写入：

• API 客户端提交任务到缓冲队列，缓冲队列接到任务后，客户端即认为任务已完成。
• 提供 getStatus 等接口查看状态。

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

AutoFitTableUpsert 的主要方法如下：

• 构造方法：

AutoFitTableUpsert(String dbUrl, String tableName, DBConnection connection, boolean ignoreNull, String[] pkeyColNames, String[] psortColumns)

参数说明：

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

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

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(Entity.DATA_TYPE.DT_INT_ARRAY, 1);
ba.Append(v1);
ba.Append(v1);
ba.Append(v1);

List<String> colNames = new ArrayList<>();
colNames.add("id");
colNames.add("value");
List<Vector> cols = new ArrayList<>();
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.out.println(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 所连接服务器的登录密码。

2. 通过 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 = "")

3. 通过 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(subscribePort);
TopicPoller poller1 = client.subscribe(serverIP, serverPort, tableName, offset);

while (true) {
   ArrayList<IMessage> msgs = poller1.poll(1000);
   if (msgs.size() > 0) {
         BasicInt value = msgs.get(0).getEntity(2);  //取数据中第一行第三个字段
   }
}

poller1 探测到流数据表有新增数据后，会拉取到新数据。无新数据发布时，Java 程序会阻塞在 poller1.poll 方法这里等待。

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

首先需要调用者定义数据处理器 handler。handler 需要实现 com.xxdb.streaming.client.MessageHandler 接口。

public class MyHandler implements MessageHandler {
       public void doEvent(IMessage msg) {
               BasicInt qty = msg.getValue(2);
               //..处理数据
       }
}

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

1. 单线程回调 ThreadedClient

   ThreadedClient client = new ThreadedClient(subscribePort);
   client.subscribe(serverIP, serverPort, tableName, new MyHandler(), offsetInt);

当流数据表有新增数据时，系统会通知 Java API 调用 MyHandler 方法，将新数据通过 msg 参数传入。

1. 多线程回调 ThreadPooledClient

   ThreadPooledClient client = new ThreadPooledClient(10000,10);
   client.subscribe(serverIP, serverPort, tableName, new MyHandler(), offsetInt);

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 函数将多个结构不同的流数据表，回放（序列化）到一个流数据表里，这个流数据表被称为异构流数据表。Java API 自 1.30.19 版本开始，新增 StreamDeserializer 类，用于构造异构流数据表反序列化器，以实现对异构流数据表的订阅和反序列化操作。

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

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

   指定表的 schema 信息：

   StreamDeserializer(Map<String, BasicDictionary> filters)

   指定表的各列类型：

   StreamDeserializer(HashMap<String, List<Entity.DATA_TYPE>> filters)

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

   StreamDeserializer(Map<String, Pair<String, String>> tableNames, DBConnection conn)

   订阅示例：

   //假设异构流数据表回放时 inputTables 如下：
   //d = dict(['msg1', 'msg2'], [table1, table2]); \
   //replay(inputTables = d, outputTables = `outTables, dateColumn = `timestampv, timeColumn = `timestampv)";
   //异构流数据表解析器的创建方法如下：
   
   {//指定 schema 的方式
       BasicDictionary table1_schema = (BasicDictionary)conn.run("table1.schema()");
       BasicDictionary table2_schema = (BasicDictionary)conn.run("table2.schema()");
       Map<String,BasicDictionary > tables = new HashMap<>();
       tables.put("msg1", table1_schema);
       tables.put("msg2", table2_schema);
       StreamDeserializer streamFilter = new StreamDeserializer(tables);
   }
   {//指定表的各列类型
       Entity.DATA_TYPE[] array1 = {DT_DATETIME,DT_TIMESTAMP,DT_SYMBOL,DT_DOUBLE,DT_DOUBLE};
       Entity.DATA_TYPE[] array2 = {DT_DATETIME,DT_TIMESTAMP,DT_SYMBOL,DT_DOUBLE};
       List<Entity.DATA_TYPE> filter1 = new ArrayList<>(Arrays.asList(array1));
       List<Entity.DATA_TYPE> filter2 = new ArrayList<>(Arrays.asList(array2));
       HashMap<String, List<Entity.DATA_TYPE>> filter = new HashMap<>();
       filter.put("msg1",filter1);
       filter.put("msg2",filter2);
       StreamDeserializer streamFilter = new StreamDeserializer(filter);
   }
   {//指定表的方式
       Map<String, Pair<String, String>> tables = new HashMap<>();
       tables.put("msg1", new Pair<>("", "table1"));
       tables.put("msg2", new Pair<>("", "table2"));
       //conn 是可选参数，如果不传入，在订阅的时候会自动使用订阅的 conn 进行构造
       StreamDeserializer streamFilter = new StreamDeserializer(tables, conn);
   }

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

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

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

     ThreadedClient client = new ThreadedClient(8676);
     client.subscribe(hostName, port, tableName, actionName, handler, 0, true, null, streamFilter, false);

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

     class Handler6 implements MessageHandler {
         private StreamDeserializer deserializer_;
         private List<BasicMessage> msg1 = new ArrayList<>();
         private List<BasicMessage> msg2 = new ArrayList<>();
     
         public Handler6(StreamDeserializer deserializer) {
             deserializer_ = deserializer;
         }
         public void batchHandler(List<IMessage> msgs) {
         }
     
         public void doEvent(IMessage msg) {
             try {
                     BasicMessage message = deserializer_.parse(msg);
                     if (message.getSym().equals("msg1")) {
                         msg1.add(message);
                     } else if (message.getSym().equals("msg2")) {
                         msg2.add(message);
                     }
             } catch (Exception e) {
                     e.printStackTrace();
             }
         }
     
         public List<BasicMessage> getMsg1() {
             return msg1;
         }
         public List<BasicMessage> getMsg2() {
             return msg2;
         }
     }
     
     ThreadedClient client = new ThreadedClient(listenport);
     Handler6 handler = new Handler6(streamFilter);
     client.subscribe(hostName, port, tableName, actionName, handler, 0, true);

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

   • 指定 subscribe 函数的 deserialize 参数：

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

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

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

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

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

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

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

API 提供 EventSchema 工具，用于在客户端自定义事件。

public class EventSchema {
  private String eventType;                         
  private List<String> fieldNames;                  
  private List<Entity.DATA_TYPE> fieldTypes;           
  private List<Entity.DATA_FORM> fieldForms;            
  private List<Integer> fieldExtraParams;              
}

参数说明：

• eventType：字符串，表示事件。
• fieldNames：字符串，表示字段名。
• fieldTypes：表示字段数据类型。
• fieldForms：表示字段数据形式。
• fieldExtraParams：非负整数，额外参数，当字段类型为 DECIMAL 时，用来指定 scale（DECIMAL 类型的特有参数，用于指定小数位数）。

API 提供 EventSender 工具，用于向存储事件信息的异构流表中写入序列化后的事件数据。DolphinDB 服务端的 CEP 引擎通常会订阅这些流表，以便捕获并处理相应的事件。

构造函数：

public EventSender(DBConnection conn, String tableName, List<EventSchema> eventSchemas, List<String> eventTimeKeys, List<String> commonKeys) {

参数介绍：

• conn：必填参数，表示已连接成功的 DBConnection 对象，注意该 DBConnection 不能是异步模式。
• tableName：必填参数，字符串，表示要写入的异构流表的名字。
• eventSchemas：必填参数，表示 CEP 引擎可以处理的所有事件结构，即可能会发送的所有事件结构。注意，事件结构必须与创建 CEP 引擎时指定的结构相同，否则无法正确处理。
• eventTimeFields：可选参数，字符串标量或字符串列表，用来指定事件的时间字段。如果待写入的异构流表第一列是时间类型，则须指定每个事件中的时间类型字段。
  • 如果所有事件中的时间字段名字相同，则该参数只需指定为一个字符串标量。
  • 如果各个事件的时间字段名称不同，则该参数需要指定为一个字符串向量。其中各元素的顺序与 eventSchema 参数的传入顺序保持一致。
• commonFields：可选参数，字符串列表，表示事件中具有相同名称的字段。可以将事件之间相同的字段单独作为列存入流表中，以便在订阅中进行过滤。

sendEvent 发送

将事件发送至 DolphinDB 服务端。

public void sendEvent(String eventType, List<Entity> attributes)

参数说明：

• eventType：必填参数，字符串，表示事件。
• attributes：必填参数，表示事件中各成员的值。注意，传入值的顺序必须与定义 EventSchema 结构时指定的字段顺序相同，且其数据类型和数据形式也须与之匹配。

API 提供 EventClient 工具，用于订阅异构流表中事件。该异构流表通常接收 DolphinDB 服务端 CEP 引擎处理后并序列化输出的事件数据。

其构造方法如下，其参数与 EventSender 构造函数的相同：

public EventClient(List<EventSchema> eventSchemas, List<String> eventTimeKeys, List<String> commonKeys)

subscribe 订阅

public void subscribe(String host, int port, String tableName, String actionName, MessageHandler handler, long offset, boolean reconnect, String userName, String password)

参数说明：

• host：必填参数，字符串，表示发布端节点的 IP 地址。
• port：必填参数，字符串，表示发布端节点的端口号。
• tableName：必填参数，字符串，表示异构流表的的名称。
• actionName：可选参数，字符串，表示订阅任务的名称，默认为 ”javaStreamingApi” 。
• handler：必填参数，表示用户自定义的回调函数，用于处理每次流入的消息。
• offset：可选参数，整数，表示订阅任务开始后的第一条消息所在的位置，默认为 -1。一条消息对应流数据表中的一行。如果没有指定 offset，或它为负数，或超过了流数据表的记录行数，订阅将会从流数据表的当前行开始。offset 与流数据表创建时的第一行对应。如果某些行因为内存限制被删除，在决定订阅开始的位置时，这些行仍然考虑在内。具体可参考DolphinDB 文档- subscribe。
• reconnect：可选参数，布尔型，表示订阅中断后是否会自动重订阅，默认为 false，表示不订阅。
• user：可选参数，字符串，表示 Java API 所连接服务器的登录用户名，默认为 “” 表示空字符串。
• password：可选参数，字符串，表示 Java API 所连接服务器的登录密码，默认为“” 表示空字符串。

回调函数使用说明

使用时，第一个参数表示事件名称，第二个参数就是各个成员的值。

EventMessageHandler handler = new EventMessageHandler() {
    @Override
    public void doEvent(String eventType, List<Entity> attribute) {
        // 处理逻辑
    }
};

unsubscribe 取消订阅

public void unsubscribe(String host, int port, String tableName, String actionName)

参数说明：

• host：必填参数，字符串，表示发布端节点的 IP 地址。
• port：必填参数，整型，表示发布端节点的端口号。
• tableName：必填参数，字符串，表示异构流表名。
• actionName：必填参数，字符串，表示任务名称。

本例将先介绍在 server 端创建事件、用于序列化事件的接口、分别用于写入和写出的异构流表、以及处理事件的 CEP 引擎；然后在 API 端使用 EventSender 向异构流表发送事件；当 CEP 引擎通过订阅接收并处理完数据后会向另一个异构流表输出消息， API 端再使用 EventClient 从该表订阅输出的事件。

Server 端创建基本要素

下述脚本将在 DolphinDB 服务端创建两个异构流表 input 和 output，两个 streamEventSerializer（序列化接口），一个 CEP 引擎，以及一个 MarketData 事件。

login("admin","123456");

class MarketData{
  market :: STRING
  code :: SYMBOL
  price :: DOUBLE
  qty :: INT
  eventTime :: TIMESTAMP
  def MarketData(m,c,p,q){
    market = m
    code = c
    price = p
    qty = q
    eventTime = now()   
  }
}   

class MainMonitor{
  
  def MainMonitor(){
  }
  def updateMarketData(event)

  def onload(){
    addEventListener(updateMarketData,'MarketData',,'all')
      }

  def updateMarketData(event){
    emitEvent(event)
  }
}

dummy = table(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs)
share streamTable(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs) as intput
share streamTable(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs) as output
schema = table(1:0, `eventType`eventKeys`eventValuesTypeString`eventValueTypeID`eventValuesFormID, [STRING, STRING, STRING, INT[], INT[]])
insert into schema values("MarketData", "market,code,price,qty,eventTime", "STRING,STRING,DOUBLE,INT,TIMESTAMP", [18 18 16 4 12], [0 0 0 0 0])
inputSerializer = streamEventSerializer(name=`serInput, eventSchemes=schema, outputTable=intput, eventTimeKey = "eventTime")
outputSerializer = streamEventSerializer(name=`serOutput, eventSchemes=schema, outputTable=output, eventTimeKey = "eventTime")

engine = createCEPEngine('cep1', <MainMonitor()>, dummy, [MarketData], 1, 'eventTime', 10000, outputSerializer)
subscribeTable(,`intput, `subopt, 0, getStreamEngine('cep1'),true)

marketData1 = MarketData('sz', 's001', 9.8, 100)
marketData2 = MarketData('sz', 's002', 6.9, 100)
marketData3 = MarketData('sz', 's003', 4.8, 100)
marketData4 = MarketData('sz', 's004', 9.8, 100)
marketData5 = MarketData('sz', 's005', 9.8, 100)
appendEvent(inputSerializer, [marketData1,marketData2,marketData3,marketData4,marketData5])

API 端使用 EventSender 向 *input 表写入数据*

// 定义 schema
EventSchema schema = new EventSchema();
schema.setEventType("MarketData");
schema.setFieldNames(Arrays.asList("market", "code", "price", "qty", "eventTime"));
schema.setFieldTypes(Arrays.asList(DT_STRING, DT_STRING, DT_DOUBLE, DT_INT, DT_TIMESTAMP));
schema.setFieldForms(Arrays.asList(DF_SCALAR, DF_SCALAR, DF_SCALAR, DF_SCALAR, DF_SCALAR));
List<EventSchema> eventSchemas = Collections.singletonList(scheme);
List<String> eventTimeKeys = Collections.singletonList("eventTime");
List<String> commonKeys = new ArrayList<>();

// 创建 EventSender
DBConnection conn = new DBConnection(true);
conn.connect(HOST, PORT, "admin", "123456");
EventSender sender = new EventSender(conn, "input", eventSchemas, eventTimeKeys, commonKeys);

// 准备数据：
List<Entity> attributes = new ArrayList<>();
attributes.add(new BasicString("sz"));
attributes.add(new BasicString("s001"));
attributes.add(new BasicDouble(9.8));
attributes.add(new BasicInt(100));
attributes.add(new BasicTimestamp(LocalDateTime.of(2024,3,13,15,11,10,630)));

sender.sendEvent("MarketData", attributes);

CEP 引擎会去订阅 input 流表，在收到消息后，将消息输出到 output 流表。

**API 端使用 *EventClient 订阅 *output 表

EventSchema scheme = new EventSchema();
scheme.setEventType("MarketData");
scheme.setFieldNames(Arrays.asList("market", "code", "price", "qty", "eventTime"));
scheme.setFieldTypes(Arrays.asList(DT_STRING, DT_STRING, DT_DOUBLE, DT_INT, DT_TIMESTAMP));
scheme.setFieldForms(Arrays.asList(DF_SCALAR, DF_SCALAR, DF_SCALAR, DF_SCALAR, DF_SCALAR));
List<EventSchema> eventSchemas = Collections.singletonList(scheme);
List<String> eventTimeKeys = Collections.singletonList("eventTime");
List<String> commonKeys = new ArrayList<>();

// 自定义的回调函数
EventMessageHandler handler = new EventMessageHandler() {
    @Override
    public void doEvent(String eventType, List<Entity> attribute) {
        // 自定义处理数据
        System.out.println("eventType: " + eventType);
        for (Entity entity : attribute) {
            System.out.println(entity.getString());
        }
        System.out.println("\n");
    }
};

// 创建事件订阅类
EventClient client = new EventClient(eventSchemas, eventTimeKeys, commonKeys);
// 订阅
client.subscribe(HOST, PORT, "output", "ss1", handler, 0, true, "admin", "123456");
Thread.sleep(10000);

// 取消订阅
client.unsubscribe(HOST, PORT, "output", "ss1");