介紹

什么是線程池？
線程的創(chuàng)建開銷很大，如果每一次線程使用后就丟棄，等下次需要使用的再重新創(chuàng)建，則帶來的浪費(fèi)就很嚴(yán)重。線程池解決了線程復(fù)用的問題，線程在執(zhí)行完任務(wù)以后并不是立即銷毀，而是繼續(xù)保留，可以在后續(xù)執(zhí)行其他任務(wù)。

概覽

實踐1

異步讀寫文件 使用singleThreadPool
任務(wù)描述： 現(xiàn)在有一個很大的csv文件（21G）, 內(nèi)容是1990-2016年的專利數(shù)據(jù)，希望將2012年及以后的數(shù)據(jù)提取出來，寫入一個新的csv文件中。關(guān)鍵點是：

• 如果讀一行寫一行，這樣總計1000萬行以上的數(shù)據(jù)，時間將會很長。解決：將讀取的文件寫入cache中，當(dāng)cache滿了以后，把cache的數(shù)據(jù)交給寫文件線程，cache清空，繼續(xù)讀取文件。
• write線程雖然和read線程是獨立的，但是write線程必須始終只有一個。如果多個的話寫文件的順序會亂掉，多個線程競爭IO，也未必會帶來性能的提升。（一個引出問題：多線程寫文件，線程鎖的問題）
• 用singleThreadPool控制寫文件操作，能夠保證寫文件線程始終是同一個，如果新的任務(wù)提交而舊的任務(wù)還沒有完成，singleThreadPool也可以阻塞等待（如何實現(xiàn)的？）

master, 核心部分

package Thread.multi_rw;
import java.io.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 對大文件（21G csv）在多線程環(huán)境下讀寫
 * 將年份時間是在2012以后的按照行寫入新的文件
 */
public class WriteAndReadMaster {
    private File inputFile = null;
    private String outputFileName = null;
    private LineCache cache;
    private int readExceptionCount = 0;

    public WriteAndReadMaster(String sourcePath, String outfileName) {
        inputFile = new File(sourcePath);
        outputFileName = outfileName;
        this.cache = new LineCache();
    }

    public void execute() {
        System.out.println("Start to execute...");
        long startTime = System.currentTimeMillis();
        try {
            BufferedReader reader = new BufferedReader(new FileReader(inputFile));
            ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
            String line = null;
            // 此處一行行讀取的，考慮是否可以一次讀取多行？
            while (reader.ready() && ((line = reader.readLine()) != null)) {
                String[] contents = line.split(",\"");
                String data;
                try {
                    data = contents[7];
                } catch (ArrayIndexOutOfBoundsException e) {
                    System.err.println("index exception, ignore it");
                    this.readExceptionCount += 1;
                    continue;
                }

                if (data != null) {
                    try {
                        String tmp = data.substring(0, 4);
                        int year = Integer.parseInt(tmp);
                        if (year >= 2012) {
                            cache.push(line);
                        }
                    } catch (Exception e) {
                        System.err.println("line error, ignore it..");
                        this.readExceptionCount += 1;
                        continue;
                    }

                    if (cache.isCacheFull()) {
                        WriteFileTask writeFileTask = new WriteFileTask(outputFileName,
                                cache.getData());
                        singleThreadPool.submit(writeFileTask);
                        cache.cleanUp();
                    }
                }
            }
            WriteFileTask writeFileTask = new WriteFileTask(outputFileName,
                    cache.getData());
            singleThreadPool.submit(writeFileTask);
            cache.cleanUp();

        } catch (FileNotFoundException e) {
            System.err.println("file not found error");
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.err.println("total read exception:" + this.readExceptionCount);
        }
    }


}

cache 緩存類

package Thread.multi_rw;


import java.util.ArrayList;

public class LineCache {
    private final int MAX_CACHE_LINE = 10000;
    private volatile int cnt;
    private ArrayList<String> data = null;
    public LineCache(){
        this.data = new ArrayList<>(MAX_CACHE_LINE);

        this.cnt = 0;
    }

    public LineCache cleanUp(){
        this.data = new ArrayList<>(MAX_CACHE_LINE);
        this.cnt = 0;
        return this;
    }

    public boolean isCacheFull(){
        if(data.size() >= MAX_CACHE_LINE){
            return true;
        }
        else return false;
    }

    public void push(String line){
        data.add(line);
    }

    public synchronized ArrayList<String> getData() {
        return data;
    }
}

負(fù)責(zé)寫文件的Runable 對象

package Thread.multi_rw;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;

public class WriteFileTask implements Runnable {

    private String fileName = null;
    private static volatile int totalWriteLines = 0;
    private ArrayList<String> writeData = null;
    public volatile static int exceptionCount = 0;
    public WriteFileTask(String fileName, ArrayList<String> data) {
        this.fileName = fileName;
        this.writeData = data;
        System.err.println("data size: " + data.size());
    }

    @Override
    public void run() {
        int cnt = 0;
        try {
            BufferedWriter writer = new BufferedWriter(new FileWriter(fileName, true));
            for (String line : this.writeData) {
                writer.write(line);
                writer.write('\n');
                totalWriteLines += 1;
                cnt += 1;
            }
        } catch (IOException e) {
            System.err.println("Write Exception");
            this.exceptionCount += 1;
            e.printStackTrace();
        } finally {
            System.out.println("線程: " + Thread.currentThread() +
                    " 寫入: " + cnt + "行，總計寫入: " + totalWriteLines + "行");
            System.err.println("總計寫入錯誤: " + exceptionCount);
        }
    }
}

參考文章

深入理解java之線程池