跳转到主内容
跳转到主内容

Java 客户端

v0.8+

用于通过其协议与数据库服务器通信的 Java 客户端库。当前实现仅支持 HTTP 接口。 该库提供自己的 API 用于向服务器发送请求，并提供用于处理不同二进制数据格式 (RowBinary* 和 Native*) 的工具。

配置​

<dependency>
  <groupId>com.clickhouse</groupId>
  <artifactId>client-v2</artifactId>
  <version>0.9.8</version>
</dependency>

初始化​

Client 对象由 com.clickhouse.client.api.Client.Builder#build() 初始化。每个客户端都有自己的上下文，客户端之间不共享对象。 Builder 提供了便于配置的方法。

示例：

 Client client = new Client.Builder()
                .addEndpoint("https://clickhouse-cloud-instance:8443/")
                .setUsername(user)
                .setPassword(password)
                .build();

Client 实现了 AutoCloseable 接口，不再需要时应将其关闭。

身份验证​

身份验证在初始化阶段为每个客户端单独配置。支持三种身份验证方式：基于密码、基于访问令牌、基于 SSL 客户端证书。

使用密码进行身份验证时，需要通过调用 setUsername(String) 和 setPassword(String) 方法来设置用户名和密码：

 Client client = new Client.Builder()
        .addEndpoint("https://clickhouse-cloud-instance:8443/")
        .setUsername(user)
        .setPassword(password)
        .build();

使用访问令牌进行身份验证时，需要调用 setAccessToken(String) 来设置访问令牌：

 Client client = new Client.Builder()
        .addEndpoint("https://clickhouse-cloud-instance:8443/")
        .setAccessToken(userAccessToken)
        .build();

通过 SSL 客户端证书进行身份验证时，需要设置用户名、启用 SSL 身份验证，并设置客户端证书和客户端密钥，可分别通过调用 setUsername(String)、useSSLAuthentication(boolean)、setClientCertificate(String) 和 setClientKey(String) 来完成：

Client client = new Client.Builder()
        .useSSLAuthentication(true)
        .setUsername("some_user")
        .setClientCertificate("some_user.crt")
        .setClientKey("some_user.key")
注意

SSL 身份验证在生产环境中可能比较难以排查，因为许多来自 SSL 库的错误信息不够详细，无法提供足够的上下文。例如，如果客户端证书与私钥不匹配，服务器会立即终止连接 (对于 HTTP 来说，这发生在连接建立阶段，此时客户端尚未发送任何 HTTP 请求，因此也就不会收到任何响应) 。

请使用 openssl 等工具验证证书和密钥：

  • 校验密钥完整性：openssl rsa -in [key-file.key] -check -noout
  • 检查客户端证书中的 CN 是否与该 USER 匹配：
    • 从用户证书中获取 CN 值：openssl x509 -noout -subject -in [user.cert]
    • 确认在数据库中已设置相同的值：select name, auth_type, auth_params from system.users where auth_type = 'ssl_certificate' (查询会输出 auth_params 字段，其内容类似于 {"common_names":["some_user"]})

配置​

所有设置均由实例方法 (亦称配置方法) 定义，从而使每个值的作用域和上下文一目了然。 主要配置参数在单一作用域 (客户端或操作) 中定义，互不覆盖。

客户端配置在创建时定义。请参阅 com.clickhouse.client.api.Client.Builder。

客户端配置​

MethodArgumentsDescriptionDefaultKey
addEndpoint(String endpoint)endpoint - URL 格式的服务器地址将服务器端点添加到可用服务器列表。目前仅支持一个端点。nonenone
addEndpoint(Protocol protocol, String host, int port, boolean secure)protocol - 连接协议
host - IP 或主机名
secure - 使用 HTTPS		将服务器端点添加到可用服务器列表。目前仅支持一个端点。nonenone
enableConnectionPool(boolean enable)enable - 启用/禁用标志设置是否启用连接池trueconnection_pool_enabled
setMaxConnections(int maxConnections)maxConnections - 连接数量设置客户端对每个服务器端点最多可以打开的连接数。10max_open_connections
setConnectionTTL(long timeout, ChronoUnit unit)timeout - 超时值
unit - 时间单位		设置连接的生存时间 (TTL)，超过该时间后连接将被视为不再活动-1connection_ttl
setKeepAliveTimeout(long timeout, ChronoUnit unit)timeout - 超时值
unit - 时间单位		设置 HTTP 连接的 Keep-Alive 超时时间。设置为 0 以禁用 Keep-Alive。-http_keep_alive_timeout
setConnectionReuseStrategy(ConnectionReuseStrategy strategy)strategy - LIFO 或 FIFO选择连接池应使用的连接复用策略FIFOconnection_reuse_strategy
setDefaultDatabase(String database)database - 数据库名称设置默认数据库。defaultdatabase

客户端标识​

查询日志中有两个字段用于标识发起请求的应用程序：client_name 和 http_user_agent。原生 TCP 协议使用 client_name 来标识应用程序。HTTP 协议使用 http_user_agent 来标识应用程序。Client builder 提供了 setClientName 方法，可为两种协议设置正确的值。 http_user_agent 字段按照 User-Agent 请求头的通用格式设置：application-name[/version] [(operating-system; architecture; ...)]。 这组值会在每一层重复出现：应用程序、客户端库、HTTP 客户端库。通过 setClientName 方法设置的内容在列表中排在最前面。

