• 海量：日志的产出代价是比较小的，随着应用规模的增大，日志数据体量会非常大。
• 要求高吞吐率：对单条日志从产出到写入的延迟没有要求，而重视的是消费短时间内产生的大量日志数据的吞吐率。
• 处理可异步化：日志是业务性比较低的数据，一般不在业务的主线上，通常是离线处理，所以可异步化。
• 可重复写入：日志是固化的数据，重复写入也不会影响数据的正确性。

• 海量（例如写入放大的群消息等）
  
• 要求高吞吐率：对单条消息的延迟不需要很高，可接受百毫秒级别的延迟，但是更注重的是短时间内产生的海量数据的写入（投递）速度。
• 处理可异步化：消息的处理是完全可以被异步化的
• 可重复写入：消息通常都会标注唯一的消息ID，且消息产生后不会更改，所以重复写入不会带来什么问题。

• 依赖了SDK提供的AsyncClient异步接口
• 导入数据会使用BatchWriteRow接口
• 单行异常重试依赖SDK提供的RetryStrategy

内部架构

图2 OTSWriter内部架构

ClientConfiguration cc = new ClientConfiguration();
cc.setRetryStrategy(new DefaultRetryStrategy()); // 可定制重试策略，若需要保证数据写入成功率，可采用更激进的重试策略
AsyncClient asyncClient = new AsyncClient(endPoint, accessId, accessKey, instanceName, cc);

// 初始化
WriterConfig config = new WriterConfig();
config.setMaxBatchSize(4 * 1024 * 1024); // 配置一次批量导入请求的大小限制，默认是4MB
config.setMaxColumnsCount(128); // 配置一行的列数的上限，默认128列
config.setBufferSize(1024); // 配置内存中最多缓冲的数据行数，默认1024行，必须是2的指数倍
config.setMaxBatchRowsCount(100); // 配置一次批量导入的行数上限，默认100
config.setConcurrency(10); // 配置最大并发数，默认10
config.setMaxAttrColumnSize(2 * 1024 * 1024); // 配置属性列的值大小上限，默认是2MB
config.setMaxPKColumnSize(1024); // 配置主键列的值大小上限，默认1KB
config.setFlushInterval(10000); // 配置缓冲区flush的时间间隔，默认10s

// 配置一个callback，OTSWriter通过该callback反馈哪些导入成功，哪些行导入失败，该callback只简单的统计写入成功和失败的行数。

AtomicLong succeedCount = new AtomicLong();
AtomicLong failedCount = new AtomicLong();
TableStoreCallback<RowChange, ConsumedCapacity> callback = new SampleCallback(succeedCount, failedCount);
ExecutorService executor = Executors.newFixedThreadPool(2);
TableStoreWriter tablestoreWriter = new DefaultTableStoreWriter(asyncClient, tableName, config, callback, executor);

1. AsyncClient：一个提供异步调用的TableStore client，注意由于重试策略是依赖于SDK自身的重试策略，所以若需要定制批量写数据的重试策略，需要在这个Client中配置，如示例所示。
2. WriterConfig：OTSWriter的相关配置，主要包括：限制项（一次批量写的行数上限、一次请求的大小限制等）、并发数（异步并发写入的并发数上限）等。
3. TableStoreCallback<RowChange, ConsumedCapacity>：处理行级别成功或失败的callback。
4. ExecutorService：用于处理callback调用的executor thread pool。

int start = id * rowsCount;
for (int i = 0; i < rowsCount; i++) {
    PrimaryKey primaryKey = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("gid", PrimaryKeyValue.fromLong(start + i))
        .addPrimaryKeyColumn("uid", PrimaryKeyValue.fromLong(start + i)).build();

    RowPutChange rowChange = new RowPutChange(tableName);
    rowChange.setPrimaryKey(primaryKey);
    rowChange.addColumn("col1", ColumnValue.fromBoolean(true));
    rowChange.addColumn("col2", ColumnValue.fromLong(10));
    rowChange.addColumn("col3", ColumnValue.fromString("Hello world."));

    tablestoreWriter.addRowChange(rowChange);
}

private static class SampleCallback implements TableStoreCallback<RowChange, ConsumedCapacity> {
    private AtomicLong succeedCount;
    private AtomicLong failedCount;

    public SampleCallback(AtomicLong succeedCount, AtomicLong failedCount) {
        this.succeedCount = succeedCount;
        this.failedCount = failedCount;
    }

    @Override
    public void onCompleted(RowChange req, ConsumedCapacity res) {

succeedCount.incrementAndGet();
    }

    @Override
    public void onFailed(RowChange req, Exception ex) {

ex.printStackTrace();
        failedCount.incrementAndGet();
    }
}

联系我们

• 阿里云TableStore官方网站
• 阿里云TableStore官方论坛
• 阿里云TableStore官方文档中心
• 阿里云云栖社区
• 阿里云工单系统