数据服务

Https协议调用数据服务

适用于需要一次性获取完整结果集的标准查询场景。

import com.alibaba.fastjson.JSON;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class HttpsCallDataServiceApi {

    private static final HttpClient httpClient = HttpClient.newBuilder()
            .version(HttpClient.Version.HTTP_1_1)
            .connectTimeout(Duration.ofSeconds(10))
            .build();

    public static void main(String[] args) {
        try {
            String response = callDmsApi();
            System.out.println("响应内容: " + response);
        } catch (Exception e) {
            System.err.println("请求失败: " + e.getMessage());
            e.printStackTrace();
        }
    }

    public static String callDmsApi() throws Exception {
        // 替换为各region具体的服务地址，具体可见数据服务首页
        String url = "https://onequery-cn-hangzhou.proxy.dms.aliyuncs.com/service/http";

        Map<String, Object> requestBody = new HashMap<>();
        // 将 ${accessKey}、${accessSecret} 替换为用户的凭证信息，即DMS内部的凭证，而非RAM AK
        requestBody.put("accessKey", "${accessKey}");
        requestBody.put("accessSecret", "${accessSecret}");
        // 替换 ${apiPath} 为具体的api路径
        requestBody.put("query", "call api ${apiPath}");

        // 构建 params，若参数为空，则无需传递此参数
        Map<String, Object> params = new HashMap<>();
        // 在实际调用中，需将${param}替换为实际的参数
        // 数字类型参数
        params.put("userId", 111);
        // 字符串类型参数
        params.put("name", "abc");
        // 布尔类型参数
        params.put("isActive", true);
        // 列表类型参数
        params.put("tags", Arrays.asList(1, 2, 3));

        // 若参数为空，则无需put
        requestBody.put("params", params);

        String jsonBody = JSON.toJSONString(requestBody);

        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(url))
                .header("Content-Type", "application/json;charset=utf-8")
                .method("POST", HttpRequest.BodyPublishers.ofString(jsonBody))
                .timeout(Duration.ofSeconds(30))
                .build();

        HttpResponse<String> response = httpClient.send(request,
                HttpResponse.BodyHandlers.ofString());

        // 检查响应状态码
        if (response.statusCode() != 200) {
            throw new RuntimeException("HTTP error code: " + response.statusCode());
        }

        return response.body();
    }
}

调用成功后，会返回一个包含完整结果集的 JSON 对象。

{
  "attributes": {
    "QueryTime": 13
  },
  "columnMetas": [
    {
      "autoIncrement": true,
      "catalogName": "test_schema",
      "columnLabel": "id",
      "columnName": "id",
      "columnTypeName": "BIGINT UNSIGNED",
      "nullable": false,
      "precision": 20,
      "scale": 0,
      "schemaName": "",
      "tableName": "test_table"
    },
    {
      "autoIncrement": false,
      "catalogName": "test_schema",
      "columnLabel": "gmt_create",
      "columnName": "gmt_create",
      "columnTypeName": "DATETIME",
      "nullable": true,
      "precision": 19,
      "scale": 0,
      "schemaName": "",
      "tableName": "test_table"
    }
  ],
  "count": 1,
  "message": null,
  "rows": [
    {
      "id": 1,
      "gmt_create": "2025-12-17 16:23:53"
    }
  ],
  "sessionId": 1560935281,
  "traceId": "HZ1560935281OCQJOTKW",
  "type": "RESULTSET"
}

SSE协议调用数据服务

适用于长查询或需要逐步获取结果的场景。客户端将收到一个包含多个事件的消息流。

import com.alibaba.fastjson.JSON;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Stream;

public class SseCallDataServiceApi {

    private static final HttpClient httpClient = HttpClient.newBuilder()
            .version(HttpClient.Version.HTTP_1_1)
            .connectTimeout(Duration.ofSeconds(10))
            .build();

    public static void main(String[] args) {
        try {
            callDmsApiSse();
        } catch (Exception e) {
            System.err.println("SSE 请求失败: " + e.getMessage());
            e.printStackTrace();
        }
    }

