BUG:在往目錄中copy大文件時，沒有復制完,flume就開始讀-->導致報錯

在代碼中體現為:
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.retireCurrentFile()方法內

解決方案:
等文件完全拷貝完成，再開始讀這個文件

1.5版本:

private Optional<FileInfo> getNextFile() {7 /* Filter to exclude finished or hidden files */8 FileFilter filter = new FileFilter() {9 public boolean accept(File candidate) { 10 String fileName = candidate.getName(); 11 if ((candidate.isDirectory()) || 12 (fileName.endsWith(completedSuffix)) || 13 (fileName.startsWith(".")) || 14 ignorePattern.matcher(fileName).matches()) { 15 return false; 16 } 17 return true; 18 } 19 }; 20 List<File> candidateFiles = Arrays.asList(spoolDirectory.listFiles(filter)); //獲取spoolDirectory下滿足條件的文件 21 if (candidateFiles.isEmpty()) { 22 return Optional.absent(); 23 } else { 24 Collections.sort(candidateFiles, new Comparator<File>() { //按最后修改時間排序文件 25 public int compare(File a, File b) { 26 int timeComparison = new Long(a.lastModified()).compareTo( 27 new Long(b.lastModified())); 28 if (timeComparison != 0) { 29 return timeComparison; 30 } 31 else { 32 return a.getName().compareTo(b.getName()); 33 } 34 } 35 }); 36 File nextFile = candidateFiles.get(0); //因為每次獲取到的文件處理完都會被標記為已完成，所以直接取拍完序的第一個 37 //修復傳輸大文件報錯文件被修改的BUG 38 this.checkFileCpIsOver(nextFile);//此處被阻塞，直到文件拷貝文件或者超過20秒 39 40 try { 41 // roll the meta file, if needed 42 String nextPath = nextFile.getPath()

1.7版本 :

private Optional<FileInfo> getNextFile() {List<File> candidateFiles = Collections.emptyList();if (consumeOrder != ConsumeOrder.RANDOM ||candidateFileIter == null ||!candidateFileIter.hasNext()) {candidateFiles = getCandidateFiles(spoolDirectory.toPath());listFilesCount++;candidateFileIter = candidateFiles.iterator();}if (!candidateFileIter.hasNext()) { // No matching file in spooling directory.return Optional.absent();}File selectedFile = candidateFileIter.next();if (consumeOrder == ConsumeOrder.RANDOM) { // Selected file is random.return openFile(selectedFile);} else if (consumeOrder == ConsumeOrder.YOUNGEST) {for (File candidateFile : candidateFiles) {long compare = selectedFile.lastModified() -candidateFile.lastModified();if (compare == 0) { // ts is same pick smallest lexicographically.selectedFile = smallerLexicographical(selectedFile, candidateFile);} else if (compare < 0) { // candidate is younger (cand-ts > selec-ts)selectedFile = candidateFile;}}} else { // default order is OLDESTfor (File candidateFile : candidateFiles) {long compare = selectedFile.lastModified() -candidateFile.lastModified();if (compare == 0) { // ts is same pick smallest lexicographically.selectedFile = smallerLexicographical(selectedFile, candidateFile);} else if (compare > 0) { // candidate is older (cand-ts < selec-ts).selectedFile = candidateFile;}}}firstTimeRead = true;//修復傳輸大文件報錯文件被修改的BUG this.checkFileCpIsOver(selectedFile);//此處被阻塞，直到文件拷貝文件或者超過20秒return openFile(selectedFile);}

?

解決代碼:

/**** @Title: checkFileCpIsOver* @Description: TODO(用來檢查文件拷貝是否完成)* @param @param currentFile 設定文件* @return void 返回類型* @throws*/ private void checkFileCpIsOver(File file) {long modified = file.lastModified();//目前文件的修改時間long length = file.length();//目前文件的大小try {Thread.sleep(1000);//等待1秒鐘} catch (InterruptedException e) {// TODO Auto-generated catch block e.printStackTrace();}File currentFile = new File(file.getAbsolutePath());int count = 0;//記錄循環次數，超過20次，也就是10秒后拋出異常while(currentFile.lastModified() != modified || currentFile.length() != length) {if(count > 20) {String message = "File Copy time too long. please check copy whether exception!" + "\n"+ "File at :" + file.getAbsolutePath() + "\n"+ "File current length is：" + currentFile.lastModified();new IllegalStateException(message);}count++;modified = currentFile.lastModified();length = currentFile.length();try {Thread.sleep(500);//等待500毫秒} catch (InterruptedException e) {// TODO Auto-generated catch block e.printStackTrace();}currentFile = new File(file.getAbsolutePath());}//一直到文件傳輸完成就可以退出 }

?

總結

以上是生活随笔為你收集整理的flume spooldir bug修复的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。