多线程读取单文件

package com.qwx.test;

import java.io.*;

/**
 * @ClassName: ReadFile
 * @Description: java类作用描述
 * @Author: qiaowenxuan
 * @CreateDate: 2021/8/16$ 18:36$
 * @Version: 1.0
 */
public class ReadFile {
    public static void main(String[] args) {
        File file = new File("F://xaa.txt");
        ReaderThread thread1 = new ReaderThread(file);
        ReaderThread thread2 = new ReaderThread(file);

        new Thread(thread1,"thread1").start();
        new Thread(thread2,"thread2").start();
    }
}

class Reader {
    public static void read(File file) throws IOException {
        FileInputStream fileInputStream = new FileInputStream(file);
        InputStreamReader reader = new InputStreamReader(fileInputStream);
        BufferedReader bufferedReader = new BufferedReader(reader);
        String line = null;
        while (null != (line = bufferedReader.readLine())) {
            System.out.println(line);
        }
        bufferedReader.close();
    }
}

class ReaderThread implements Runnable {

    private File file;
    public ReaderThread(File file) {
        this.file = file;
    }

    @Override
    public void run() {
        try {
            Reader.read(file);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

多线程读取文件列表

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import redis.clients.jedis.Jedis;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class FavFileUtil {
    public static List getFileList(String filePath) {
        List fileList = new ArrayList();
        File file = new File(filePath);
        if (!file.isDirectory()) {
            System.out.println("文件【" + file.getName() + "】：" + file.getAbsolutePath());
            fileList.add(file);
        } else {
            System.out.println("文件夹【" + file.getName() + "】：" + file.getAbsolutePath());
            File[] files = file.listFiles();
            for (int i = 0; i < files.length; i++) {
                if (!files[i].isDirectory()) {
                    System.out.println("　　文件【" + files[i].getName() + "】：");
                    fileList.add(files[i]);
                } else if (files[i].isDirectory()) {
                    getFileList(files[i].getAbsolutePath());
                }
            }
        }
        return fileList;
    }

    public static List subFiles(List list, int startIndex, int endIndex) {
        if (endIndex > list.size()) {
            return list.subList(startIndex, list.size());
        }
        return list.subList(startIndex, endIndex);
    }

    static int i = 0;
    public static void readFile(File file) throws IOException {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        //读取文件
        FileInputStream is = new FileInputStream(file);

        //获取文件的字符流
        InputStreamReader isr = new InputStreamReader(is);

        //缓冲
        BufferedReader br = new BufferedReader(isr);
        String line = null;
        while ((line = br.readLine()) != null) {
            String[] split = line.split("----");
            String key = split[0];
            String val = split[1];
            //if (null == key) key = "";
            //if(null == val) val ="";
            jedis.set(key,val);
            System.out.println(i);
            System.out.println(Thread.currentThread().getName() + ":::" + line);
            System.out.println(Thread.currentThread().getName() + ":::" + key+"----"+val);
            i++;
        }
        br.close();
    }

}

import java.io.File;
import java.io.IOException;
import java.util.List;

public class FavThreadUtil implements Runnable {
    private List<File> fileList;
    public FavThreadUtil(List fileList) {
        this.fileList = fileList;
    }
    @Override
    public void run() {
        for(File file : fileList){
            try {
                FavFileUtil.readFile(file);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}

一、背景知识

1. 多线程可以提高任务的执行效率，尤其是CPU计算密集型任务
2. 读写文件是IO密集型任务
3. 过多的线程执行同一个任务，并不一定能提高效率，因为线程切换，需要耗时。再比如IO密集型任务，IO是瓶颈，并不是线程越多，IO会越快。
4. 开多线程去执行任务，需要综合考虑实际情况。

二、问题
最近项目中遇到，安装APK（其中有拷贝apk文件到具体路劲的逻辑），考虑到优化，缩短这里的apk文件拷贝时间。优化过程，使用了断点拷贝和多线程拷贝。

1. 断点拷贝：使用Java的随机读取RandomAccessFile来实现。通过可以seek()到上次断点的位置，接着继续读写。

       /**
    * 随机读取，支持断点读写
    *
    * @param src 源文件
    * @param dst 目标文件
    * @throws IOException
    */
   @WorkerThread
   public static void copyFileRandomSingleThread(String src, String dst) throws IOException {
   
       RandomAccessFile srcFile = new RandomAccessFile(src, "rw");
       RandomAccessFile dstFile = new RandomAccessFile(dst, "rw");
   
       try {
           long currentDstLength = dstFile.length();
           srcFile.seek(currentDstLength);
           dstFile.seek(currentDstLength);
           Log.d(TAG, "Read " + src + " and Write " + dst + " from " + currentDstLength);
   
           byte[] buffer = new byte[8192];
           int read = -1;
           while ((read = srcFile.read(buffer)) != -1) {
               dstFile.write(buffer, 0, read);
           }
       } catch (IOException e) {
           Log.d(TAG, "copyFileRandom exception, ", e);
       } finally {
           try {
               srcFile.close();
           } catch (IOException e) {
               Log.d(TAG, "copyFileRandom close exception, ", e);
           }
           try {
               dstFile.close();
           } catch (IOException e) {
               Log.d(TAG, "copyFileRandom close exception, ", e);
           }
       }
   
   
   }
   
   

2. 多线程拷贝（如果必须在多线程拷贝完文件后，再去做其他任务，此处需要用到CountDownlatch进行同步）
   将File分片，然后每个线程去读不同的片。

   /**
    * 多线程拷贝文件
    *
    * @param sourcePath
    * @param targetPath
    * @param threadNums
    */
   private static void copyFileRandomMultiThread(String sourcePath, String targetPath, int threadNums) {
       long totalFileLength = new File(sourcePath).length();
       long currentTargetLength = new File(targetPath).length();
   
   
       long segmentLength = (totalFileLength - currentTargetLength) / (threadNums);
       //同步锁，全部写完，才能继续下一步处理
       CountDownLatch mCountDownLatch;
       //如果不可以整除,则需要多一个线程
       if (totalFileLength % threadNums != 0) {
           mCountDownLatch = new CountDownLatch(threadNums + 1);
       } else {
           mCountDownLatch = new CountDownLatch(threadNums);
       }
   
       int i;
       Log.d(TAG, "threadNums-->" + mCountDownLatch.getCount());
       long start = System.currentTimeMillis();
       for (i = 0; i < threadNums; i++) {
   
           sThreadPool.execute(new CopyFileRunnable(mCountDownLatch, sourcePath, targetPath, i * segmentLength + currentTargetLength, (i + 1) * segmentLength));
   
       }
   
       if (totalFileLength % threadNums != 0) {
           sThreadPool.execute(new CopyFileRunnable(mCountDownLatch, sourcePath, targetPath, i * segmentLength + currentTargetLength, totalFileLength));
       }
       try {
           mCountDownLatch.await();
           Log.d(TAG, "multi thread copyFile costs-->" + (System.currentTimeMillis() - start));
       } catch (InterruptedException e) {
           Log.d(TAG, "countDownLatch exception: ", e);
       }
   }
   
   
   private static class CopyFileRunnable implements Runnable {
       private RandomAccessFile in;
       private RandomAccessFile out;
       private long start;
       private long end;
       private CountDownLatch latch;
   
       /**
        * @param countDownLatch 同步锁
        * @param in             源文件地址
        * @param out            目标文件地址
        * @param start          分段复制的开始位置
        * @param end            分段复制的结束位置
        */
       CopyFileRunnable(CountDownLatch countDownLatch, String in, String out,
                        long start, long end) {
           this.start = start;
           this.end = end;
           try {
               this.in = new RandomAccessFile(in, "rw");
               this.out = new RandomAccessFile(out, "rw");
           } catch (FileNotFoundException e) {
               e.printStackTrace();
           }
   
           this.latch = countDownLatch;
       }
   
       @Override
       public void run() {
           try {
               in.seek(start);
               out.seek(start);
               android.util.Log.d(TAG, "multi thread from: " + start + "; end:" + end);
               int hasRead = 0;
               byte[] buff = new byte[8 * 1024];
               while (start < end && (hasRead = in.read(buff)) != -1) {
                   start += hasRead;
                   out.write(buff, 0, hasRead);
               }
   
           } catch (IOException e) {
               e.printStackTrace();
           } finally {
               try {
                   in.close();
                   out.close();
               } catch (IOException e) {
                   e.printStackTrace();
               }
   
               latch.countDown();
           }
   
       }
   }
   

3. 完整的程序代码

   public class Utils {
   
   private static final String TAG = Utils.class.getSimpleName();
   
   private static final int CPU_SIZE = Runtime.getRuntime().availableProcessors();
   private static final int CORE_POOL_SIZE = 2 * CPU_SIZE + 1;
   private static final int MAX_POLL_SIZE = CORE_POOL_SIZE;
   private static final ThreadFactory THREAD_FACTORY = new ThreadFactory() {
       private AtomicInteger index = new AtomicInteger(1);
   
       @Override
       public Thread newThread(@NonNull Runnable r) {
   
           Thread thread = new Thread(r);
           thread.setName("Sandbox_Utils-thread-" + index.getAndIncrement());
           return thread;
       }
   };
   private static final int DEFAULT_THREAD_NUM = 1;
   private static ExecutorService sThreadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POLL_SIZE, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(128), THREAD_FACTORY);
   
   
   /**
    * 随机读取，支持断点读写
    *
    * @param src 源文件
    * @param dst 目标文件
    * @throws IOException
    */
   @WorkerThread
   public static void copyFileRandomSingleThread(String src, String dst) throws IOException {
   
       RandomAccessFile srcFile = new RandomAccessFile(src, "rw");
       RandomAccessFile dstFile = new RandomAccessFile(dst, "rw");
   
       try {
           long currentDstLength = dstFile.length();
           srcFile.seek(currentDstLength);
           dstFile.seek(currentDstLength);
           Log.d(TAG, "Read " + src + " and Write " + dst + " from " + currentDstLength);
   
           byte[] buffer = new byte[8192];
           int read = -1;
           while ((read = srcFile.read(buffer)) != -1) {
               dstFile.write(buffer, 0, read);
           }
       } catch (IOException e) {
           Log.d(TAG, "copyFileRandom exception, ", e);
       } finally {
           try {
               srcFile.close();
           } catch (IOException e) {
               Log.d(TAG, "copyFileRandom close exception, ", e);
           }
           try {
               dstFile.close();
           } catch (IOException e) {
               Log.d(TAG, "copyFileRandom close exception, ", e);
           }
       }
   
   
   }
   
   
   /**
    * copy 文件单线程
    *
    * @param sourcePath
    * @param targetPath
    */
   public static void copyFileRandom(String sourcePath, String targetPath) {
       copyFileRandom(sourcePath, targetPath, 1);
   }
   
   /**
    * @param sourcePath 源文件路径
    * @param targetPath 目标文件路径
    * @param threadNums 设定的线程数
    */
   @WorkerThread
   public static void copyFileRandom(String sourcePath, String targetPath, int threadNums) {
   
       if (threadNums <= 0) {
           threadNums = DEFAULT_THREAD_NUM;
       }
   
       //单线程拷贝
       if (threadNums == 1) {
           try {
               copyFileRandomSingleThread(sourcePath, targetPath);
           } catch (IOException e) {
               e.printStackTrace();
           }
           //多线程拷贝
       } else {
           copyFileRandomMultiThread(sourcePath, targetPath, threadNums);
       }
   
   
   }
   
   /**
    * 多线程拷贝文件
    *
    * @param sourcePath
    * @param targetPath
    * @param threadNums
    */
   private static void copyFileRandomMultiThread(String sourcePath, String targetPath, int threadNums) {
       long totalFileLength = new File(sourcePath).length();
       long currentTargetLength = new File(targetPath).length();
   
   
       long segmentLength = (totalFileLength - currentTargetLength) / (threadNums);
       //同步锁，全部写完，才能继续下一步处理
       CountDownLatch mCountDownLatch;
       //如果不可以整除,则需要多一个线程
       if (totalFileLength % threadNums != 0) {
           mCountDownLatch = new CountDownLatch(threadNums + 1);
       } else {
           mCountDownLatch = new CountDownLatch(threadNums);
       }
   
       int i;
       Log.d(TAG, "threadNums-->" + mCountDownLatch.getCount());
       long start = System.currentTimeMillis();
       for (i = 0; i < threadNums; i++) {
   
           sThreadPool.execute(new CopyFileRunnable(mCountDownLatch, sourcePath, targetPath, i * segmentLength + currentTargetLength, (i + 1) * segmentLength));
   
       }
   
       if (totalFileLength % threadNums != 0) {
           sThreadPool.execute(new CopyFileRunnable(mCountDownLatch, sourcePath, targetPath, i * segmentLength + currentTargetLength, totalFileLength));
       }
       try {
           mCountDownLatch.await();
           Log.d(TAG, "multi thread copyFile costs-->" + (System.currentTimeMillis() - start));
       } catch (InterruptedException e) {
           Log.d(TAG, "countDownLatch exception: ", e);
       }
   }
   
   
   private static class CopyFileRunnable implements Runnable {
       private RandomAccessFile in;
       private RandomAccessFile out;
       private long start;
       private long end;
       private CountDownLatch latch;
   
       /**
        * @param countDownLatch 同步锁
        * @param in             源文件地址
        * @param out            目标文件地址
        * @param start          分段复制的开始位置
        * @param end            分段复制的结束位置
        */
       CopyFileRunnable(CountDownLatch countDownLatch, String in, String out,
                        long start, long end) {
           this.start = start;
           this.end = end;
           try {
               this.in = new RandomAccessFile(in, "rw");
               this.out = new RandomAccessFile(out, "rw");
           } catch (FileNotFoundException e) {
               e.printStackTrace();
           }
   
           this.latch = countDownLatch;
       }
   
       @Override
       public void run() {
           try {
               in.seek(start);
               out.seek(start);
               android.util.Log.d(TAG, "multi thread from: " + start + "; end:" + end);
               int hasRead = 0;
               byte[] buff = new byte[8 * 1024];
               while (start < end && (hasRead = in.read(buff)) != -1) {
                   start += hasRead;
                   out.write(buff, 0, hasRead);
               }
   
           } catch (IOException e) {
               e.printStackTrace();
           } finally {
               try {
                   in.close();
                   out.close();
               } catch (IOException e) {
                   e.printStackTrace();
               }
   
               latch.countDown();
           }
   
       }
   }