在 Java 应用中集成 ksqlDB 客户端执行流式查询，能够借助 ksqlDB 的流式 SQL 能力快速处理 Kafka 中的实时数据流，无需手动编写复杂的流处理逻辑。这种方式适合需要实时统计、实时过滤、实时关联等场景的 Java 项目。
环境准备与依赖引入
首先需要在 Java 项目中引入 ksqlDB 客户端的官方依赖，目前主流的构建工具都支持对应的依赖配置。如果使用 Maven 管理项目，可以在 pom.xml 中添加如下依赖：
<dependency>
<groupId>io.confluent.ksql</groupId>
<artifactId>ksqldb-api-client</artifactId>
<version>7.5.0</version>
</dependency>
如果使用 Gradle 构建项目，则在 build.gradle 中添加对应依赖：
implementation 'io.confluent.ksql:ksqldb-api-client:7.5.0'
注意版本需要和部署的 ksqlDB 服务端版本保持兼容，避免接口不匹配导致调用失败。
建立 ksqlDB 客户端连接
使用客户端前需要先配置连接参数，核心是指定 ksqlDB 服务端的地址，默认端口是 8088。创建客户端的代码示例如下：
import io.confluent.ksql.api.client.Client;
import io.confluent.ksql.api.client.ClientOptions;
import io.confluent.ksql.api.client.KsqlDBClient;
public class KsqlDBStreamQueryDemo {
public static void main(String[] args) {
// 配置客户端选项，指定 ksqlDB 服务端地址
ClientOptions options = ClientOptions.create()
.setHost("127.0.0.1")
.setPort(8088)
.setUseTls(false); // 本地测试不使用 TLS 加密
// 创建客户端实例
Client client = KsqlDBClient.create(options);
// 后续查询操作使用这个 client 实例
}
}
执行流式查询的完整流程
1. 发送流式查询语句
流式查询需要使用 CREATE STREAM 或者 SELECT 语句，其中 SELECT 查询如果要持续获取结果，需要配合 EMIT CHANGES 关键字。首先可以先执行一个创建流的语句，确保有可查询的流对象：
import io.confluent.ksql.api.client.ExecutionResult;
import java.util.concurrent.ExecutionException;
public class KsqlDBStreamQueryDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ClientOptions options = ClientOptions.create()
.setHost("127.0.0.1")
.setPort(8088)
.setUseTls(false);
Client client = KsqlDBClient.create(options);
// 创建测试流，假设已经有名为 test_topic 的 Kafka 主题
String createStreamSql = "CREATE STREAM IF NOT EXISTS test_stream (id INT, name STRING, score DOUBLE) WITH (KAFKA_TOPIC='test_topic', VALUE_FORMAT='JSON');";
ExecutionResult createResult = client.executeStatement(createStreamSql).get();
System.out.println("创建流结果：" + createResult.queryId());
}
}
2. 执行流式查询并获取结果
执行带 EMIT CHANGES 的查询语句，客户端会持续接收流数据的变化结果，需要注册回调来处理每一批返回的数据：
import io.confluent.ksql.api.client.*;
import java.util.concurrent.ExecutionException;
public class KsqlDBStreamQueryDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ClientOptions options = ClientOptions.create()
.setHost("127.0.0.1")
.setPort(8088)
.setUseTls(false);
Client client = KSQLDBClient.create(options);
// 先创建流
String createStreamSql = "CREATE STREAM IF NOT EXISTS test_stream (id INT, name STRING, score DOUBLE) WITH (KAFKA_TOPIC='test_topic', VALUE_FORMAT='JSON');";
client.executeStatement(createStreamSql).get();
// 执行流式查询，持续获取流数据变化
String querySql = "SELECT id, name, score FROM test_stream WHERE score > 60 EMIT CHANGES;";
client.streamQuery(querySql)
.thenAccept(streamedQueryResult -> {
// 处理查询结果的元数据
System.out.println("查询ID：" + streamedQueryResult.queryId());
// 注册行回调，每收到一批数据就触发
streamedQueryResult.onRow(row -> {
try {
int id = row.getInt("id");
String name = row.getString("name");
double score = row.getDouble("score");
System.out.println("收到流数据：id=" + id + ", name=" + name + ", score=" + score);
} catch (Exception e) {
e.printStackTrace();
}
});
})
.exceptionally(throwable -> {
System.err.println("查询执行失败：" + throwable.getMessage());
return null;
});
// 保持主线程运行，避免程序退出导致流查询中断
Thread.sleep(Long.MAX_VALUE);
}
}
常见问题与注意事项
- 流式查询会持续占用连接资源，如果不需要继续接收数据，需要调用
streamedQueryResult.close()方法主动关闭查询，释放资源。 - 查询语句中的流名称、字段名称需要和 ksqlDB 中已定义的元数据一致，否则会抛出语句解析异常。
- 如果 ksqlDB 服务端开启了认证，需要在
ClientOptions中配置对应的用户名和密码，否则连接会被拒绝。 - 流查询的结果默认是增量变化，如果需要获取全量数据后再监听变化，可以在查询语句中添加合适的窗口配置。
结果处理优化建议
在实际生产环境中，流查询返回的数据量可能较大，建议不要在行回调中执行耗时的业务逻辑，可以将数据先放入阻塞队列，再由单独的线程池处理，避免阻塞流数据的接收。同时需要添加异常重试机制，当连接断开时自动重连并重新执行查询，保证流式查询的稳定性。