例如：

client.setClientName("my-app-01/1.0");

将得到如下 http_user_agent 值：

my-app-01/1.0 clickhouse-java-v2/0.9.6-SNAPSHOT (Linux; jvm:17.0.17) Apache-HttpClient/5.4.4

应用程序可以设置自定义 HTTP 请求头 User-Agent 来标识自身。但 clickhouse-java-v2/0.9.6-SNAPSHOT 部分将被追加到该请求头的末尾。

操作标识​

查询日志还包含两个字段 query_id 和 log_comment，可用于标识某个操作并为查询日志添加额外信息。

query_id 是操作的唯一标识符。应用程序可以通过调用 QuerySettings 类的 setQueryId 方法来设置它。

QuerySettings querySettings = new QuerySettings();
querySettings.setQueryId("some-query-id");

log_comment 是可以添加到查询日志的注释。可以在应用程序中通过调用 QuerySettings 类的 logComment 方法来设置。

QuerySettings querySettings = new QuerySettings();
querySettings.logComment("some-comment");

服务器设置​

服务器端设置可以在创建客户端时于客户端级别配置一次 (参见 Builder 的 serverSetting 方法) ，也可以在操作级别配置 (参见操作设置类的 serverSetting 方法) 。

 try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
        .setUsername("default")
        .setPassword(ClickHouseServerForTest.getPassword())
        .compressClientRequest(true)

        // Client level
        .serverSetting("max_threads", "10")
        .serverSetting("async_insert", "1")
        .serverSetting("roles", Arrays.asList("role1", "role2"))

        .build()) {

	// Operation level
	QuerySettings querySettings = new QuerySettings();
	querySettings.serverSetting("session_timezone", "Europe/Zurich");

	...
}

⚠️ 当通过 setOption 方法设置选项时 (无论是在 Client.Builder 还是在操作设置类中) ，服务器设置名称应以 clickhouse_setting_ 为前缀。在这种情况下，com.clickhouse.client.api.ClientConfigProperties#serverSetting() 可能会很有用。

自定义 HTTP 请求头​

可以在客户端级别为所有操作设置自定义 HTTP 请求头，也可以在操作级别仅为单个操作设置。


QuerySettings settings = new QuerySettings()
    .httpHeader(HttpHeaders.REFERER, clientReferer)
    .setQueryId(qId);

当通过 setOption 方法设置选项时 (无论是 Client.Builder 还是操作设置类) ，自定义头部名称应以 http_header_ 为前缀。此情况下，方法 com.clickhouse.client.api.ClientConfigProperties#httpHeader() 可能会很有用。

常用术语定义​

ClickHouseFormat​

支持的格式 的枚举类型，包含 ClickHouse 支持的所有格式。

  • raw - 用户需自行对原始数据进行转码
  • full - 客户端可以自行对数据进行转码，并接收原始数据流
  • - - ClickHouse 不支持对该格式执行此操作

此客户端版本支持：

格式输入输出
TabSeparatedrawraw
TabSeparatedRawrawraw
TabSeparatedWithNamesrawraw
TabSeparatedWithNamesAndTypesrawraw
TabSeparatedRawWithNamesrawraw
TabSeparatedRawWithNamesAndTypesrawraw
Templaterawraw
TemplateIgnoreSpacesraw-
CSVrawraw
CSVWithNamesrawraw
CSVWithNamesAndTypesrawraw
CustomSeparatedrawraw
CustomSeparatedWithNamesrawraw
CustomSeparatedWithNamesAndTypesrawraw
SQLInsert-raw
Valuesrawraw
Vertical-raw
JSONrawraw
JSONAsStringraw-
JSONAsObjectraw-
JSONStringsrawraw
JSONColumnsrawraw
JSONColumnsWithMetadatarawraw
JSONCompactrawraw
JSONCompactStrings-raw
JSONCompactColumnsrawraw
JSONEachRowrawraw
PrettyJSONEachRow-raw
JSONEachRowWithProgress-raw
JSONStringsEachRowrawraw
JSONStringsEachRowWithProgress-raw
JSONCompactEachRowrawraw
JSONCompactEachRowWithNamesrawraw
JSONCompactEachRowWithNamesAndTypesrawraw
JSONCompactStringsEachRowrawraw
JSONCompactStringsEachRowWithNamesrawraw
JSONCompactStringsEachRowWithNamesAndTypesrawraw
JSONObjectEachRowrawraw
BSONEachRowrawraw
TSKVrawraw
Pretty-raw
PrettyNoEscapes-raw
PrettyMonoBlock-raw
PrettyNoEscapesMonoBlock-raw
PrettyCompact-raw
PrettyCompactNoEscapes-raw
PrettyCompactMonoBlock-raw
PrettyCompactNoEscapesMonoBlock-raw
PrettySpace-raw
PrettySpaceNoEscapes-raw
PrettySpaceMonoBlock-raw
PrettySpaceNoEscapesMonoBlock-raw
Prometheus-raw
Protobufrawraw
ProtobufSinglerawraw
ProtobufListrawraw
Avrorawraw
AvroConfluentraw-
Parquetrawraw
ParquetMetadataraw-
Arrowrawraw
ArrowStreamrawraw
ORCrawraw
Oneraw*
Npyrawraw
RowBinaryfullfull
RowBinaryWithNamesfullfull
RowBinaryWithNamesAndTypesfullfull
RowBinaryWithDefaultsfull-
Nativefullraw
Null*raw
XML-原始
CapnProto原始原始
LineAsString原始raw
Regexpraw*
RawBLOBrawraw
MsgPackrawraw
MySQLDumpraw-
DWARFraw*
Markdown-raw
Formraw*

