Merge pull request #285 from NigelWu95/dev

吴炳亨 · web-flow · commit 60edfc90dfd1 · 2019-12-26T16:40:11.000+08:00
Dev
diff --git a/README.md b/README.md
@@ -392,9 +392,15 @@ java.net.SocketTimeoutException: timeout
 5、断点续操作时建议修改下 save-path，便于和上一次保存的结果做区分（7.72 及以下版本中断点参数请和其他参数保持一致放在命令行或配置文件中，7.72 以上
 版本无此限制，只要提供断点参数无论是否与其他参数同在命令行或配置文件中均可生效）。  
 
-**注意：如果是系统宕机、断电或者强制关机或者进程强行 kill 等情况，无法得到输出的断点文件提示，因此只能通过[<位置记录日志>](#9-程序日志)来查看最后
+**注意：  
+（1）如果是系统宕机、断电或者强制关机或者进程强行 kill 等情况，无法得到输出的断点文件提示，因此只能通过[<位置记录日志>](#9-程序日志)来查看最后
 的断点信息，在 8.2.1 版本以上设置了 log 参数可用于启用日志记录的断点，即取出运行路径下 logs 目录中的 procedure[x].log 日志，将该日志文件设置
-为 `-log=<procedure[x].log's path>` 再运行可完成断点续操作。**  
+为 `-log=<procedure[x].log's path>` 再运行可完成断点续操作。
+（2）如果原任务包含 process 过程（只有数据源读取不包含 process 操作可以不考虑该问题），执行断点操作时，由于断点日志粒度按照 unit-len 来记录的，
+当 unit-len 比较大（默认一般都是 1000 以上，甚至是 10000）时，可能存在记录的断点比实际 process 的进度要滞后很多条记录，因此对于一些不希望存在数
+据重复执行的 process，如 qupload/syncupload/mirror/fetch/asyncfetch 等，数据重复执行会影响效率和增加流量消耗，那么建议操作断点时设置一些
+check 参数，如 check=stat（参考：[过滤和检查](docs/datamigration.md#过滤和检查)），或者对于其他操作有这个需求的话可以自行对 process 的
+[结果输出文件](#关于持久化文件名)进行检查，查看执行到的数据行位置并对断点设置的 json 文件进行调整。**  
 
 ### 11 分布式任务方案
 对于不同账号或空间可以直接在不同的机器上执行任务，对于单个空间资源数量太大无法在合适条件下使用单台机器完成作业时，可分机器进行作业，如对一个空间列举完
diff --git a/pom.xml b/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>com.qiniu</groupId>
     <artifactId>qsuits</artifactId>
-    <version>8.3.8</version>
+    <version>8.3.9</version>
     <name>qsuits</name>
     <description>qiniu-suits is a efficient tools for qiniu api implemented by java8.</description>
     <url>https://github.com/NigelWu95/qiniu-suits-java</url>
diff --git a/src/main/java/com/qiniu/datasource/TextContainer.java b/src/main/java/com/qiniu/datasource/TextContainer.java
@@ -131,6 +131,7 @@ public void export(ITextReader reader, IResultOutput saver, ILineProcess<T> proc
         Map<String, String> map = urisMap.get(reader.getName());
         JsonObject json = map != null ? JsonUtils.toJsonObject(map) : (lastLine != null ? new JsonObject() : null);
         while (lastLine != null) {
+            if (stopped) break;
             if (LocalDateTime.now(DatetimeUtils.clock_Default).isAfter(pauseDateTime)) {
                 synchronized (object) {
                     object.wait();
@@ -146,7 +147,6 @@ public void export(ITextReader reader, IResultOutput saver, ILineProcess<T> proc
                     if (retry == 0) throw e;
                 }
             }
-            statistics.addAndGet(srcList.size());
             convertedList = converter.convertToVList(srcList);
             if (converter.errorSize() > 0) saver.writeError(converter.errorLines(), false);
             if (stringConverter != null) {
@@ -166,6 +166,8 @@ public void export(ITextReader reader, IResultOutput saver, ILineProcess<T> proc
                     if (e.response != null) e.response.close();
                 }
             }
+            statistics.addAndGet(srcList.size());
+            if (stopped) break;
             lastLine = reader.currentEndLine();
             json.addProperty("start", lastLine);
             recordLister(reader.getName(), json.toString());
diff --git a/src/main/java/com/qiniu/datasource/TextFileContainer.java b/src/main/java/com/qiniu/datasource/TextFileContainer.java
@@ -69,7 +69,7 @@ private List<ITextReader> splitSingleFile(File file) throws IOException {
         if (linesNumber < threads * 2) {
             return new ArrayList<ITextReader>(){{ add(new TextFileReader(file, null, unitLen)); }};
         }
-        long avgLines = linesNumber / threads;
+        long avgLines = (linesNumber + threads - 1) / threads;
         long avgSize = avgLines * lineSize;
         RandomAccessFile[] accessFiles = new RandomAccessFile[threads];
         accessFiles[0] = new RandomAccessFile(file, "r");
diff --git a/src/main/java/com/qiniu/datasource/TextFileRandomReader.java b/src/main/java/com/qiniu/datasource/TextFileRandomReader.java
@@ -69,10 +69,7 @@ public List<String> readLines() throws IOException {
             }
         } else {
             while (true) {
-                if (srcList.size() >= limit) {
-                    line = null;
-                    break;
-                }
+                if (srcList.size() >= limit) break;
                 try {
                     // 避免文件过大，行数过多，使用 lines() 的 stream 方式直接转换可能会导致内存泄漏，故使用 readLine() 的方式
                     originalLine = accessFile.readLine();
diff --git a/src/main/java/com/qiniu/entry/CommonParams.java b/src/main/java/com/qiniu/entry/CommonParams.java
@@ -1142,14 +1142,14 @@ private void checkFilterForProcess() throws IOException {
     private void setUnitLen() throws IOException {
         String unitLen = entryParam.getValue("unit-len", "-1").trim();
         if (unitLen.startsWith("-")) {
-            if (isSelfUpload) this.unitLen = 3;
-            if ("qiniu".equals(source) || "local".equals(source)) this.unitLen = 10000;
+            if (isSelfUpload) this.unitLen = 20;
+            else if ("qiniu".equals(source)) this.unitLen = 10000;
             else this.unitLen = 1000;
         } else {
             ParamsUtils.checked(unitLen, "unit-len", "\\d+");
             this.unitLen = Integer.parseInt(unitLen);
             if (isSelfUpload && this.unitLen > 100) {
-                throw new IOException("file upload shouldn't has too big unit-len, suggest setting unit-len smaller than 100.");
+                throw new IOException("file upload shouldn't have too big unit-len, suggest to set unit-len smaller than 100.");
             }
         }
     }
@@ -1243,20 +1243,17 @@ private void setSavePath() throws IOException {
         } else {
             File file = new File(savePath);
             File[] files = file.listFiles();
-            boolean isOk = false;
+            boolean isOk;
             if (files != null && files.length > 0) {
-                for (File file1 : files) {
-                    if (file1.length() > 0) {
-                        if (file1.getName().startsWith(source) || (!"".equals(process) && file1.getName().startsWith(process))) {
-                            isOk = true;
-                            break;
-                        }
-                    }
+                if ("".equals(process)) {
+                    isOk = Arrays.asList(files).parallelStream().anyMatch(f -> f.getName().startsWith(source));
+                } else {
+                    isOk = Arrays.asList(files).parallelStream().anyMatch(f -> f.getName().startsWith(source) || f.getName().startsWith(process));
                 }
                 if (isOk) {
                     if (pathConfigMap == null || pathConfigMap.size() <= 0) {
                         throw new IOException(String.format("please change the save-path \"%s\", " +
-                                "because there are last listed files, for not cover them.", savePath));
+                                "because there are remained files from last job, for not cover them.", savePath));
                     }
                 } else {
                     throw new IOException(String.format("please change save-path \"%s\" because it's not empty.", savePath));
diff --git a/src/main/java/com/qiniu/process/Base.java b/src/main/java/com/qiniu/process/Base.java
@@ -171,7 +171,7 @@ protected List<T> parseBatchResult(List<T> processList, String result) throws Ex
      * @throws IOException 处理失败可能抛出的异常
      */
     private void batchProcess(List<T> lineList, int batchSize, int retryTimes) throws IOException {
-        int times = lineList.size()/batchSize + 1;
+        int times = (lineList.size() + batchSize - 1) / batchSize;
         List<T> processList;
         int retry;
         String message = null;
diff --git a/version.properties b/version.properties
@@ -1 +1 @@
-version=8.3.8
+version=8.3.9

Original file line number	Diff line number	Diff line change
`@@ -131,6 +131,7 @@ public void export(ITextReader reader, IResultOutput saver, ILineProcess<T> proc`
`131`	`131`	`Map<String, String> map = urisMap.get(reader.getName());`
`132`	`132`	`JsonObject json = map != null ? JsonUtils.toJsonObject(map) : (lastLine != null ? new JsonObject() : null);`
`133`	`133`	`while (lastLine != null) {`
	`134`	`+ if (stopped) break;`
`134`	`135`	`if (LocalDateTime.now(DatetimeUtils.clock_Default).isAfter(pauseDateTime)) {`
`135`	`136`	`synchronized (object) {`
`136`	`137`	`object.wait();`
`@@ -146,7 +147,6 @@ public void export(ITextReader reader, IResultOutput saver, ILineProcess<T> proc`
`146`	`147`	`if (retry == 0) throw e;`
`147`	`148`	`}`
`148`	`149`	`}`
`149`		`- statistics.addAndGet(srcList.size());`
`150`	`150`	`convertedList = converter.convertToVList(srcList);`
`151`	`151`	`if (converter.errorSize() > 0) saver.writeError(converter.errorLines(), false);`
`152`	`152`	`if (stringConverter != null) {`
`@@ -166,6 +166,8 @@ public void export(ITextReader reader, IResultOutput saver, ILineProcess<T> proc`
`166`	`166`	`if (e.response != null) e.response.close();`
`167`	`167`	`}`
`168`	`168`	`}`
	`169`	`+ statistics.addAndGet(srcList.size());`
	`170`	`+ if (stopped) break;`
`169`	`171`	`lastLine = reader.currentEndLine();`
`170`	`172`	`json.addProperty("start", lastLine);`
`171`	`173`	`recordLister(reader.getName(), json.toString());`
Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,7 @@ private List<ITextReader> splitSingleFile(File file) throws IOException {`
`69`	`69`	`if (linesNumber < threads * 2) {`
`70`	`70`	`return new ArrayList<ITextReader>(){{ add(new TextFileReader(file, null, unitLen)); }};`
`71`	`71`	`}`
`72`		`- long avgLines = linesNumber / threads;`
	`72`	`+ long avgLines = (linesNumber + threads - 1) / threads;`
`73`	`73`	`long avgSize = avgLines * lineSize;`
`74`	`74`	`RandomAccessFile[] accessFiles = new RandomAccessFile[threads];`
`75`	`75`	`accessFiles[0] = new RandomAccessFile(file, "r");`