MultithreadedTableWriter

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

   Callback callbackHandler = new Callback(){
       public void writeCompletion(Table callbackTable){
           List<String> failedIdList = new ArrayList<>();
           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));
               }
           }
       }
   };

2. 构造 MultithreadedTableWriter 对象并传入回调对象：

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

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

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

ErrorCodeInfo insert(Object... args)

参数说明

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

函数说明

插入单行数据。返回一个 ErrorCodeInfo 对象，包含 errorCode 和 errorInfo，分别表示错误代码和错误信息。

ErrorCodeInfo 介绍：

• errorCode："" 空字符串表示没有错误发生；如果有错误发生，值为错误码，例如："A2"。
• errorInfo：表示错误信息。当没有错误发生时，返回 null；发生错误时，返回包含错误信息的字符串。
• hasError()：判断插入待转换队列时是否发生错误。如果发生错误，返回 true；反之，返回 false。
• succeed()：判断是否成功插入数据缓存队列。如成功插入，返回 true；反之，返回 false。

ErrorCodeInfo 类提供了 hasError 和 succeed 方法用于获取数据是否成功放入数据队列的结果。在插入一行数据后，推荐调用 hasError 来判断是否成功将数据插入到待转换队列。

如果构造 MultithreadedTableWriter 时开启了回调（见上述参数说明），则每次调用 insert 时，需要在每行数据前面增加一列 String 类型的数据作为每行的标识符（id），此 id 列仅用于回调时返回给用户，不会写入表中。

示例

下例为演示采用多线程模式向表中插入一行记录。