插入 API​

insert(String tableName, InputStream data, ClickHouseFormat format)​

接受以指定格式编码的字节输入流 (InputStream) 形式的数据。假定 data 已按 format 进行编码。

签名

CompletableFuture<InsertResponse> insert(String tableName, InputStream data, ClickHouseFormat format, InsertSettings settings)
CompletableFuture<InsertResponse> insert(String tableName, InputStream data, ClickHouseFormat format)

参数

tableName - 目标表的名称。

data - 已编码数据的输入流。

format - 数据编码所使用的格式。

settings - 请求设置。

返回值

InsertResponse 类型的 Future，包含操作结果以及服务器端指标等附加信息。

示例

try (InputStream dataStream = getDataStream()) {
    try (InsertResponse response = client.insert(TABLE_NAME, dataStream, ClickHouseFormat.JSONEachRow,
            insertSettings).get(3, TimeUnit.SECONDS)) {

        log.info("Insert finished: {} rows written", response.getMetrics().getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong());
    } catch (Exception e) {
        log.error("Failed to write JSONEachRow data", e);
        throw new RuntimeException(e);
    }
}

insert(String tableName, List<?> data, InsertSettings settings)​

向数据库发送写入请求。对象列表会被转换为高效的内部格式，然后发送到服务器。列表项所属的类应事先使用 register(Class, TableSchema) 方法进行注册。

签名

client.insert(String tableName, List<?> data, InsertSettings settings)
client.insert(String tableName, List<?> data)

参数

tableName - 目标表名称。

data - DTO (数据传输对象) 对象集合。

settings - 请求设置。

返回值

返回 InsertResponse 类型的 Future，包含操作结果以及服务器端指标等附加信息。

示例

// Important step (done once) - register class to pre-compile object serializer according to the table schema.
client.register(ArticleViewEvent.class, client.getTableSchema(TABLE_NAME));

List<ArticleViewEvent> events = loadBatch();

try (InsertResponse response = client.insert(TABLE_NAME, events).get()) {
    // handle response, then it will be closed and connection that served request will be released.
}

InsertSettings​

插入操作的配置选项。

配置方法

方法说明
setQueryId(String queryId)设置将要分配给该操作的查询 ID。默认值：null。
setDeduplicationToken(String token)设置去重令牌。该令牌将发送到服务器，并可用于标识查询。默认值：null。
setInputStreamCopyBufferSize(int size)复制缓冲区的大小。在写入操作期间，该缓冲区用于将用户提供的输入流中的数据复制到输出流。默认值：8196。
serverSetting(String name, String value)为操作设置单独的服务器设置。
serverSetting(String name, Collection values)为某个操作设置可包含多个值的单个服务器配置项。集合中的元素应为 String 值。
setDBRoles(Collection dbRoles)设置在执行操作前要生效的数据库角色。集合中的元素应为 String 值。
setOption(String option, Object value)以原始格式设置配置选项。这不是服务器设置。

InsertResponse​

用于保存插入操作结果的响应对象。仅当客户端从服务器收到响应时才可用。

注意

应尽快关闭该对象以释放连接，因为在完全读取完上一个响应的全部数据之前，该连接无法被复用。

方法描述
OperationMetrics getMetrics()返回一个包含该操作各项指标的对象。
String getQueryId()返回为该操作分配的查询 ID，该 ID 可由应用程序 (通过操作设置) 指定，或由服务器分配。

查询 API​

query(String sqlQuery)​

按原样发送 sqlQuery。响应格式由查询设置指定。QueryResponse 将持有对响应流的引用，该响应流应由能够处理该格式的读取器来消费。

签名

CompletableFuture<QueryResponse> query(String sqlQuery, QuerySettings settings)
CompletableFuture<QueryResponse> query(String sqlQuery)

参数

sqlQuery - 单条 SQL 语句。该查询按原样发送到服务器。

settings - 请求设置。

返回值

类型为 QueryResponse 的 Future —— 结果数据集以及服务器端指标等附加信息。使用完数据集后应关闭 Response 对象。

示例

final String sql = "select * from " + TABLE_NAME + " where title <> '' limit 10";

// Default format is RowBinaryWithNamesAndTypesFormatReader so reader have all information about columns
try (QueryResponse response = client.query(sql).get(3, TimeUnit.SECONDS);) {

    // Create a reader to access the data in a convenient way
    ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);

    while (reader.hasNext()) {
        reader.next(); // Read the next record from stream and parse it

        // get values
        double id = reader.getDouble("id");
        String title = reader.getString("title");
        String url = reader.getString("url");

        // collecting data
    }
} catch (Exception e) {
    log.error("Failed to read data", e);
}

