Merge pull request #307 from NigelWu95/dev

吴炳亨 · web-flow · commit 0846876c4e98 · 2020-11-26T04:36:08.000+08:00
fix procedure log process.
diff --git a/Makefile b/Makefile
@@ -1,4 +1,4 @@
-version=8.4.7
+version=8.4.8
 
 package_no_test:
 	mvn clean package -Dmaven.test.skip=true
diff --git a/pom.xml b/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>com.qiniu</groupId>
     <artifactId>qsuits</artifactId>
-    <version>8.4.7</version>
+    <version>8.4.8</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/CloudStorageContainer.java b/src/main/java/com/qiniu/datasource/CloudStorageContainer.java
@@ -268,8 +268,7 @@ void listing(IStorageLister<E> lister) {
                 processorMap.put(orderStr, lineProcessor);
             }
             export(lister, saver, lineProcessor);
-            procedureLogger.info("{}-|-", lister.getPrefix());
-            progressMap.remove(lister.getPrefix()); // 只有 export 成功情况下才移除 record
+            removeRecordedPrefix(lister.getPrefix()); // 只有 export 成功情况下才移除 record
         } catch (QiniuException e) {
             try { FileUtils.createIfNotExists(errorLogFile); } catch (IOException ignored) {}
             errorLogger.error("{}: {}, {}", lister.getPrefix(), progressMap.get(lister.getPrefix()), e.error(), e);
@@ -383,7 +382,7 @@ private List<IStorageLister<E>> filteredListerByPrefixes(Stream<String> prefixes
             if (generated == null) return false;
             else if (generated.currents().size() > 0 || generated.hasNext()) return true;
             else {
-                progressMap.remove(generated.getPrefix());
+                removeRecordedPrefix(generated.getPrefix());
                 generated.close();
                 return false;
             }
@@ -412,7 +411,7 @@ private void processNodeLister(IStorageLister<E> lister) {
         if (lister.currents().size() > 0 || lister.hasNext()) {
             executorPool.execute(() -> listing(lister));
         } else {
-            progressMap.remove(lister.getPrefix());
+            removeRecordedPrefix(lister.getPrefix());
             lister.close();
         }
     }
@@ -483,12 +482,13 @@ private List<String> checkListerInPool(List<IStorageLister<E>> listerList, int c
                         // lister 的 prefix 为 final 对象，不能因为 truncate 的操作之后被修改
                         prefix = lister.getPrefix();
                         nextMarker = lister.truncate();
-                        // 防止 truncate 过程中原来的线程中丢失了 prefixAndEndedMap 的操作，这里再判断一次
                         endMap = prefixAndEndedMap.get(prefix);
                         prefixMap = new HashMap<>();
                         if (endMap == null) {
+                            // 做标记，endMap 为空说明该 prefix 本就不是尾结点，重新组合列举时不能因为排序最后而加入到 prefixEndedMap 中去
                             prefixMap.put("remove", "remove");
                         } else {
+                            // 防止 truncate 过程中原来的线程中丢失了 prefixAndEndedMap 的操作，这里再判断一次 endKey 进行更新
                             start = lister.currentEndKey();
                             if (start != null) endMap.put("start", start);
                         }
@@ -645,7 +645,7 @@ public void export() throws Exception {
                 if (startLister.currents().size() > 0 || startLister.hasNext()) {
                     listing(startLister);
                 } else {
-                    progressMap.remove(startLister.getPrefix());
+                    removeRecordedPrefix(startLister.getPrefix());
                     startLister.close();
                 }
             } else {
diff --git a/src/main/java/com/qiniu/datasource/DatasourceActor.java b/src/main/java/com/qiniu/datasource/DatasourceActor.java
@@ -94,11 +94,20 @@ void recordLister(String key, String record) {
         procedureLogger.info("{}-|-{}", key, record);
     }
 
+    void removeRecordedPrefix(String prefix) {
+        procedureLogger.info("{}-|-", prefix);
+        progressMap.remove(prefix);
+    }
+
     void refreshRecordAndStatistics() {
         rootLogger.info("finished count: {}.", statistics.get());
         if (procedureLogFile.length() > 536870912) { // 超过 512M 就处理一次
             try {
                 breakpointSaver.clear(breakpointFileName);
+//                StringBuilder record = new StringBuilder("{\"");
+//                progressMap.forEach((key, m) -> record.append(key).append("\":").append(m).append(",\""));
+//                record.delete(record.length() - 2, record.length()).append("}");
+//                breakpointSaver.writeToKey(breakpointFileName, record.toString().replace("\\", "\\\\"), true);
                 breakpointSaver.writeToKey(breakpointFileName, JsonUtils.toJsonWithoutUrlEscape(progressMap), true);
                 procedureLogFile.delete();
             } catch (IOException e) {
@@ -125,6 +134,10 @@ void endAction() throws IOException {
         }
         String record = "{}";
         if (progressMap.size() > 0) {
+//            StringBuilder re = new StringBuilder("{\"");
+//            progressMap.forEach((key, m) -> re.append(key).append("\":").append(m).append(",\""));
+//            re.delete(re.length() - 2, re.length()).append("}");
+//            record = re.toString().replace("\\", "\\\\");
             record = JsonUtils.toJsonWithoutUrlEscape(progressMap);
             breakpointSaver.writeToKey(breakpointFileName, record, true);
             breakpointSaver.closeWriters();
diff --git a/src/main/java/com/qiniu/entry/CommonParams.java b/src/main/java/com/qiniu/entry/CommonParams.java
@@ -715,51 +715,51 @@ private void setPrivateType() throws IOException {
     }
 
     private void fromProcedureLog(String logFile, boolean withMarker, boolean withEnd) throws IOException {
-        String lastLine = FileUtils.lastLineOfFile(logFile);
-        if (lastLine != null && !"".equals(lastLine)) {
-            try {
+        try {
+            String lastLine = FileUtils.lastLineOfFile(logFile);
+            if (lastLine != null && !"".equals(lastLine)) {
                 parseConfigMapFromJson(JsonUtils.toJsonObject(lastLine), withMarker, withEnd);
-            } catch (Exception e) {
-                File file = new File(logFile);
-                FileReader fileReader = new FileReader(file);
-                BufferedReader bufferedReader = new BufferedReader(fileReader);
-                int index;
-                String line;
-                String value;
-                Map<String, String> map = new HashMap<>();
-                while ((line = bufferedReader.readLine()) != null) {
-                    index = line.indexOf("-|-");
-                    if (index < 0) {
-                        try {
-                            parseConfigMapFromJson(JsonUtils.toJsonObject(line), withMarker, withEnd);
-                            return;
-                        } catch (Exception exception) {
-                            exception.printStackTrace();
-                        }
-                    } else {
-                        map.put(line.substring(0, index), line.substring(index));
+            }
+        } catch (Exception e) {
+            File file = new File(logFile);
+            FileReader fileReader = new FileReader(file);
+            BufferedReader bufferedReader = new BufferedReader(fileReader);
+            int index;
+            String line;
+            String value;
+            Map<String, String> map = new HashMap<>();
+            while ((line = bufferedReader.readLine()) != null) {
+                index = line.indexOf("-|-");
+                if (index < 0) {
+                    try {
+                        parseConfigMapFromJson(JsonUtils.toJsonObject(line), withMarker, withEnd);
+                        return;
+                    } catch (Exception exception) {
+                        exception.printStackTrace();
                     }
+                } else {
+                    map.put(line.substring(0, index), line.substring(index + 3));
                 }
-                Map<String, String> configMap;
-                for (String key : map.keySet()) {
-                    value = map.get(key);
-                    if (!"".equals(value)) {
-                        try {
-                            configMap = JsonUtils.fromJson(value, map.getClass());
-                        } catch (Exception e1) {
-                            e1.printStackTrace();
-                            continue;
-                        }
-                        pathConfigMap.put(key, configMap);
+            }
+            Map<String, String> configMap;
+            for (String key : map.keySet()) {
+                value = map.get(key);
+                if (!"".equals(value)) {
+                    try {
+                        configMap = JsonUtils.fromJson(value, map.getClass());
+                    } catch (Exception e1) {
+                        e1.printStackTrace();
+                        continue;
                     }
+                    pathConfigMap.put(key, configMap);
                 }
-                try {
-                    bufferedReader.close();
-                    fileReader.close();
-                } catch (IOException ioe) {
-                    bufferedReader = null;
-                    fileReader = null;
-                }
+            }
+            try {
+                bufferedReader.close();
+                fileReader.close();
+            } catch (IOException ioe) {
+                bufferedReader = null;
+                fileReader = null;
             }
         }
     }
@@ -1489,6 +1489,10 @@ public void setBucket(String bucket) {
         this.bucket = bucket;
     }
 
+    public void setLogFilepath(String logFilepath) {
+        this.logFilepath = logFilepath;
+    }
+
     public void setPathConfigMap(Map<String, Map<String, String>> pathConfigMap) {
         this.pathConfigMap = pathConfigMap;
     }
@@ -1689,6 +1693,10 @@ public String getBucket() {
         return bucket;
     }
 
+    public String getLogFilepath() {
+        return logFilepath;
+    }
+
     public Map<String, Map<String, String>> getPathConfigMap() {
         return pathConfigMap;
     }
diff --git a/src/main/java/com/qiniu/util/JsonUtils.java b/src/main/java/com/qiniu/util/JsonUtils.java
@@ -53,7 +53,8 @@ public static String toJsonWithoutUrlEscape(Object srcObject) {
                 .replace("\"{}\"", "{}")
                 .replace("\"{\\\"", "{\"")
                 .replace("\\\"}\"", "\"}")
-                .replace("\\\":\\\"", "\":\"");
+                .replace("\\\":\\\"", "\":\"")
+                .replace("\\\",\\\"", "\",\"");
 //                .replace("\\\\", "\\");
     }
 
diff --git a/src/test/java/com/qiniu/entry/CommonParamsTest.java b/src/test/java/com/qiniu/entry/CommonParamsTest.java
@@ -4,6 +4,7 @@
 import com.qiniu.interfaces.IEntryParam;
 import org.junit.Test;
 
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.util.Map;
 
@@ -26,4 +27,11 @@ public void test() throws Exception {
         Map<String,String> map = commonParams.getIndexMap();
         System.out.println(map);
     }
+
+    @Test
+    public void testFromProcedureLog() throws Exception {
+        CommonParams commonParams = new CommonParams();
+        commonParams.setLogFilepath("/Users/wubingheng/Downloads/procedure0.log.txt");
+//        commonParams.setPathConfigMap("", null, true, true);
+    }
 }
diff --git a/version.properties b/version.properties
@@ -1 +1 @@
-version=8.4.7
+version=8.4.8

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-version=8.4.7`
	`1`	`+version=8.4.8`
`2`	`2`
`3`	`3`	`package_no_test:`
`4`	`4`	`mvn clean package -Dmaven.test.skip=true`
Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,8 @@ public static String toJsonWithoutUrlEscape(Object srcObject) {`
`53`	`53`	`.replace("\"{}\"", "{}")`
`54`	`54`	`.replace("\"{\\\"", "{\"")`
`55`	`55`	`.replace("\\\"}\"", "\"}")`
`56`		`- .replace("\\\":\\\"", "\":\"");`
	`56`	`+ .replace("\\\":\\\"", "\":\"")`
	`57`	`+ .replace("\\\",\\\"", "\",\"");`
`57`	`58`	`// .replace("\\\\", "\\");`
`58`	`59`	`}`
`59`	`60`