Data Gateway 客户端

DataGatewayClient 是一个附着在 查询网关 上的 Java 库。与其分别通过驱动连接到 PostgreSQL · Aster TSDB · Spark · 时序引擎，不如向 一个网关 发送查询，并以 JSON 格式接收结果。

项	值
坐标	`plantpulse-platform:plantpulse-data-gateway-client`
服务器	`plantpulse-data-gateway` — 默认端口 5501（TLS）
要求	Java 21 及以上 — 客户端使用 21 编译（低于该版本将 `UnsupportedClassVersionError`）
认证	API 密钥

四种查询语言

网关根据语言选择背后的引擎。你无需知道数据在哪里。

方法	语言	背后引擎	使用场景
`pql()`	PQL	PostgreSQL	主数据 · 配置（`mm_*` 表）
`cql()`	CQL	Aster TSDB	时序原始数据 · 聚合（`tm_*` 表）
`sql()`	SQL	Spark	湖仓 · 大规模分析
`tql()`	TQL	时序引擎	时间窗口聚合 · 缺口填充
`cache()`	—	Valkey/Redis	当前值（`PREFIX:ID` 形式）

开始使用

import plantpulse.data.gateway.client.DataGatewayClient;
import plantpulse.data.gateway.client.Protocol;
import plantpulse.json.JSONArray;
import plantpulse.json.JSONObject;

// AutoCloseable — try-with-resources 가 연결을 닫습니다.
try (DataGatewayClient client = DataGatewayClient.builder()
        .protocol(Protocol.HTTPS)
        .host("plant-server.example.com")
        .port(5501)
        .build()) {

    // 한 번 인증하면 이후 호출이 이 키를 답니다.
    client.connect(System.getenv("PP_DATA_GATEWAY_API_KEY"));

    // 언어를 고르면 자기 자신을 돌려주므로 query() 로 이어집니다.
    JSONArray databases = client.sql().query("SHOW DATABASES");

    JSONObject params = new JSONObject();
    params.put("tag_id:string", "TAG-0001");
    params.put("limit:int", 10);
    JSONArray points = client.pql().query(
            "SELECT * FROM mm_point WHERE tag_id = :tag_id LIMIT :limit", params);

    JSONObject cached = client.cache("TAG:TAG-0001");
}

参数类型 — `이름:타입`

名称后的后缀决定了 网关如何转换类型。通过 X-Params 请求头传递。

类型	Java	示例
`int`	Integer	`params.put("count:int", 10)`
`long`	Long	`params.put("id:long", 9999999999L)`
`double` · `float`	Double · Float	`params.put("temp:float", 36.5f)`
`string`	String	`params.put("name:string", "abc")`
`epoch` · `ts-long`	时间戳(ms)	`params.put("ts:epoch", 1700000000000L)`
`ts-string`	`yyyy-MM-dd HH:mm:ss`	`params.put("ts:ts-string", "2026-01-01 00:00:00")`

不要用字符串拼接来构建查询

这正是后缀参数存在的原因。将值拼接为字符串会破坏类型，并容易受到注入攻击。

查询之外的功能

// 단일 행
JSONObject one = client.get("SELECT * FROM mm_tag WHERE id = :id",
        new JSONObject().put("id:int", 1));

// 갱신 — {"count": N} 을 돌려줍니다
JSONObject r = client.sql().update(
        "UPDATE mm_tag SET name = :name WHERE id = :id",
        new JSONObject().put("name", "test").put("id:int", 1));

// 비동기 — 같은 패턴으로 queryAsync · cacheAsync · updateAsync
CompletableFuture<JSONArray> f = client.queryAsync(QueryLanguage.PQL,
        "SELECT * FROM mm_tag", null);
JSONArray rows = f.get(10, TimeUnit.SECONDS);

处理失败

429 · 502 · 503 · 504 触发 客户端自动指数退避重试。其他失败将抛出异常。

异常	何时
`HttpResponseException`	HTTP 错误响应 — 包含 `statusCode` · `errorCode` · `message`
`ConnectionException`	连接失败
`DataGatewayClientException`	响应解析失败等库通用错误

try {
    JSONArray result = client.pql().query("SELECT 1");
} catch (HttpResponseException ex) {
    System.err.println("HTTP " + ex.getStatusCode() + ": " + ex.getErrorCode());
} catch (DataGatewayClientException ex) {
    System.err.println("client error: " + ex.getMessage());
}

网关拒绝的内容

DDL · 一次多个语句 · TQL 更新都会被 服务器拒绝。 更改模式的操作不应通过此路径进行。

另请参阅

CEP 客户端 · Data Gateway 组件 · 控制台登录

四种查询语言​

开始使用​

参数类型 — 이름:타입​

查询之外的功能​

处理失败​

另请参阅​