// put business logic outside of the reading block to release http connection asap.

query(String sqlQuery, Map<String, Object> queryParams, QuerySettings settings)​

按原样发送 sqlQuery，并同时发送查询参数，以便服务器能够编译该 SQL 表达式。

签名

CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Object> queryParams, QuerySettings settings)

参数

sqlQuery - 带有占位符 {} 的 SQL 表达式。

queryParams - 变量到值的映射，用于在服务器端补全 SQL 表达式。

settings - 请求设置。

返回值

类型为 QueryResponse 的 Future —— 结果数据集以及服务器端指标等附加信息。使用完数据集后应关闭 Response 对象。

示例


// define parameters. They will be sent to the server along with the request.
Map<String, Object> queryParams = new HashMap<>();
queryParams.put("param1", 2);

try (QueryResponse response =
        client.query("SELECT * FROM " + table + " WHERE col1 >= {param1:UInt32}", queryParams, new QuerySettings()).get()) {

    // Create a reader to access the data in a convenient way
    ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);

    while (reader.hasNext()) {
        reader.next(); // Read the next record from stream and parse it

        // reading data
    }

} catch (Exception e) {
    log.error("Failed to read data", e);
}

queryAll(String sqlQuery)​

以 RowBinaryWithNamesAndTypes 格式查询数据。结果以集合形式返回。读取性能与使用 reader 相同，但需要更多内存来保存整个数据集。

签名

List<GenericRecord> queryAll(String sqlQuery)

参数

sqlQuery - 用于从服务器查询数据的 SQL 表达式。

返回值

以 GenericRecord 对象列表表示的完整数据集，这些对象以按行的方式提供对结果数据的访问。

示例

try {
    log.info("Reading whole table and process record by record");
    final String sql = "select * from " + TABLE_NAME + " where title <> ''";

    // Read whole result set and process it record by record
    client.queryAll(sql).forEach(row -> {
        double id = row.getDouble("id");
        String title = row.getString("title");
        String url = row.getString("url");

        log.info("id: {}, title: {}, url: {}", id, title, url);
    });
} catch (Exception e) {
    log.error("Failed to read data", e);
}

QuerySettings​

查询操作的配置选项。

配置方法

方法描述
setQueryId(String queryId)设置将分配给该操作的查询 ID。
setFormat(ClickHouseFormat format)设置响应格式。完整列表请参见 RowBinaryWithNamesAndTypes。
setMaxExecutionTime(Integer maxExecutionTime)设置操作在服务器上的执行时间，不影响读取超时。
waitEndOfQuery(Boolean waitEndOfQuery)请求服务器在发送响应之前等待查询结束。
setUseServerTimeZone(Boolean useServerTimeZone)服务端时区 (参见客户端配置) 将用于解析操作结果中的日期/时间类型值。默认值为 false。
setUseTimeZone(String timeZone)请求服务器使用 timeZone 进行时间转换。详见 session_timezone。
serverSetting(String name, String value)为操作设置单独的服务器配置项。
serverSetting(String name, Collection values)为某个操作设置可包含多个值的单个服务器配置项。集合中的元素应为 String 值。
setDBRoles(Collection dbRoles)设置在执行操作前要启用的数据库角色。集合中的元素应为 String 类型的值。
setOption(String option, Object value)以原始格式设置配置选项。这不是服务器端设置。

QueryResponse​

用于保存查询执行结果的响应对象。仅在客户端已从服务器收到响应时才可用。

注意

应尽快关闭该对象以释放连接，因为在完全读取完上一个响应的全部数据之前，该连接无法被复用。

方法描述
ClickHouseFormat getFormat()返回响应数据的编码格式。
InputStream getInputStream()返回按指定格式编码的未压缩数据字节流。
OperationMetrics getMetrics()返回包含该操作指标的对象。
String getQueryId()返回为该操作分配的查询 ID，该 ID 可由应用程序 (通过操作设置) 或服务器指定。
TimeZone getTimeZone()返回在处理响应中的 Date/DateTime 类型时应使用的时区。

示例​

通用 API​

getTableSchema(String table)​

获取 table 的表 schema。

签名

TableSchema getTableSchema(String table)
TableSchema getTableSchema(String table, String database)

参数

table - 要获取其 schema 数据的表名。

database - 定义目标表的数据库。

返回值

返回一个 TableSchema 对象，其中包含该表的列列表。

getTableSchemaFromQuery(String sql)​

从 SQL 语句中获取 schema。

签名

TableSchema getTableSchemaFromQuery(String sql)

参数

sql - "SELECT" SQL 语句，其 schema 将被返回。

返回值

返回一个列与 sql 表达式匹配的 TableSchema 对象。

TableSchema​

register(Class<?> clazz, TableSchema schema)​

为 Java 类编译一个序列化/反序列化层，以便在基于 schema 进行数据读写时使用。该方法将为 getter/setter 方法对及其对应的列创建序列化器和反序列化器。 列匹配是通过从方法名中提取列名来完成的。例如，getFirstName 将对应列 first_name 或 firstname。