    public static void callDmsApiSse() throws Exception {
        // 替换为各region具体的服务地址，具体可见数据服务首页
        String url = "https://onequery-cn-hangzhou.proxy.dms.aliyuncs.com/service/sse";

        Map<String, Object> requestBody = new HashMap<>();
        // 将 ${accessKey}、${accessSecret} 替换为用户的凭证信息，即DMS内部的凭证，而非RAM AK
        requestBody.put("accessKey", "${accessKey}");
        requestBody.put("accessSecret", "${accessSecret}");
        // 替换 ${apiPath} 为具体的api路径
        requestBody.put("query", "call api ${apiPath}");

        // 构建 params，若参数为空，则无需传递该参数
        Map<String, Object> params = new HashMap<>();
        // 在实际调用中，需将${param}替换为实际的参数
        // 数字类型参数
        params.put("userId", 111);
        // 字符串类型参数
        params.put("name", "abc");
        // 布尔类型参数
        params.put("isActive", true);
        // 列表类型参数
        params.put("tags", Arrays.asList(1, 2, 3));
        // 若参数为空，则无需put
        requestBody.put("params", params);

        String jsonBody = JSON.toJSONString(requestBody);

        // 构建 HttpRequest，添加 SSE 必要的 Header
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(url))
                .header("Content-Type", "application/json;charset=utf-8")
                .header("Accept", "text/event-stream") // 指定接受 SSE 流
                .header("Cache-Control", "no-cache")
                .method("POST", HttpRequest.BodyPublishers.ofString(jsonBody))
                .timeout(Duration.ofMinutes(5)) // SSE 通常是长连接，超时时间设长一点
                .build();

        // 发送请求并使用 ofLines 处理流式响应
        System.out.println("开始接收 SSE 流...");

        httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofLines())
                .thenAccept(response -> {
                    if (response.statusCode() != 200) {
                        System.err.println("HTTP 错误: " + response.statusCode());
                        return;
                    }

                    // 逐行解析响应内容
                    try (Stream<String> lines = response.body()) {
                        lines.forEach(line -> {
                            if (line.startsWith("data:")) {
                                // 提取 data: 之后的内容
                                String data = line.substring(5).trim();
                                System.out.println("收到数据: " + data);
                            } else if (line.startsWith("error:")) {
                                System.err.println("收到错误: " + line);
                            }
                        });
                    }
                })
                .join(); // 为了演示效果，主线程等待异步执行完成
    }
}

SSE 响应是一个事件流，每个事件都以 data: 开头，其后跟随一个 JSON 对象。您需要关注该 JSON 对象中的 type 字段来判断事件类型。

各类事件的 JSON 示例：

• META 事件：

  {
    "attributes": {
      "QueryTime": 13
    },
    "columnMetas": [
      {
        "autoIncrement": true,
        "catalogName": "test_schema",
        "columnLabel": "id",
        "columnName": "id",
        "columnTypeName": "BIGINT UNSIGNED",
        "nullable": false,
        "precision": 20,
        "scale": 0,
        "schemaName": "",
        "tableName": "test_table"
      },
      {
        "autoIncrement": false,
        "catalogName": "test_schema",
        "columnLabel": "gmt_create",
        "columnName": "gmt_create",
        "columnTypeName": "DATETIME",
        "nullable": true,
        "precision": 19,
        "scale": 0,
        "schemaName": "",
        "tableName": "test_table"
      }
    ],
    "message": null,
    "resultIndex": 1,
    "sessionId": 1560946281,
    "traceId": "HZ1560946281OGMJSFEX",
    "type": "META"
  }

• ROWS 事件：

  {
    "attributes": {},
    "count": 1,
    "message": null,
    "resultIndex": 1,
    "rows": [
      [
        1,
        "2025-12-17 16:23:53"
      ]
    ],
    "sessionId": 1560946281,
    "traceId": "HZ1560946281OGMJSFEX",
    "type": "ROWS"
  }

• EOF 事件：

  {
    "attributes": {
      "TotalRowCount": 1
    },
    "lastResult": true,
    "message": null,
    "resultIndex": 1,
    "sessionId": 1560946281,
    "traceId": "HZ1560946281OGMJSFEX",
    "type": "EOF"
  }

• ERROR 事件：

  {
    "attributes": {},
    "message": "[3050: Building Plan Error] PlanDAG build failed: The API does not exist. Publish it and try again.",
    "sessionId": 1560944281,
    "traceId": "HZ1560944281VWJLYLAK",
    "type": "ERROR"
  }