Skip to content

Commit dfdec5d

Browse files
kitmanzhengkitmanzheng
committed
list resume
增加功能：接着最后一次运行的结果，继续往下遍历源的文件列表
1 parent eba04fa commit dfdec5d

File tree

10 files changed

+193
-45
lines changed

10 files changed

+193
-45
lines changed

‎conf/config.ini‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,9 @@ executeTimeWindow=00:00,24:00
5252
# 迁移成功的结果，按日期归档此目录，为空即不输出。格式每一行为：绝对路径\t文件大小\t最后修改时间，该目录需要存在。
5353
outputFinishedFileFolder=./result
5454

55+
# 是否接着最后一次运行的结果，继续往下遍历源的文件列表
56+
resume=false
57+
5558
# 从本地迁移到COS配置分节
5659
[migrateLocal]
5760
# 本地路径, 表示将该路径下的数据都迁移到COS, 对于linux绝对路径, 如/a/b/c, 对于windows绝对路径，注意分隔符为两个反斜杠，如E:\\a\\b\\c

‎dep/cos_migrate_tool-1.0-jar-with-dependencies.jar‎

2.23 KB
Binary file not shown.

‎src/main/java/com/qcloud/cos_migrate_tool/config/CommonConfig.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,19 @@ public class CommonConfig {
3636
private String batchTaskPath = "";
3737
private boolean realTimeCompare = false;
3838
private String outputFinishedFilePath = "";
39+
private boolean isResume = false;
40+
41+
public void setResume(String isResume) {
42+
if (isResume.compareToIgnoreCase("true") == 0) {
43+
this.isResume = true;
44+
} else if (isResume.compareToIgnoreCase("false") != 0) {
45+
throw new IllegalArgumentException("resume invalid.should be true or false");
46+
}
47+
}
48+
49+
public boolean isResume() {
50+
return this.isResume;
51+
}
3952

4053
public void setRealTimeCompare(String realTimeCompare) throws IllegalArgumentException {
4154
if (realTimeCompare.equalsIgnoreCase("on")) {

‎src/main/java/com/qcloud/cos_migrate_tool/config/ConfigParser.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ public class ConfigParser {
5252
private static final String COMMON_BATCH_TASK_PATH = "batchTaskPath";
5353
private static final String COMMON_REAL_TIME_COMPARE = "realTimeCompare";
5454
private static final String COMMON_OUTPUT_FINISHED_FILE = "outputFinishedFileFolder";
55+
private static final String COMMON_RESUME = "resume";
5556

5657
private static final String LOCAL_SECTION_NAME = "migrateLocal";
5758
private static final String LOCAL_LOCALPATH = "localPath";
@@ -546,7 +547,11 @@ private boolean initCommonConfig(Preferences prefs, CommonConfig commonConfig) {
546547
commonConfig.setOutputFinishedFilePath(finishedFileFolder);
547548
}
548549

549-
550+
String resume = getConfigValue(prefs, COMMON_SECTION_NAME, COMMON_RESUME);
551+
if (resume!=null && !resume.isEmpty()) {
552+
commonConfig.setResume(resume);
553+
}
554+
550555
} catch (Exception e) {
551556
System.err.println(e.getMessage());
552557
log.error(e.getMessage());

‎src/main/java/com/qcloud/cos_migrate_tool/record/RecordDb.java‎

Lines changed: 87 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,14 @@
11
package com.qcloud.cos_migrate_tool.record;
22

33
import java.io.BufferedOutputStream;
4+
import java.io.BufferedReader;
5+
import java.io.FileInputStream;
46
import java.io.FileNotFoundException;
57
import java.io.FileOutputStream;
68
import java.io.IOException;
9+
import java.io.InputStreamReader;
710
import java.io.UnsupportedEncodingException;
11+
import java.util.LinkedList;
812

913
import org.rocksdb.FlushOptions;
1014
import org.rocksdb.Options;
@@ -38,12 +42,14 @@ public enum QUERY_RESULT {
3842
private RocksDB db;
3943
private Options options;
4044
private final String requestIdPrefix = "x-cos-requestId-";
45+
private String dbFolder;
46+
4147

4248
public RecordDb() {}
4349

4450
public boolean init(String historyDbFolder, String comment) {
45-
4651
try {
52+
dbFolder = historyDbFolder;
4753
options = new Options();
4854
options.setCreateIfMissing(true);
4955
options.setWriteBufferSize(16 * SizeUnit.MB).setMaxWriteBufferNumber(4)
@@ -70,6 +76,86 @@ public boolean init(String historyDbFolder, String comment) {
7076
}
7177
return true;
7278
}
79+
80+
public boolean saveListProgress(String prefix, String marker) {
81+
String value = prefix + "|" + marker;
82+
return saveKV("listProgress", value);
83+
}
84+
85+
86+
public String[] getListProgress() {
87+
String value = queryKV("listProgress");
88+
if (value == null) {
89+
return null;
90+
}
91+
92+
int i = value.lastIndexOf("|");
93+
if (i < 0) {
94+
return null;
95+
}
96+
97+
String prefix = value.substring(0, i);
98+
String marker = "";
99+
if (i != value.length()-1) {
100+
marker = value.substring(i+1, value.length()-1);
101+
}
102+
103+
String[] arr1 = new String[] {prefix, marker};
104+
return arr1;
105+
106+
}
107+
108+
public boolean saveDirProgress(String curDir, String lastItr, LinkedList<String> dirList) {
109+
110+
String progressFile = this.dbFolder + "/PROGRESS";
111+
112+
try {
113+
BufferedOutputStream bos =
114+
new BufferedOutputStream(new FileOutputStream(progressFile, false));
115+
bos.write(lastItr.getBytes(ENCODING_TYPE));
116+
bos.write("\n".getBytes(ENCODING_TYPE));
117+
bos.write(curDir.getBytes(ENCODING_TYPE));
118+
bos.write("\n".getBytes(ENCODING_TYPE));
119+
for (String x: dirList) {
120+
bos.write(x.getBytes(ENCODING_TYPE));
121+
bos.write("\n".getBytes(ENCODING_TYPE));
122+
}
123+
124+
bos.close();
125+
} catch (FileNotFoundException e) {
126+
log.error(e.toString());
127+
return false;
128+
} catch (IOException e) {
129+
log.error(e.toString());
130+
return false;
131+
}
132+
133+
return true;
134+
}
135+
136+
public LinkedList<String> getDirProgress() {
137+
String progressFile = this.dbFolder + "/PROGRESS";
138+
LinkedList<String> result = new LinkedList<String>();
139+
try {
140+
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(progressFile)));
141+
String line;
142+
while ((line = bufferedReader.readLine()) != null) {
143+
result.addLast(line);
144+
log.info("[{}]", line);
145+
}
146+
} catch (FileNotFoundException e) {
147+
// TODO Auto-generated catch block
148+
//log.error(e.toString());
149+
//e.printStackTrace();
150+
return null;
151+
} catch (IOException e) {
152+
// TODO Auto-generated catch block
153+
log.error(e.toString());
154+
e.printStackTrace();
155+
return null;
156+
}
157+
return result;
158+
}
73159

74160
// 保存记录
75161
public boolean saveRecord(RecordElement recordElement) {

‎src/main/java/com/qcloud/cos_migrate_tool/task/MigrateAliTaskExecutor.java‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -84,13 +84,18 @@ public String buildTaskDbFolderPath() {
8484
}
8585

8686
public void buildTask() {
87-
final int maxKeys = 200;
87+
final int maxKeys = 1000;
8888
final String keyPrefix = this.srcPrefix;
8989
String nextMarker = "";
9090
ObjectListing objectListing;
9191

9292
int retry_num = 0;
9393

94+
String[] progress = this.recordDb.getListProgress();
95+
if (config.isResume() && progress != null) {
96+
nextMarker = progress[1];
97+
}
98+
9499
do {
95100
try {
96101
do {
@@ -103,13 +108,17 @@ public void buildTask() {
103108
// AddTask
104109
MigrateAliTask task = new MigrateAliTask(config, ossClient,
105110
com.qcloud.cos.utils.UrlEncoderUtils.urlDecode(s.getKey()),
106-
s.getSize(), s.getETag(), s.getLastModified(), smallFileTransferManager,
107-
bigFileTransferManager, recordDb, semaphore);
111+
s.getSize(), s.getETag(), s.getLastModified(),
112+
smallFileTransferManager, bigFileTransferManager, recordDb,
113+
semaphore);
108114

109-
AddTask(task);
115+
AddTask(task);
110116
}
111117
nextMarker = com.qcloud.cos.utils.UrlEncoderUtils
112118
.urlDecode(objectListing.getNextMarker());
119+
if (nextMarker != null) {
120+
this.recordDb.saveListProgress(keyPrefix, nextMarker);
121+
}
113122
} while (objectListing.isTruncated());
114123

115124
TaskStatics.instance.setListFinished(true);

‎src/main/java/com/qcloud/cos_migrate_tool/task/MigrateAwsTaskExecutor.java‎

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -90,29 +90,40 @@ public String buildTaskDbFolderPath() {
9090
}
9191

9292
public void buildTask() {
93-
93+
String nextMarker = "";
94+
String[] progress = this.recordDb.getListProgress();
95+
if (config.isResume() && progress != null) {
96+
nextMarker = progress[1];
97+
}
9498
try {
9599
ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
96100
listObjectsRequest.setBucketName(srcBucket);
97101
listObjectsRequest.setPrefix(srcPrefix);
102+
listObjectsRequest.setMarker(nextMarker);
98103
ObjectListing objectListing = null;
99104
do {
105+
100106
objectListing = s3Client.listObjects(listObjectsRequest);
101107
for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries()) {
102108
// AddTask
103109
MigrateAwsTask task = new MigrateAwsTask(config, s3Client,
104110
objectSummary.getKey(), objectSummary.getSize(),
105111
objectSummary.getETag(), smallFileTransferManager,
106112
bigFileTransferManager, recordDb, semaphore);
107-
log.info("list key: {}, size: {}, etag: {}", objectSummary.getKey(), objectSummary.getSize(), objectSummary.getETag());
113+
log.info("list key: {}, size: {}, etag: {}", objectSummary.getKey(),
114+
objectSummary.getSize(), objectSummary.getETag());
108115

109-
AddTask(task);
116+
AddTask(task);
117+
}
118+
nextMarker = objectListing.getNextMarker();
119+
listObjectsRequest.setMarker(nextMarker);
120+
if (nextMarker != null) {
121+
this.recordDb.saveListProgress(srcPrefix, nextMarker);
110122
}
111-
listObjectsRequest.setMarker(objectListing.getNextMarker());
112123
} while (objectListing.isTruncated());
113-
124+
114125
TaskStatics.instance.setListFinished(true);
115-
126+
116127
} catch (AmazonServiceException ase) {
117128
log.error("list fail AmazonServiceException errorcode: {}, msg: {}", ase.getErrorCode(),
118129
ase.getMessage());
@@ -122,7 +133,7 @@ public void buildTask() {
122133
TaskStatics.instance.setListFinished(false);
123134
} catch (Exception e) {
124135
log.error(e.getMessage());
125-
TaskStatics.instance.setListFinished(false);
136+
TaskStatics.instance.setListFinished(false);
126137
}
127138

128139
}

‎src/main/java/com/qcloud/cos_migrate_tool/task/MigrateCopyBucketTaskExecutor.java‎

Lines changed: 30 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -100,13 +100,13 @@ public void buildTask() {
100100
e1.printStackTrace();
101101
return;
102102
}
103-
103+
104104
BufferedReader bufferedReader = new BufferedReader(read);
105105
String srcKey = null;
106106

107107
try {
108-
while ((srcKey = bufferedReader.readLine()) != null){
109-
108+
while ((srcKey = bufferedReader.readLine()) != null) {
109+
110110
srcKey = UrlEncoderUtils.urlDecode(srcKey);
111111

112112
String copyDestKey = null;
@@ -116,18 +116,18 @@ public void buildTask() {
116116
} else {
117117
copyDestKey = config.getCosPath() + srcKey;
118118
}
119-
120-
MigrateCopyBucketTask task = new MigrateCopyBucketTask(semaphore,
121-
(CopyBucketConfig) config, smallFileTransferManager,
122-
bigFileTransferManager, recordDb, srcCosClient, srcKey, 0,
123-
"", copyDestKey);
124-
125-
AddTask(task);
126-
119+
120+
MigrateCopyBucketTask task =
121+
new MigrateCopyBucketTask(semaphore, (CopyBucketConfig) config,
122+
smallFileTransferManager, bigFileTransferManager, recordDb,
123+
srcCosClient, srcKey, 0, "", copyDestKey);
124+
125+
AddTask(task);
126+
127127
}
128-
128+
129129
TaskStatics.instance.setListFinished(true);
130-
130+
131131
} catch (IOException e) {
132132
log.error(e.toString());
133133
TaskStatics.instance.setListFinished(false);
@@ -137,7 +137,7 @@ public void buildTask() {
137137
e.printStackTrace();
138138
TaskStatics.instance.setListFinished(false);
139139
}
140-
140+
141141
try {
142142
bufferedReader.close();
143143
} catch (IOException e) {
@@ -148,43 +148,51 @@ public void buildTask() {
148148
} catch (IOException e) {
149149
e.printStackTrace();
150150
}
151-
151+
152152

153153
} else {
154+
String nextMarker = "";
155+
String[] progress = this.recordDb.getListProgress();
156+
if (config.isResume() && progress != null) {
157+
nextMarker = progress[1];
158+
}
154159

155160
ListObjectsRequest listObjectsRequest =
156-
new ListObjectsRequest(srcBucketName, srcCosPath, null, null, 1000);
161+
new ListObjectsRequest(srcBucketName, srcCosPath, nextMarker, null, 1000);
157162

158163
ObjectListing objectListing;
159164
int retry_num = 0;
160165

161166
do {
162167
try {
163168
while (true) {
169+
listObjectsRequest.setMarker(nextMarker);
164170
objectListing = srcCosClient.listObjects(listObjectsRequest);
165171
List<COSObjectSummary> cosObjectSummaries =
166172
objectListing.getObjectSummaries();
167-
173+
168174
for (COSObjectSummary cosObjectSummary : cosObjectSummaries) {
169175
String srcKey = cosObjectSummary.getKey();
170176
String srcEtag = cosObjectSummary.getETag();
171177
long srcSize = cosObjectSummary.getSize();
172178
String keyName = srcKey.substring(lastDelimiter);
173179
String copyDestKey = config.getCosPath() + keyName;
174-
180+
175181
MigrateCopyBucketTask task = new MigrateCopyBucketTask(semaphore,
176182
(CopyBucketConfig) config, smallFileTransferManager,
177183
bigFileTransferManager, recordDb, srcCosClient, srcKey, srcSize,
178184
srcEtag, copyDestKey);
179-
185+
180186
AddTask(task);
181187
}
182-
188+
nextMarker = objectListing.getNextMarker();
189+
if (nextMarker != null) {
190+
this.recordDb.saveListProgress(srcCosPath, nextMarker);
191+
}
183192
if (!objectListing.isTruncated()) {
184193
break;
185194
}
186-
187-
listObjectsRequest.setMarker(objectListing.getNextMarker());
195+
188196
}
189197

190198
TaskStatics.instance.setListFinished(true);

‎src/main/java/com/qcloud/cos_migrate_tool/task/MigrateUpyunTask.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -82,11 +82,12 @@ private String buildCOSPath() {
8282

8383
@Override
8484
public void doTask() {
85+
8586

8687
String cosPath = buildCOSPath();
8788

8889
this.etag = this.lastModify.toString();
89-
90+
9091
MigrateCompetitorRecordElement upyunRecordElement = new MigrateCompetitorRecordElement(
9192
MigrateType.MIGRATE_FROM_UPYUN, config.getBucketName(), cosPath, etag, fileSize);
9293

0 commit comments

Comments
 (0)