签名

void register(Class<?> clazz, TableSchema schema)

参数

clazz - 表示用于读取/写入数据的 POJO 的 Class。

schema - 用于与 POJO 属性进行匹配的数据 schema。

示例

client.register(ArticleViewEvent.class, client.getTableSchema(TABLE_NAME));

使用示例​

完整示例代码存放在代码仓库的 'example` 目录 中：

  • client-v2 - 主要示例代码集。
  • demo-service - 在 Spring Boot 应用中使用该客户端的示例项目。
  • demo-kotlin-service - 示例代码，演示如何在 Ktor (Kotlin) 应用中使用该客户端。

读取数据​

读取数据通常有两种方式：

  • query() 方法返回底层的 QueryResponse 对象，该对象包含用于承载数据的 InputStream。通常与 ClickHouseBinaryFormatReader 搭配使用以进行流式读取，但 也可以与任意其他自定义读取器实现配合使用。QueryResponse 还提供对结果集元数据和指标的访问能力。
  • queryAll() 方法与 GenericRecord 搭配使用，可方便按行访问数据。在这种情况下，会将整个结果集加载到内存中。
  • queryRecords() 方法返回 com.clickhouse.client.api.query.Records —— 一个用于迭代 GenericRecord 对象的迭代器。该方法采用流式处理方式 (不会将数据加载到内存中) ，并通过 GenericRecord 来访问数据。

注意： 流式处理方式要求快速读取，否则由于数据直接从网络流读取，可能导致服务器写入超时。

读取数组​

ClickHouseBinaryFormatReader 方法

  • getList(...) - 将任意 Array(...) 读取为 List<T>。适合作为灵活类型读取的默认选项。支持嵌套数组。
  • getByteArray(...), getShortArray(...), getIntArray(...), getLongArray(...), getFloatArray(...), getDoubleArray(...), getBooleanArray(...) - 最适合读取元素类型与 Java 原始类型兼容的一维数组。
  • getStringArray(...) - 用于读取 Array(String) (以及以名称表示的枚举值) 。
  • getObjectArray(...) - 针对任意 Array(...) 元素类型 (包括嵌套数组) 的通用方法。用于读取包含 Nullable 值和嵌套数组的数组。

所有方法都提供基于索引和基于名称的重载。索引从 1 开始计数。基于索引的重载会直接访问对应的列。 基于名称的方法每次调用时都需要执行一次索引查找。

try (QueryResponse response = client.query("SELECT * FROM my_table").get()) {
    ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);
    while (reader.next() != null) {

        Object[] uint64 = reader.getObjectArray("uint64_arr"); // Array(UInt64) -> BigInteger[]
        Object[] arr2d = reader.getObjectArray("arr2d");       // Array(Array(Int64)) -> Object[]

        // nested arrays are returned as nested Object[]:
        Object[] firstInner = (Object[]) arr2d[0];
        Long firstValue = (Long) firstInner[0];
    }
}

GenericRecord 方法

  • getList(...) - 将任意 Array(...) 读取为 List<T>。适合作为灵活类型读取的默认选项。支持嵌套数组。
  • getByteArray(...), getShortArray(...), getIntArray(...), getLongArray(...), getFloatArray(...), getDoubleArray(...), getBooleanArray(...) - 最适合用于元素值与 Java 原始类型兼容的一维数组。
  • getStringArray(...) - 用于 Array(String) (以及以名称表示的枚举值) 。
  • getObjectArray(...) - 针对任意 Array(...) 元素类型 (包括嵌套数组) 的通用方法。用于读取包含 Nullable 值和嵌套数组的数组。

所有方法都提供基于索引和基于名称的重载。索引从 1 开始计数。基于索引的重载会直接访问对应的列。 基于名称的方法每次调用时都需要执行一次索引查找。

try (QueryResponse response = client.query("SELECT * FROM my_table").get()) {
    List<GenericRecord> rows = client.queryAll(
        "SELECT int_arr, arr2d_nullable FROM test_arrays ORDER BY id");

    for (GenericRecord row : rows) {
        Object[] intArr = row.getObjectArray("int_arr");                 // Array(Int32) -> Integer[]
        Object[] arr2d = row.getObjectArray("arr2d_nullable");           // Array(Array(Nullable(Int32)))

        Object[] inner = (Object[]) arr2d[0];
        Object maybeNull = inner[1]; // may be null
    }
}

迁移指南​

旧版客户端 (V1) 以 com.clickhouse.client.ClickHouseClient#builder 作为入口点。新版客户端 (V2) 采用类似的模式，使用 com.clickhouse.client.api.Client.Builder。主要区别如下：

  • 不使用 service loader 来获取实现。com.clickhouse.client.api.Client 是一个外观类，未来可用于各类实现。
  • 配置来源更少了：一种由 builder 提供，另一种通过操作设置 (QuerySettings、InsertSettings) 提供。上一版本支持按节点配置，并且在某些情况下还会加载 环境变量。

配置参数对照​