MultithreadedTableWriter multithreadedTableWriter = new MultithreadedTableWriter("localhost", 8848, "admin", "123456", "dfs://valuedb3", "pdatetest",
                false, false, null, 10000, 1,
                5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
ErrorCodeInfo pErrorInfo = multithreadedTableWriter_.insert(new Date(2022, 3, 23), "AAAAAAAB", 10000000L);

List<List<Entity>> getUnwrittenData()

函数说明

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

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

示例

下例为演示获取未写入服务器的数据。

List<List<Entity>> unwrittenData = multithreadedTableWriter_.getUnwrittenData();

ErrorCodeInfo insertUnwrittenData(List<List<Entity>> records)

参数说明

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

函数说明

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

示例

下例为演示插入之前未写入服务器的数据。

List<List<Entity>> unwrittenData = multithreadedTableWriter_.getUnwrittenData();
ErrorCodeInfo ret = multithreadedTableWriter_.insertUnwrittenData(unwrittenData);

Status getStatus()

函数说明

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

返回值说明

Status 是 MultithreadedTableWriter.Status 类，继承自 ErrorCodeInfo 实现。

Status 类具有以下属性和方法。

Status 类属性：

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

Status 类方法：

• hasError()：true 表示数据写入存在错误；false 表示数据写入无错误。
• succeed()：true 表示数据写入成功；false 表示数据写入失败。

示例

下例为演示获取 Writer 工作状态。

MultithreadedTableWriter.Status writeStatus = new MultithreadedTableWriter.Status();
writeStatus = multithreadedTableWriter_.getStatus();

以一个 MultithreadedTableWriterStatus 打印输出为例：

errorCode     : A1
 errorInfo     : Data conversion error: Cannot convert double to LONG
 isExiting     : True
 sentRows      : 2493
 unsentRows    : 0
 sendFailedRows: 7507
 threadStatus  : 
        threadId        sentRows        unsentRows      sendFailedRows
               0               0                 0                7507
        3567691520           415                 0                   0
        3489658624           831                 0                   0
        3481265920           416                 0                   0
        3472873216           416                 0                   0
        3464480512           415                 0                   0
<dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0102c76d30>

以上输出内容显示，在本次 MTW 写入流程中发生了异常，异常代码是 A1，异常信息为 Data conversion error: Cannot convert double to LONG。

isExiting=True 表示当前线程正在退出。

sentRows=2493 表示当前有 2493 条数据已经写入到 DolphinDB 服务端。

unsentRows=0/sendFailedRows=7507 则表示当前仍在 API 的未成功写入 DolphinDB 服务端的数据一共有 0+7507=7507 条。

在后面的表格中，列出了各个后台线程的处理情况，threadId=0 表示所有线程的统计总数。通常而言，成功写入的结果（sentRows）会分别显示在各个线程中，例如上述输出中的 sentRows 列；写入失败的结果（unsentRows、sendFailedRows）会集中显示在 threadId=0 行，表示整个过程中的写入失败条数。

waitForThreadCompletion()

函数说明

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

示例

下例为演示等待 Writer 写入完毕。

multithreadedTableWriter_.waitForThreadCompletion();

下例为演示用 MultithreadedTableWriter 向表中写入一系列数据并在发生错误后在控制台打印错误信息。

步骤包括：

1. 创建 DFS 表。
2. 创建 MTW 对象。
3. 写入数据。
4. 若有异常，进行异常处理。

@Test
public void testMultithreadedTableWriter() {
    DBConnection conn= new DBConnection();
    conn.connect(HOST, PORT, "admin", "123456");
    Random random = new Random();
    
    // 1、创建 DFS 表
    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');";
    conn.run(script);
    
    // 2、创建 MTW 对象
    MultithreadedTableWriter multithreadedTableWriter_ = new MultithreadedTableWriter(HOST, PORT, "admin", "123456", "dfs://valuedb3", "pdatetest",
            false, false, null, 10000, 1,
            5, "id", new int[]{Vector.COMPRESS_LZ4, Vector.COMPRESS_LZ4, Vector.COMPRESS_DELTA});
    ErrorCodeInfo ret;
    
    // 3、写入数据
    try {
        // 插入100行正确数据
        for (int i = 0; i < 100; ++i) {
            ret = multithreadedTableWriter_.insert(new Date(2022, 3, 23), "AAAAAAAB", random.nextInt() % 10000);
        }
    }
    catch (Exception e) {
        // MTW 抛出异常
        System.out.println("MTW exit with exception {0}" + e.getMessage());
    }

    // 等待 MTW 插入完成
    multithreadedTableWriter_.waitForThreadCompletion();
    MultithreadedTableWriter.Status writeStatus = new MultithreadedTableWriter.Status();
    writeStatus = multithreadedTableWriter_.getStatus();
    if (!writeStatus.errorInfo.equals("")) {
        // 如果写入时发生错误
        System.out.println("error in writing !");
    }
    System.out.println("writeStatus: {0}\n" + writeStatus.toString());
    System.out.println(((BasicLong)conn.run("exec count(*) from pt")).getLong());
}

以上代码输出结果为：

writeStatus: {0}
errorCode     : 
errorInfo     : 
isExiting     : true
sentRows      : 100
unsentRows    : 0
sendFailedRows: 0
threadStatus  :
        threadId        sentRows      unsentRows  sendFailedRows
              13              30               0               0
              14              18               0               0
              15              15               0               0
              16              20               0               0
              17              17               0               0

100

以上为多线程写入的结果，因此 threadId 与 sentRows 列的值会根据实际情况而有所不同，用户需要根据实际情况进行判断，正常情况下 sentRows 各数总和应为100。

这里介绍 Mode.M_Upsert 模式的使用介绍：

@Test(timeout = 120000)
public void test_mtw_tableUpsert_DP_updateFirst() throws Exception {
    String script = "if(existsDatabase(\"dfs://upsert\")) {\n" +
            "dropDatabase(\"dfs://upsert\")\n" +
            "}\n" +
            "sym=\"A\" \"B\" \"C\" \"A\" \"D\" \"B\" \"A\"\n" +
            "date=take(2021.12.10,3) join take(2021.12.09, 3) join 2021.12.10\n" +
            "price=8.3 7.2 3.7 4.5 6.3 8.4 7.6\n" +
            "val=10 19 13 9 19 16 10\n" +
            "t=table(sym, date, price, val)\n" +
            "db=database(\"dfs://upsert\", VALUE,\"A\" \"B\" \"C\")\n" +
            "pt=db.createPartitionedTable(t, `pt, `sym)\n" +
            "pt.append!(t)";
    conn.run(script);
    MultithreadedTableWriter mtw = new MultithreadedTableWriter(HOST,PORT,"admin","123456","dfs://upsert","pt",
            false,false,null,1000,1,10,"sym",null,
            MultithreadedTableWriter.Mode.M_Upsert, 
            new String[]{"ignoreNull=false", "keyColNames=`sym"});
    mtw.insert(new BasicString("A"),new BasicDate(LocalDate.of(2021,12,9)),new BasicDouble(11.1),new BasicInt(12));
    mtw.insert(new BasicString("B"),new BasicDate(LocalDate.of(2021,12,9)),new BasicDouble(10.5),new BasicInt(9));
    mtw.insert(new BasicString("E"),new BasicDate(LocalDate.of(2021,12,9)),new BasicDouble(6.9),new BasicInt(11));
    mtw.waitForThreadCompletion();
    BasicTable ua = (BasicTable) conn.run("select * from pt;");
    System.out.println(ua.getString());
}

执行结果：

sym date       price val
--- ---------- ----- ---
A   2021.12.09 11.1  12 
A   2021.12.09 4.5   9  
A   2021.12.10 7.6   10 
B   2021.12.09 10.5  9  
B   2021.12.09 8.4   16 
C   2021.12.10 3.7   13 
D   2021.12.09 6.3   19 
E   2021.12.09 6.9   11