FileReadLineThread文件行读取线程

import org.apache.log4j.Logger;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;

/**
 * object: FileReadLineThread
 * author: 程晓鹏
 * description: 文件读取行线程类
 * date: 2018-04-13 16:51
 */
public class FileReadLineThread extends Thread implements Closeable{
    protected static final Logger log = Logger.getLogger(FileReadLineThread.class);
    private ReadWriteBooleanLock isRun; //是否运行
    private ReadWriteBooleanLock isBreakWhile; //是否跳出while循环
    private IFileReadLineNotify readLineNotify;
    private List<ReadFileInfo> files; //待处理文件集合
    private long closeValidSleepTime; //关闭验证时休眠时间
    private long closeStartTime; //关闭开始时间

    /**
     * 默认构造函数
     */
    public FileReadLineThread(){
        this.setName(this.getClass().getName().toString()); //设置线程名
        this.files = new ArrayList<ReadFileInfo>();
        this.isRun = new ReadWriteBooleanLock(true);
        this.isBreakWhile = new ReadWriteBooleanLock(false);
        this.closeValidSleepTime = 5;
        this.closeStartTime = 0;
    }

    public void setReadLineNotify(IFileReadLineNotify readLineNotify) {
        this.readLineNotify = readLineNotify;
    }

    private boolean isRun(){
        return this.isRun.read().booleanValue();
    }

    private boolean isBreakWhile(){
        return this.isBreakWhile.read().booleanValue();
    }

    @Override
    public void run() {
        exec();
    }

    /**
     * 执行操作
     */
    private void exec(){
        while(this.isRun()){
            this.exec0();
            synchronized (this.isRun){
                try {
                    if(!this.isRun() && this.files.size()==0){
                        break;
                    }
                    this.isRun.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        this.isBreakWhile.write(true);
        this.exec0(); //再次验证，是否还有未处理的数据
        long endTime = System.currentTimeMillis(); //结束时间
        this.finishCloseLog(endTime-this.closeStartTime);
    }

    /**
     * 执行文件读取
     */
    private void exec0(){
        while(this.files.size()>0){
            ReadFileInfo file = this.files.remove(0);
            this.execRead(file);
        }
    }

    /**
     * 执行读取
     * @param file 文件信息
     */
    private void execRead(ReadFileInfo file){
        long startTime = System.currentTimeMillis(); //开始时间
        RandomAccessFile raf = null;
        try {
            raf = new RandomAccessFile(file.getFilePath(),"r");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        long lLine = 0; //行数
        while(true){
            String strLine = null;
            try {
                strLine = raf.readLine(); //读取一行
                if(strLine != null){
                    String strLineData = this.formatCharsetData(strLine, file.getCharsetName()); //进行字符集转化
                    this.readLineNotify.fileReadLineNotify(file.getFileKey(), file.getFilePath(), ++lLine, strLineData); //进行数据通知
                }else{
                    raf.close();
                    long endTime = System.currentTimeMillis(); //结束时间
                    this.readLineNotify.fileReadLineFinishNotify(file.getFileKey(), file.getFilePath(), lLine, endTime-startTime);
                    this.printLog(file.getFileKey(), file.getFilePath(), lLine,endTime-startTime, this.files.size());
                    break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 打印日志
     * @param fileKey 文件KEY
     * @param filePath 文件路径
     * @param fileLines 文件行数
     * @param useTime 执行用时
     * @param noReadFiles 剩余没有读取的文件数
     */
    public void printLog(String fileKey, String filePath, long fileLines, long useTime, int noReadFiles){
        if(noReadFiles>0){
            log.info("FileKey: "+fileKey +" 文件路径: " + filePath +" 文件行数:" + String.valueOf(fileLines) + " 读取用时："+ String.valueOf(useTime)+"ms. [未处理：" +String.valueOf(noReadFiles)+"个]");
        }else{
            log.info("FileKey: "+fileKey +" 文件路径: " + filePath +" 文件行数:" + String.valueOf(fileLines) + " 读取用时："+ String.valueOf(useTime)+"ms. [处理完成]");
        }
    }

    /**
     * 格式化字符集数据
     * @param data　数据
     * @param charset　字符集
     * @return 格式化后的字符串数据
     */
    private String formatCharsetData(String data, String charset){
        String result = null;
        byte[] byteData = null;
        try {
            byteData = data.getBytes("ISO-8859-1");
            if(charset !=null && charset.length()>0){
                result = new String(byteData,charset);
            }
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            result = data;
        }

        return result;
    }

    /**
     * 读取文件
     * @param filePath　文件路径
     * @param charsetName 字符集名称
     */
    public void readFile(String filePath, String charsetName){
        this.readFile("SystemDefault", filePath, charsetName);
    }

    /**
     * 读取文件
     * @param fileKey 文件KEY
     * @param filePath 文件路径
     * @param charsetName 文件的字符集
     */
    public void readFile(String fileKey, String filePath, String charsetName){
        if(this.isRun()) {
            ReadFileInfo file = new ReadFileInfo();
            file.setCharsetName(charsetName);
            file.setFileKey(fileKey);
            file.setFilePath(filePath);
            this.files.add(file);

            synchronized (this.isRun) {
                this.isRun.notifyAll();
            }
        }
    }

    public void setCloseValidSleepTime(long closeValidSleepTime) {
        this.closeValidSleepTime = closeValidSleepTime;
    }

    public void beginCloseLog(){
        log.info("开始进行，文件读取线程, 关闭操作... ...");
    }

    public void finishCloseLog(long useTime){
        log.info("文件读取线程, 关闭完成. [用时:"+String.valueOf(useTime)+"ms]");
    }

    @Override
    public void close() {
        this.closeStartTime = System.currentTimeMillis(); //关闭操作，开始时间
        this.beginCloseLog();
        synchronized (this.isRun) {
            this.isRun.write(false);
            this.isRun.notifyAll();
        }

        while(!(this.files.size()==0 && this.isBreakWhile())) {
            try {
                Thread.sleep(this.closeValidSleepTime);
                if(!this.isBreakWhile()) {
                    synchronized (this.isRun) {
                        this.isRun.notifyAll();
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}