V1 中有 3 个与配置相关的枚举类：

  • com.clickhouse.client.config.ClickHouseDefaults - 在大多数使用场景下通常需要设置的配置参数，例如 USER 和 PASSWORD。
  • com.clickhouse.client.config.ClickHouseClientOption - 客户端专用的配置参数，例如 HEALTH_CHECK_INTERVAL。
  • com.clickhouse.client.http.config.ClickHouseHttpOption - HTTP 接口专用的配置参数。例如 RECEIVE_QUERY_PROGRESS。

它们的设计初衷是对参数进行分组并提供清晰的分隔。然而在某些情况下，这会导致混淆 (例如，com.clickhouse.client.config.ClickHouseDefaults#ASYNC 与 com.clickhouse.client.config.ClickHouseClientOption#ASYNC 之间是否存在区别) 。新的 V2 客户端使用 com.clickhouse.client.api.Client.Builder 作为所有可能的客户端配置选项的统一字典。com.clickhouse.client.api.ClientConfigProperties 中列出了所有配置参数名称。

下表列出了新客户端中支持的旧版选项及其新含义。

图例： ✔ = 支持，✗ = 已移除

V1 ConfigurationV2 Builder MethodComments
ClickHouseDefaults#HOSTClient.Builder#addEndpoint
ClickHouseDefaults#PROTOCOL✗Only HTTP supported in V2
ClickHouseDefaults#DATABASE
ClickHouseClientOption#DATABASE		Client.Builder#setDefaultDatabase
ClickHouseDefaults#USERClient.Builder#setUsername
ClickHouseDefaults#PASSWORDClient.Builder#setPassword
ClickHouseClientOption#CONNECTION_TIMEOUTClient.Builder#setConnectTimeout
ClickHouseClientOption#CONNECTION_TTLClient.Builder#setConnectionTTL
ClickHouseHttpOption#MAX_OPEN_CONNECTIONSClient.Builder#setMaxConnections
ClickHouseHttpOption#KEEP_ALIVE
ClickHouseHttpOption#KEEP_ALIVE_TIMEOUT		Client.Builder#setKeepAliveTimeout
ClickHouseHttpOption#CONNECTION_REUSE_STRATEGYClient.Builder#setConnectionReuseStrategy
ClickHouseHttpOption#USE_BASIC_AUTHENTICATIONClient.Builder#useHTTPBasicAuth

主要差异​

  • Client V2 uses less proprietary classes to increase portability. For example, V2 works with any implementation of java.io.InputStream for writing data to a server.
  • Client V2 的 async 设置默认是 off。这意味着不会创建额外线程，应用程序对客户端拥有更多控制权。在大多数使用场景下，此设置都应保持为 off。启用 async 会为每个请求创建一个单独的线程。只有在使用由应用程序控制的 executor 时，这样做才有意义 (参见 com.clickhouse.client.api.Client.Builder#setSharedOperationExecutor)

写入数据​

  • 使用 java.io.InputStream 的任意实现。支持 V1 的 com.clickhouse.data.ClickHouseInputStream，但不推荐使用。
  • 检测到输入流结束后，会自动进行相应处理。此前则需要关闭请求的输出流。

V1 以 TSV 格式插入数据。

InputStream inData = getInData();
ClickHouseRequest.Mutation request = client.read(server)
        .write()
        .table(tableName)
        .format(ClickHouseFormat.TSV);
ClickHouseConfig config = request.getConfig();
CompletableFuture<ClickHouseResponse> future;
try (ClickHousePipedOutputStream requestBody = ClickHouseDataStreamFactory.getInstance()
        .createPipedOutputStream(config)) {
    // start the worker thread which transfer data from the input into ClickHouse
    future = request.data(requestBody.getInputStream()).execute();

    // Copy data from inData stream to requestBody stream

    // We need to close the stream before getting a response
    requestBody.close();

    try (ClickHouseResponse response = future.get()) {
        ClickHouseResponseSummary summary = response.getSummary();
        Assert.assertEquals(summary.getWrittenRows(), numRows, "Num of written rows");
    }
}

V2 以 TSV 格式插入数据。

InputStream inData = getInData();
InsertSettings settings = new InsertSettings().setInputStreamCopyBufferSize(8198 * 2); // set copy buffer size
try (InsertResponse response = client.insert(tableName, inData, ClickHouseFormat.TSV, settings).get(30, TimeUnit.SECONDS)) {

  // Insert is complete at this point

} catch (Exception e) {
 // Handle exception
}
  • 只需调用一个方法。无需另外创建请求对象。
  • 当所有数据复制完成后，请求体流会自动关闭。
  • new low-level API is available com.clickhouse.client.api.Client#insert(java.lang.String, java.util.List<java.lang.String>, com.clickhouse.client.api.DataStreamWriter, com.clickhouse.data.ClickHouseFormat, com.clickhouse.client.api.insert.InsertSettings)。com.clickhouse.client.api.DataStreamWriter 用于实现自定义数据写入逻辑，例如从 队列读取数据。

读取数据​

  • 默认情况下，数据以 RowBinaryWithNamesAndTypes 格式读取。当需要进行数据绑定时，目前仅支持此格式。
  • 可以使用 List<GenericRecord> com.clickhouse.client.api.Client#queryAll(java.lang.String) 方法将数据读取为一组记录。该方法会将数据读入内存并自动释放连接，无需额外处理。GenericRecord 提供访问结果数据的接口，并实现了一些类型转换。
Collection<GenericRecord> records = client.queryAll("SELECT * FROM table");
for (GenericRecord record : records) {
    int rowId = record.getInteger("rowID");
    String name = record.getString("name");
    LocalDateTime ts = record.getLocalDateTime("ts");
}

用于通过其协议与数据库服务器通信的 Java 客户端库。当前实现仅支持 HTTP 接口。该库提供自己的 API，用于向服务器发送请求。

弃用

该库即将被弃用。对于新项目，请使用最新的 Java 客户端

配置​

<!-- https://mvnrepository.com/artifact/com.clickhouse/clickhouse-http-client -->
<dependency>
    <groupId>com.clickhouse</groupId>
    <artifactId>clickhouse-http-client</artifactId>
    <version>0.7.2</version>
</dependency>

自 0.5.0 版本起，该驱动程序使用新的 HTTP 客户端库，需要将其作为依赖添加。

<!-- https://mvnrepository.com/artifact/org.apache.httpcomponents.client5/httpclient5 -->
<dependency>
    <groupId>org.apache.httpcomponents.client5</groupId>
    <artifactId>httpclient5</artifactId>
    <version>5.3.1</version>
</dependency>

初始化​

连接 URL 的格式如下：protocol://host[:port][/database][?param[=value][&param[=value]][#tag[,tag]]，例如：

  • http://localhost:8443?ssl=true&sslmode=NONE
  • https://(https://explorer@play.clickhouse.com:443

连接到单个节点：

ClickHouseNode server = ClickHouseNode.of("http://localhost:8123/default?compress=0");

连接到多节点集群：

ClickHouseNodes servers = ClickHouseNodes.of(
    "jdbc:ch:http://server1.domain,server2.domain,server3.domain/my_db"
    + "?load_balancing_policy=random&health_check_interval=5000&failover=2");

查询 API​

try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
     ClickHouseResponse response = client.read(servers)
        .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
        .query("select * from numbers limit :limit")
        .params(1000)
        .executeAndWait()) {
            ClickHouseResponseSummary summary = response.getSummary();
            long totalRows = summary.getTotalRowsToRead();
}

流式查询 API​

try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
     ClickHouseResponse response = client.read(servers)
        .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
        .query("select * from numbers limit :limit")
        .params(1000)
        .executeAndWait()) {
            for (ClickHouseRecord r : response.records()) {
            int num = r.getValue(0).asInteger();
            // type conversion
            String str = r.getValue(0).asString();
            LocalDate date = r.getValue(0).asDate();
        }
}

