dws-client

简介

dws-client是一款基于DWS JDBC实现的高性能、便捷入库工具，用户在使用时必须保证JDBC可以连接。其中使用dws-client入库具备如下优势：

1. dws-client提供对缓存的空间、时间维度的限制，支持攒批提升入库性能，以满足业务在高峰低谷期的入库实时性。
   

   攒批：在实时性要求不严格的场景，对单条数据操作进行缓存，待缓存至多条后批量操作，以提升写入性能。

2. 支持并发入库。
3. 内部实现多种高性能入库方式、主键冲突策略，满足各种场景入库需求。
4. API方式交互，低门槛使用。

图1 dws-client交互场景

引入依赖

1
2
3
4
5

<dependency>
   <groupId>com.huaweicloud.dws</groupId>
   <artifactId>dws-client</artifactId>
   <version>${version}</version>
</dependency>

场景使用

前提条件：初始化client

1
2
3
4
5
6
7
8
9

public DwsClient getClient(){
        DwsConfig config = DwsConfig
                .builder()
                .withUrl("jdbc:gaussdb://***/gaussdb")
                .withUsername("***")
                .withPassword("****")
                .build();
        return new DwsClient(config);
     }

1
2
3
4
5
6
7

return DwsConfig.builder()
                .withUrl(System.getenv("db_url"))
                .withPassword(System.getenv("db_pwd"))
                .withUsername(System.getenv("db_username"))
                .withAutoFlushBatchSize(1000) // 默认攒批1000
                .withTableConfig("test.t_c_batch_size_2", new TableConfig()
                        .withAutoFlushBatchSize(500)); // 对于表test.t_c_batch_size_2 攒批500;

场景二：使用数据库连接执行SQL

该接口主要用于一些特殊业务，在目前支持的功能中无法满足时使用。例如： 数据查询，可以直接使用原生JDBC连接操作数据库。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public void sql() throws DwsClientException {
        Integer id = getClient().sql(connection -> {
            try (ResultSet resultSet = connection.createStatement().executeQuery("select id from test.user where name = 'zhangsan'")) {
                if (resultSet.next()) {
                    return resultSet.getInt("id");
                }
            }
            return null;
        });
        System.out.println("zhangsan id = " + id);
     }

场景三：获取表信息

1
2
3

public void getTableSchema() throws DwsClientException {
        TableSchema tableSchema = getClient().getTableSchema(TableName.valueOf("test.test"));
     }

场景四：数据入库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public void write() throws DwsClientException {
        getClient().write("test.test")
                .setObject("id", 1)
                .setObject("name", "test")
                // 只有在insert时会生效该设置，主键冲突时不更新
                .setObject("age", 38, true)
                // 异步入库，放入后台缓存即可返回
                //.commit()
                // 同步入库 会等待入库成功返回
                .syncCommit();
     }

场景五：数据删除

public void delete() throws DwsClientException {
        getClient().delete("test.test")
                .setObject("id", 1)
                // 异步入库，放入后台缓存即可返回
                //.commit()
                // 同步入库 会等待入库成功返回
                .syncCommit();
    }

public void flush() throws DwsClientException {
        getClient().flush();
    }

场景七：关闭资源

public void close() throws IOException {
        getClient().close();
     }

监听数据入库成功的事件

在异步入库的场景，想要知道哪些数据已经入库，则可通过绑定flushSuccess函数接口实现，该接口会在数据库事务提交完成后回调，回调时会将该批入库数据传递给接口。

public DwsClient getClient() {
        DwsConfig config = DwsConfig
                .builder()
                .withUrl("jdbc:postgresql://***/gaussdb")
                .withUsername("***")
                .withPassword("****")
                .onFlushSuccess(records -> {
                    for (Record record : records) {
                        log.info("flush success. value = {}, pk = {}", RecordUtil.toMap(record), RecordUtil.getRecordPrimaryKeyValue(record));
                    }
                })
                .build();
        return new DwsClient(config);
     }

监听后台异常任务

异步入库数据会由后台任务执行，后台任务执行失败想要感知可通过绑定error函数接口实现，否则只能在下次提交时发生异常错误，由业务感知，如果绑定接口不上报异常，那么该异常将消除，不会再下次提交时报出，否则会以接口异常在下次提交时报出给业务。

public DwsClient getClient() {
        DwsConfig config = DwsConfig
                .builder()
                .withUrl("jdbc:postgresql://***/gaussdb")
                .withUsername("***")
                .withPassword("****")
                .onError((clientException, client) -> {
                    if (clientException instanceof DwsClientRecordException) {
                        DwsClientRecordException recordException = (DwsClientRecordException) clientException;
                        List<Record> records = recordException.getRecords();
                        List<DwsClientException> exceptions = recordException.getExceptions();
                        for (int i = 0; i < records.size(); i++) {
                            log.error("pk = {} . error = {}", RecordUtil.getRecordPrimaryKeyValue(records.get(i)), exceptions.get(i));
                        }
                    }
                    if (clientException.getCode() != ExceptionCode.CONNECTION_ERROR && clientException.getCode() != ExceptionCode.LOCK_ERROR) {
                        throw clientException;
                    }
                    log.error("code = {}", clientException.getCode(), clientException.getOriginal());
                    return null;
                })
                .build();
        return new DwsClient(config);
     }

异常处理

异常可分为三类：

1. InvalidException运行时异常不显示抛出，触发在请求参数无效时。
2. DwsClientException对所有异常的封装，包含被解析的code以及原始异常。
3. DwsClientRecordException对DwsClientException的扩展，包含写入异常的数据集合以及对应DwsClientException异常。

异常code对照：

public enum ExceptionCode {
    /**
     * 无效参数
     */
    INVALID_CONFIG(1),

    /**
     * 连接异常
     */
    CONNECTION_ERROR(100),
    /**
     * 只读
     */
    READ_ONLY(101),
    /**
     * 超时
     */
    TIMEOUT(102),
    /**
     * 连接数过多
     */
    TOO_MANY_CONNECTIONS(103),
    /**
     * 加锁异常
     */
    LOCK_ERROR(104),


    /**
     * 认证失败
     */
    AUTH_FAIL(201),
    /**
     * 已经关闭
     */
    ALREADY_CLOSE(202),
    /**
     * 无权限
     */
    PERMISSION_DENY(203),
    SYNTAX_ERROR(204),
    /**
     * 内部异常
     */
    INTERNAL_ERROR(205),
    /**
     * 中断异常
     */
    INTERRUPTED(206),
    /**
     * 表未发现
     */
    TABLE_NOT_FOUND(207),
    CONSTRAINT_VIOLATION(208),
    DATA_TYPE_ERROR(209),
    DATA_VALUE_ERROR(210),

    /**
     * 解析不到的异常
     */
    UNKNOWN_ERROR(500);
    private final int code;

 }

详细配置说明