在Flume進行實時資料的採集過程中，可能會出現宕機的情況，再重新工作時不能斷點續傳，導致資料丟失或資料讀取重複而浪費資源。自定義Source可以解決這個問題。

1）寫一個類繼承AbstractSource抽象類並實現Configurable介面和EventDrivenSource介面。

2）宣告自定義source中的引數：filePath（檔案路徑），offsetPath（存放偏移量檔案的路徑），interval（檔案中沒有新內容寫入時sleep的引數），charset（字符集編碼）。

3）重寫Configurable的抽象方法configure，獲取source中的引數值。

4）寫一個synchronized修飾的start方法，建立執行緒池、建立實現了Runnerble的介面，將executor放入執行緒池中，呼叫其父類的start()方法。

5）寫一個synchronized修飾的stop方法，呼叫其父類的stop()方法，在停止flume時執行一次。

6）寫一個靜態內部類實現Runnable介面。定義偏移量的初始值0LWie內部類的成員變數。類中是實現Runnable介面方法的構造方法，裡面寫斷點續傳邏輯：判斷是否有偏移量檔案（如果不存在就建立一個）；如果存在的話就讀取檔案中是否記錄了偏移量，將偏移量轉為Long型別；如果記錄了偏移量，就接著偏移量位置讀，沒有的話就從偏移量的初始值開始讀。

7）重寫實現Runnable介面類的run方法：定期讀取檔案，判斷是否有新內容，如果有的話將資料封裝到Event物件中，傳送給Channel，獲取最新偏移量值，再更新偏移量；如果日誌檔案中沒有新的內容，就讓執行緒sleep一下。

詳細程式碼如下：

package flume;
import org.apache.commons.io.FileUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
 
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * 自定義source，實現多點續傳
 */

public class MyFileSource extends AbstractSource implements Configurable,EventDrivenSource {

    private static final Logger logger = LoggerFactory.getLogger(MyFileSource.class);
    private String filePath;
    private String offsetPath;
    private Long interval;
    private String charset;
    private ExecutorService executor;
    private  MyRunnable myRunner;
    public void configure(Context context) {
         filePath = context.getString("filePath");
offsetPath = context.getString("offsetPath");
interval = context.getLong("interval",1000L);
charset = context.getString("charset","UTF-8");
}

    public synchronized void start(){

        //建立一個執行緒池       |      得到一個Channel物件
executor = Executors.newSingleThreadExecutor();
ChannelProcessor channelProcessor = getChannelProcessor();
//new一個Runnable的物件
myRunner = new MyRunnable(filePath,offsetPath,interval,charset,channelProcessor);
//將exector放入執行緒池中
executor.execute(myRunner);
        super.start();
}

    public synchronized void stop() {
        //停掉執行緒
myRunner.setFlag(false);
//停掉執行緒池
executor.shutdown();
        while (!executor.isTerminated()) {
            logger.debug("Waiting for exec executor service to stop");
            try {
                executor.awaitTermination(500, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
                logger.debug("Interrupted while waiting for exec executor service "
+ "to stop. Just exiting.");
Thread.currentThread().interrupt();
}
        }

        super.stop();
}
    public static class MyRunnable implements Runnable{

        private String filePath;
        private String offsetPath;
        private Long interval;
        private String charset;
        private ChannelProcessor channelProcessor;
        private Long offset = 0L;
        private File osfile;
        private boolean flag = true;
        private RandomAccessFile raf;
        public MyRunnable(String filePath, String offsetPath, Long interval, String charset, ChannelProcessor channelProcessor) {
            this.filePath = filePath;
            this.offsetPath = offsetPath;
            this.interval = interval;
            this.charset = charset;
            this.channelProcessor = channelProcessor;
//把偏移量檔案裝進File物件裡
osfile = new File(offsetPath);
//判斷是否有偏移量檔案,如果不存在就建立一個
if(!osfile.exists()){
                try {
                    osfile.createNewFile();
} catch (IOException e) {
                    logger.debug("create osfile error",e);
}
            }
            //存在的話，判斷檔案裡有沒有內容,先得到檔案中的內容轉為String
try {
                String offsetStr = FileUtils.readFileToString(osfile);
//如果有內容的話，就把內容轉為Long型別,沒有的話就初值
if(offsetStr!=null && !"".equals(offsetStr)){
                    offset = Long.parseLong(offsetStr);
}
            } catch (IOException e) {
                logger.debug("read offset error",e);
}

            //如果有偏移量，就接著讀，沒有的話就從頭讀      new一個隨機讀取檔案內容的物件
try {
                raf = new RandomAccessFile(filePath, "r");
raf.seek(offset);
} catch (FileNotFoundException e) {
                logger.debug("file not found error",e);
} catch (IOException e) {
                logger.debug("read offset error",e);
}
        }

        public void run() {

            while(flag) {
                //定期讀取檔案，是否有新內容
try {
                    String line = raf.readLine();
//將資料封裝進event物件
if(line!=null){
                        Event event = EventBuilder.withBody(line, Charset.forName(charset));
//event物件傳送給channel
channelProcessor.processEvent(event);
//獲取新的偏移量，再更新偏移量
offset = raf.getFilePointer();
FileUtils.writeStringToFile(osfile, offset + "");
}else {
                        Thread.sleep(interval);
}


                } catch (IOException e) {
                    logger.debug("read line error",e);
} catch (InterruptedException e) {
                    logger.debug("thread sleep error",e);
}
            }
        }

        public void setFlag(boolean flag){
            this.flag  = flag;
}
    }
}