请参阅代码仓库中的完整代码示例。

插入 API​


try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
     ClickHouseResponse response = client.read(servers).write()
        .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
        .query("insert into my_table select c2, c3 from input('c1 UInt8, c2 String, c3 Int32')")
        .data(myInputStream) // `myInputStream` is source of data in RowBinary format
        .executeAndWait()) {
            ClickHouseResponseSummary summary = response.getSummary();
            summary.getWrittenRows();
}

请参阅代码仓库中的完整代码示例。

RowBinary 编码

RowBinary 格式在其页面中有说明。

可参考以下代码示例。

功能​

压缩​

默认情况下，客户端使用 LZ4 压缩，需要添加以下依赖：

<!-- https://mvnrepository.com/artifact/org.lz4/lz4-java -->
<dependency>
    <groupId>org.lz4</groupId>
    <artifactId>lz4-java</artifactId>
    <version>1.8.0</version>
</dependency>

您也可以在连接 URL 中将 compress_algorithm 参数设置为 gzip，以改用 gzip 压缩。

此外，您还可以通过以下方式禁用压缩。

  1. 在连接 URL 中将 compress 设置为 0 来禁用压缩：http://localhost:8123/default?compress=0
  2. 通过客户端配置禁用：
ClickHouseClient client = ClickHouseClient.builder()
   .config(new ClickHouseConfig(Map.of(ClickHouseClientOption.COMPRESS, false)))
   .nodeSelector(ClickHouseNodeSelector.of(ClickHouseProtocol.HTTP))
   .build();

请参阅压缩文档，以进一步了解不同的压缩选项。

多条查询​

在同一会话中，在一个工作线程内按顺序执行多条查询：

CompletableFuture<List<ClickHouseResponseSummary>> future = ClickHouseClient.send(servers.apply(servers.getNodeSelector()),
    "create database if not exists my_base",
    "use my_base",
    "create table if not exists test_table(s String) engine=Memory",
    "insert into test_table values('1')('2')('3')",
    "select * from test_table limit 1",
    "truncate table test_table",
    "drop table if exists test_table");
List<ClickHouseResponseSummary> results = future.get();

命名参数​

您可以按名称传递参数，而不必仅依赖它们在参数列表中的位置。该功能可通过 params 函数实现。

try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
     ClickHouseResponse response = client.read(servers)
        .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
        .query("select * from my_table where name=:name limit :limit")
        .params("Ben", 1000)
        .executeAndWait()) {
            //...
        }
}
参数

所有涉及 String 类型的 params 签名 (String、String[]、Map<String, String>) 均假定传入的键是有效的 ClickHouse SQL 字符串。例如：

