本文介绍如何使用数据库传输服务 DTS Java SDK 快速调用 API 执行创建数据传输任务和查询任务信息。 ## 前提条件 1. [下载安装 SDK 已完成](https://www.volcengine.com/docs/6390/2124639)。 2. [初始化客户端已完成](https://www.volcengine.com/docs/6390/2124640)。 ## 创建数据传输任务 ### 创建数据迁移任务 ```Java @Test public void createTransmissionTaskMigrate() { String reqStringMigrate = "{\"TaskName\":\"zqh_test_0319_sdk_01\",\"TrafficSpec\":\"Standard\",\"ProjectName\":\"default\",\"Tags\":[],\"ChargeConfig\":{\"ChargeType\":\"PostPaid\"},\"SrcConfig\":{\"EndpointType\":\"Volc_MySQL\",\"VolcMySQLSettings\":{\"RegionSettings\":{\"Region\":\"cn-beijing\"},\"DBInstanceId\":\"mysql-350fc6******\",\"Username\":\"zqh_******\",\"Password\":\"zqh_******\"}},\"DestConfig\":{\"EndpointType\":\"Volc_MySQL\",\"VolcMySQLSettings\":{\"RegionSettings\":{\"Region\":\"cn-beijing\"},\"DBInstanceId\":\"mysql-91d29f14d1e8\",\"Username\":\"ybw_te******\",\"Password\":\"ybw_te******\"}},\"SolutionSettings\":{\"SolutionType\":\"MySQL2MySQL\",\"MySQL2MySQLSettings\":{\"PolicyForPrimaryKeyConflict\":\"Override\",\"IncrTransmissionSettings\":{\"EnableIncr\":true},\"EnableForeignKeyChecks\":false,\"ObjectMappings\":[{\"ObjectType\":\"Database\",\"SrcObjName\":\"zqh_test_0319\",\"DestObjName\":\"zqh_test_0319\",\"MappingList\":[{\"ObjectType\":\"Table\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\",\"Incr\",\"Full\"],\"SetObjectTransType\":true}},{\"ObjectType\":\"View\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Trigger\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Function\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Procedure\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Event\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}}],\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\",\"Full\",\"Incr\"],\"SetObjectTransType\":true}}],\"EnableTriggerPreMigration\":false,\"EnableTrxRewrite\":false,\"EnableFullTrxPartitionByChunk\":true,\"MetaTransmissionSettings\":{\"EnableMeta\":true},\"ErrorBehaviorSettings\":{\"MaxRetrySeconds\":600},\"FullTransmissionSettings\":{\"EnableFull\":true,\"ExtraConditions\":[]},\"AccountTransmissionSettings\":{\"EnableAccount\":false}}},\"TaskType\":\"DataMigration\"}"; String reqString = reqStringMigrate; CreateTransmissionTaskRequest createTransmissionTaskRequest = jsonUtil.deserialize(reqString, CreateTransmissionTaskRequest.class); createTransmissionTaskRequest.getChargeConfig().setOneStep(true); try { CreateTransmissionTaskResponse response = dtsApi.createTransmissionTask(createTransmissionTaskRequest); System.out.println(response); this.taskId = response.getTaskId(); } catch (ApiException e) { System.out.println("createTransmissionTask got exception"); throw new RuntimeException(e); } } ``` ### 创建数据订阅任务 ```Java @Test public void createTransmissionTaskSubscribe() { String reqStringSubscription = "{\"TaskName\":\"zqh_test_0319_sdk_sub_01\",\"TrafficSpec\":\"Standard\",\"ProjectName\":\"default\",\"Tags\":[],\"ChargeConfig\":{\"ChargeType\":\"PostPaid\"},\"SrcConfig\":{\"EndpointType\":\"Volc_MySQL\",\"VolcMySQLSettings\":{\"RegionSettings\":{\"Region\":\"cn-beijing\"},\"DBInstanceId\":\"mysql-350fc626808b\",\"Username\":\"zqh_******\",\"Password\":\"zqh_******\"}},\"DestConfig\":{\"EndpointType\":\"Volc_RocketMQ\",\"VolcRocketMQSettings\":{\"RegionSettings\":{\"Region\":\"cn-beijing\"},\"InstanceId\":\"rocketmq-cnoeb389de8*****\",\"Topic\":\"zqh_test_0319\",\"Username\":\"WCN1a31TrzxCfllf8kWrBBk5\",\"Password\":\"zgOUp6jt3vhGJF2ptG******\",\"AuthType\":\"PLAIN\"}},\"SolutionSettings\":{\"SolutionType\":\"MySQL2RocketMQ\",\"MySQL2RocketMQSettings\":{\"SubscriptionSettings\":{\"Protocol\":\"CanalJSON\",\"PartitionPolicy\":\"CustomColumn\"},\"MySQLPosition\":{},\"IncrTransmissionSettings\":{\"Statements\":[\"StmtDMLInsert\",\"StmtDMLUpdate\",\"StmtDMLDelete\",\"StmtDDLAll\"],\"EnableIncr\":true},\"ObjectMappings\":[{\"ObjectType\":\"Database\",\"SrcObjName\":\"zqh_test_0319\",\"DestObjName\":\"zqh_test_0319\",\"MappingList\":[{\"ObjectType\":\"Table\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"PartitionColumn\":\"zqh_test_0319_string_field_01\"}}],\"ObjectMappingSetting\":{\"PartitionColumn\":\"zqh_test_0319_string_field_01\"}}],\"ErrorBehaviorSettings\":{\"MaxRetrySeconds\":600},\"ETLSettings\":{\"Script\":\"\"},\"MetaTransmissionSettings\":{\"EnableMeta\":true},\"FullTransmissionSettings\":{\"EnableFull\":false}}},\"TaskType\":\"DataSubscription\"}"; String reqString = reqStringSubscription; CreateTransmissionTaskRequest createTransmissionTaskRequest = jsonUtil.deserialize(reqString, CreateTransmissionTaskRequest.class); createTransmissionTaskRequest.getChargeConfig().setOneStep(true); try { CreateTransmissionTaskResponse response = dtsApi.createTransmissionTask(createTransmissionTaskRequest); System.out.println(response); this.taskId = response.getTaskId(); } catch (ApiException e) { System.out.println("createTransmissionTask got exception"); throw new RuntimeException(e); } } ``` ### 创建数据同步任务 ```Java @Test public void createTransmissionTaskSync() { String reqStringSync = "{\"TaskName\":\"zqh_test_0319_sdk_sync_01\",\"TrafficSpec\":\"Compact\",\"ProjectName\":\"default\",\"Tags\":[],\"CreateBackwardSyncTask\":false,\"ChargeConfig\":{\"ChargeType\":\"PostPaid\"},\"SrcConfig\":{\"EndpointType\":\"Volc_MySQL\",\"VolcMySQLSettings\":{\"RegionSettings\":{\"Region\":\"cn-beijing\"},\"DBInstanceId\":\"mysql-350fc626808b\",\"Username\":\"zqh_te******\",\"Password\":\"zqh_******\"}},\"DestConfig\":{\"EndpointType\":\"Volc_MySQL\",\"VolcMySQLSettings\":{\"RegionSettings\":{\"Region\":\"cn-beijing\"},\"DBInstanceId\":\"mysql-91d29f1*****\",\"Username\":\"ybw_te******\",\"Password\":\"ybw_te******\"}},\"SolutionSettings\":{\"SolutionType\":\"MySQL2MySQL\",\"MySQL2MySQLSettings\":{\"PolicyForPrimaryKeyConflict\":\"Override\",\"IncrTransmissionSettings\":{\"Statements\":[\"StmtDDLAll\",\"StmtDMLDelete\",\"StmtDMLInsert\",\"StmtDMLUpdate\"],\"EnableIncr\":true},\"EnableForeignKeyChecks\":false,\"ObjectMappings\":[{\"ObjectType\":\"Database\",\"SrcObjName\":\"zqh_test_0319\",\"DestObjName\":\"ybw_test_0601\",\"MappingList\":[{\"ObjectType\":\"Table\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\",\"Incr\",\"Full\"],\"SetObjectTransType\":true,\"PolicyForKeyConflict\":\"TableOverride\"}},{\"ObjectType\":\"View\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Function\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Procedure\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Trigger\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}},{\"ObjectType\":\"Event\",\"SrcObjName\":\"*\",\"DestObjName\":\"*\",\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\"]}}],\"ObjectMappingSetting\":{\"ObjectTransTypes\":[\"Meta\",\"Full\",\"Incr\"],\"SetObjectTransType\":true,\"PolicyForKeyConflict\":\"TableOverride\"}}],\"EnableTriggerPreMigration\":false,\"EnableTrxRewrite\":false,\"EnableFullTrxPartitionByChunk\":false,\"MetaTransmissionSettings\":{\"EnableSyncPostMeta\":false,\"EnableMeta\":true},\"ETLSettings\":{\"Script\":\"\"},\"ErrorBehaviorSettings\":{\"MaxRetrySeconds\":600},\"FullTransmissionSettings\":{\"EnableFull\":true,\"ExtraConditions\":[]},\"AccountTransmissionSettings\":{\"EnableAccount\":false}}},\"TaskType\":\"DataSynchronization\"}"; String reqString = reqStringSync; CreateTransmissionTaskRequest createTransmissionTaskRequest = jsonUtil.deserialize(reqString, CreateTransmissionTaskRequest.class); createTransmissionTaskRequest.getChargeConfig().setOneStep(true); try { CreateTransmissionTaskResponse response = dtsApi.createTransmissionTask(createTransmissionTaskRequest); System.out.println(response); this.taskId = response.getTaskId(); } catch (ApiException e) { System.out.println("createTransmissionTask got exception"); throw new RuntimeException(e); } } ``` ## 查询数据传输任务信息 ```Java @Test public void describeTransmissionTaskInfo() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); DescribeTransmissionTaskInfoRequest describeTransmissionTaskInfoRequest = jsonUtil.deserialize(reqString, DescribeTransmissionTaskInfoRequest.class); try { DescribeTransmissionTaskInfoResponse response = dtsApi.describeTransmissionTaskInfo(describeTransmissionTaskInfoRequest); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 暂停数据传输任务 ```Java @Test public void suspendTransmissionTask() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); SuspendTransmissionTaskRequest request = jsonUtil.deserialize(reqString, SuspendTransmissionTaskRequest.class); try { SuspendTransmissionTaskResponse response = dtsApi.suspendTransmissionTask(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 启动数据传输任务 ```Java @Test public void startTransmissionTask() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); StartTransmissionTaskRequest request = jsonUtil.deserialize(reqString, StartTransmissionTaskRequest.class); try { StartTransmissionTaskResponse response = dtsApi.startTransmissionTask(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 重试数据传输任务 ```Java @Test public void RetryTransmissionTask() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); RetryTransmissionTaskRequest request = jsonUtil.deserialize(reqString, RetryTransmissionTaskRequest.class); try { RetryTransmissionTaskResponse response = dtsApi.retryTransmissionTask(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 恢复数据传输任务 ```Java @Test public void resumeTransmissionTask() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); ResumeTransmissionTaskRequest request = jsonUtil.deserialize(reqString, ResumeTransmissionTaskRequest.class); try { ResumeTransmissionTaskResponse response = dtsApi.resumeTransmissionTask(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 终止数据传输任务 ```Java @Test public void stopTransmissionTask() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); StopTransmissionTaskRequest request = jsonUtil.deserialize(reqString, StopTransmissionTaskRequest.class); try { StopTransmissionTaskResponse response = dtsApi.stopTransmissionTask(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 删除数据传输任务 ```Java @Test public void deleteTransmissionTask() { String reqString = String.format("{\"TaskId\":\"%s\"}", this.taskId); DeleteTransmissionTaskRequest request = jsonUtil.deserialize(reqString, DeleteTransmissionTaskRequest.class); try { DeleteTransmissionTaskResponse response = dtsApi.deleteTransmissionTask(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 创建预检查任务 ```Java ApiClient apiClient = new ApiClient(); apiClient.setEndpoint("dts.cn-beijing.volcengineapi.com"); apiClient.setRegion("cn-beijing"); apiClient.setCredentials(Credentials.getCredentials(AK, SK)); dtsApi = new Dts20180101Api(apiClient); @Test public void preCheckAsync() { String reqString = String.format("{\"TaskID\":\"%s\"}", this.taskId); PreCheckAsyncRequest request = jsonUtil.deserialize(reqString, PreCheckAsyncRequest.class); try { PreCheckAsyncResponse response = dtsApi.preCheckAsync(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ``` ## 查询预检查结果 ```Java @Test public void getAsyncPreCheckResult() { String reqString = String.format("{\"ID\":\"%s\"}", this.preCheckID); GetAsyncPreCheckResultRequest request = jsonUtil.deserialize(reqString, GetAsyncPreCheckResultRequest.class); try { GetAsyncPreCheckResultResponse response = dtsApi.getAsyncPreCheckResult(request); System.out.println(response); } catch (ApiException e) { throw new RuntimeException(e); } } ```