try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
     ClickHouseResponse response = client.read(servers)
        .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
        .query("select * from my_table where name=:name")
        .params(Map.of("name","'Ben'"))
        .executeAndWait()) {
            //...
        }
}

如果您不想手动将 String 对象解析为 ClickHouse SQL，可以使用位于 com.clickhouse.data 的辅助函数 ClickHouseValues.convertToSqlExpression：

try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
     ClickHouseResponse response = client.read(servers)
        .format(ClickHouseFormat.RowBinaryWithNamesAndTypes)
        .query("select * from my_table where name=:name")
        .params(Map.of("name", ClickHouseValues.convertToSqlExpression("Ben's")))
        .executeAndWait()) {
            //...
        }
}

在上述示例中，ClickHouseValues.convertToSqlExpression 会对内部的单引号进行转义，并在变量两侧加上一对有效的单引号。

其他类型 (如 Integer、UUID、Array 和 Enum) 会在 params 中自动转换。

节点发现​

Java 客户端支持自动发现 ClickHouse 节点。自动发现功能默认禁用。要手动启用该功能，请将 auto_discovery 设置为 true：

properties.setProperty("auto_discovery", "true");

或者在连接 URL 中：

jdbc:ch://my-server/system?auto_discovery=true

启用自动发现后，无需在连接 URL 中指定所有 ClickHouse 节点。URL 中指定的节点将作为种子节点，Java 客户端会自动从系统表以及 clickhouse-keeper 或 zookeeper 中发现更多节点。

以下选项用于配置自动发现功能：

属性默认值说明
auto_discoveryfalse是否允许客户端从系统表和/或 clickhouse-keeper/zookeeper 中发现更多节点。
node_discovery_interval0节点发现间隔 (以毫秒为单位) ；零值或负值表示仅发现一次。
node_discovery_limit100单次可发现的最大节点数；值为 0 或负数表示不限制。

负载均衡​

Java 客户端会根据负载均衡策略选择 ClickHouse 节点来发送请求。一般来说，负载均衡策略负责以下工作：

  1. 从托管的节点列表中获取一个节点。
  2. 管理节点的状态。
  3. (可选) 如果启用了自动发现，则调度一个用于节点发现的后台进程并执行健康检查。

以下是可用于配置负载均衡的一组选项：

属性默认值描述
load_balancing_policy""负载均衡策略可以是以下之一：
  • firstAlive - 请求将被发送到受管节点列表中第一个健康节点
  • random - 请求将被发送到受管节点列表中随机选取的一个节点
  • roundRobin - 请求将按照轮询方式依次发送到受管节点列表中的每个节点
  • 实现 ClickHouseLoadBalancingPolicy 的完全限定类名 - 自定义负载均衡策略
    • 如果未指定该策略，请求将被发送到受管节点列表中的第一个节点
    load_balancing_tags""用于过滤节点的负载均衡标签。请求仅发送至具有指定标签的节点。
    health_check_interval0健康检查间隔 (以毫秒为单位) ，零或负值表示仅执行一次。
    health_check_methodClickHouseHealthCheckMethod.SELECT_ONE健康检查方法。可以为以下之一：
  • ClickHouseHealthCheckMethod.SELECT_ONE - 通过执行 select 1 查询进行检查
  • ClickHouseHealthCheckMethod.PING - 协议级检查，通常更快
    • node_check_interval0节点检查间隔，单位为毫秒；负数视为 0。仅当距上次检查已过去指定时间时，才会检查该节点的状态。
    health_check_interval 与 node_check_interval 的区别在于：health_check_interval 会调度后台作业来检查一组节点 (全部节点或故障节点) 的状态，而 node_check_interval 指定的是某个特定节点距离上次检查需间隔多长时间后才会再次检查。
    check_all_nodesfalse是否对所有节点执行健康检查，还是仅对故障节点进行检查。

    故障切换和重试​

    Java 客户端提供一些配置选项，用于设置故障切换和失败查询的重试行为：

    属性默认值描述
    故障转移0单个请求可进行故障转移的最大次数。零或负值表示不进行故障转移。发生故障转移时，会将失败的请求发送到其他节点 (根据负载均衡策略) ，以便从故障中恢复。
    重试0单个请求允许重试的最大次数。为零或负值时表示不进行重试。仅当 ClickHouse 服务器返回 NETWORK_ERROR 错误码时，才会在同一节点上对该请求进行重试。
    repeat_on_session_locktrue当会话被锁定时，是否重复执行直到会话超时 (根据 session_timeout 或 connect_timeout) 。如果 ClickHouse 服务端返回 SESSION_IS_LOCKED 错误码，则会对该失败请求进行重试。

    添加自定义 HTTP 请求头​

    Java 客户端在使用 HTTP/S 传输层时支持为请求添加自定义 HTTP 头。 应使用 custom_http_headers 属性，多个请求头使用 , 分隔，单个请求头的键和值使用 = 分隔。

    Java 客户端支持​

    options.put("custom_http_headers", "X-ClickHouse-Quota=test, X-ClickHouse-Test=test");

    JDBC 驱动​

    properties.setProperty("custom_http_headers", "X-ClickHouse-Quota=test, X-ClickHouse-Test=test");