Java多线程实现快速切分文件的程序

前段时间需要进行大批量数据导入,DBA给提供的是CVS文件,但是每个CVS文件都好几个GB大小,直接进行load,数据库很慢还会产生内存不足的问题,为了实现这个功能,写了个快速切分文件的程序。

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import java.io.*;
import java.util.*;
import java.util.concurrent.*;

public class FileSplitUtil {

  private final static Logger log = LogManager.getLogger(FileSplitUtil.class);
  private static final long originFileSize = 1024 * 1024 * 100;// 100M
  private static final int blockFileSize = 1024 * 1024 * 64;// 防止中文乱码,必须取2的N次方
  /**
   * CVS文件分隔符
   */
  private static final char cvsSeparator = '^';
  public static void main(String args[]){
    long start = System.currentTimeMillis();
    try {
      String fileName = "D:\\csvtest\\aa.csv";
      File sourceFile = new File(fileName);
      if (sourceFile.length() >= originFileSize) {
        String cvsFileName = fileName.replaceAll("\\\\", "/");
        FileSplitUtil fileSplitUtil = new FileSplitUtil();
        List<String> parts=fileSplitUtil.splitBySize(cvsFileName, blockFileSize);
        for(String part:parts){
          System.out.println("partName is:"+part);
        }
      }
      System.out.println("总文件长度"+sourceFile.length()+",拆分文件耗时:" + (System.currentTimeMillis() - start) + "ms.");
    }catch (Exception e){
      log.info(e.getStackTrace());
    }

  }

  /**
   * 拆分文件
   *
   * @param fileName 待拆分的完整文件名
   * @param byteSize 按多少字节大小拆分
   * @return 拆分后的文件名列表
   */
  public List<String> splitBySize(String fileName, int byteSize)
      throws IOException, InterruptedException {
    List<String> parts = new ArrayList<String>();
    File file = new File(fileName);
    int count = (int) Math.ceil(file.length() / (double) byteSize);
    int countLen = (count + "").length();
    RandomAccessFile raf = new RandomAccessFile(fileName, "r");
    long totalLen = raf.length();
    CountDownLatch latch = new CountDownLatch(count);

    for (int i = 0; i < count; i++) {
      String partFileName = file.getPath() + "."
          + leftPad((i + 1) + "", countLen, '0') + ".cvs";
      int readSize=byteSize;
      long startPos=(long)i * byteSize;
      long nextPos=(long)(i+1) * byteSize;
      if(nextPos>totalLen){
        readSize= (int) (totalLen-startPos);
      }
      new SplitRunnable(readSize, startPos, partFileName, file, latch).run();
      parts.add(partFileName);
    }
    latch.await();//等待所有文件写完
    //由于切割时可能会导致行被切断,加工所有的的分割文件,合并行
    mergeRow(parts);
    return parts;
  }

  /**
   * 分割处理Runnable
   *
   * @author supeidong
   */
  private class SplitRunnable implements Runnable {
    int byteSize;
    String partFileName;
    File originFile;
    long startPos;
    CountDownLatch latch;
    public SplitRunnable(int byteSize, long startPos, String partFileName,
               File originFile, CountDownLatch latch) {
      this.startPos = startPos;
      this.byteSize = byteSize;
      this.partFileName = partFileName;
      this.originFile = originFile;
      this.latch = latch;
    }

    public void run() {
      RandomAccessFile rFile;
      OutputStream os;
      try {
        rFile = new RandomAccessFile(originFile, "r");
        byte[] b = new byte[byteSize];
        rFile.seek(startPos);// 移动指针到每“段”开头
        int s = rFile.read(b);
        os = new FileOutputStream(partFileName);
        os.write(b, 0, s);
        os.flush();
        os.close();
        latch.countDown();
      } catch (IOException e) {
        log.error(e.getMessage());
        latch.countDown();
      }
    }
  }

  /**
   * 合并被切断的行
   *
   * @param parts
   */
  private void mergeRow(List<String> parts) {
    List<PartFile> partFiles = new ArrayList<PartFile>();
    try {
      //组装被切分表对象
      for (int i=0;i<parts.size();i++) {
        String partFileName=parts.get(i);
        File splitFileTemp = new File(partFileName);
        if (splitFileTemp.exists()) {
          PartFile partFile = new PartFile();
          BufferedReader reader=new BufferedReader(new InputStreamReader(new FileInputStream(splitFileTemp),"gbk"));
          String firstRow = reader.readLine();
          String secondRow = reader.readLine();
          String endRow = readLastLine(partFileName);
          partFile.setPartFileName(partFileName);
          partFile.setFirstRow(firstRow);
          partFile.setEndRow(endRow);
          if(i>=1){
            String prePartFile=parts.get(i - 1);
            String preEndRow = readLastLine(prePartFile);
            partFile.setFirstIsFull(getCharCount(firstRow+preEndRow)>getCharCount(secondRow));
          }

          partFiles.add(partFile);
          reader.close();
        }
      }
      //进行需要合并的行的写入
      for (int i = 0; i < partFiles.size() - 1; i++) {
        PartFile partFile = partFiles.get(i);
        PartFile partFileNext = partFiles.get(i + 1);
        StringBuilder sb = new StringBuilder();
        if (partFileNext.getFirstIsFull()) {
          sb.append("\r\n");
          sb.append(partFileNext.getFirstRow());
        } else {
          sb.append(partFileNext.getFirstRow());
        }
        writeLastLine(partFile.getPartFileName(),sb.toString());
      }
    } catch (Exception e) {
      log.error(e.getMessage());
    }
  }

  /**
   * 得到某个字符出现的次数
   * @param s
   * @return
   */
  private int getCharCount(String s) {
    int count = 0;
    for (int i = 0; i < s.length(); i++) {
      if (s.charAt(i) == cvsSeparator) {
        count++;
      }
    }
    return count;
  }

  /**
   * 采用BufferedInputStream方式读取文件行数
   *
   * @param filename
   * @return
   */
  public int getFileRow(String filename) throws IOException {
    InputStream is = new BufferedInputStream(new FileInputStream(filename));
    byte[] c = new byte[1024];
    int count = 0;
    int readChars = 0;
    while ((readChars = is.read(c)) != -1) {
      for (int i = 0; i < readChars; ++i) {
        if (c[i] == '\n')
          ++count;
      }
    }
    is.close();
    return count;
  }

  /**
   * 读取最后一行数据
   * @param filename
   * @return
   * @throws IOException
   */
  private String readLastLine(String filename) throws IOException {
    // 使用RandomAccessFile , 从后找最后一行数据
    RandomAccessFile raf = new RandomAccessFile(filename, "r");
    long len = raf.length();
    String lastLine = "";
    if(len!=0L) {
      long pos = len - 1;
      while (pos > 0) {
        pos--;
        raf.seek(pos);
        if (raf.readByte() == '\n') {
          lastLine = raf.readLine();
          lastLine=new String(lastLine.getBytes("8859_1"), "gbk");
          break;
        }
      }
    }
    raf.close();
    return lastLine;
  }
  /**
   * 修改最后一行数据
   * @param fileName
   * @param lastString
   * @return
   * @throws IOException
   */
  private void writeLastLine(String fileName,String lastString){
    try {
      // 打开一个随机访问文件流,按读写方式
      RandomAccessFile randomFile = new RandomAccessFile(fileName, "rw");
      // 文件长度,字节数
      long fileLength = randomFile.length();
      //将写文件指针移到文件尾。
      randomFile.seek(fileLength);
      //此处必须加gbk,否则会出现写入乱码
      randomFile.write(lastString.getBytes("gbk"));
      randomFile.close();
    } catch (IOException e) {
      log.error(e.getMessage());
    }
  }
  /**
   * 左填充
   *
   * @param str
   * @param length
   * @param ch
   * @return
   */
  public static String leftPad(String str, int length, char ch) {
    if (str.length() >= length) {
      return str;
    }
    char[] chs = new char[length];
    Arrays.fill(chs, ch);
    char[] src = str.toCharArray();
    System.arraycopy(src, 0, chs, length - src.length, src.length);
    return new String(chs);
  }

  /**
   * 合并文件行内部类
   */
  class PartFile {
    private String partFileName;
    private String firstRow;
    private String endRow;
    private boolean firstIsFull;

    public String getPartFileName() {
      return partFileName;
    }

    public void setPartFileName(String partFileName) {
      this.partFileName = partFileName;
    }

    public String getFirstRow() {
      return firstRow;
    }

    public void setFirstRow(String firstRow) {
      this.firstRow = firstRow;
    }

    public String getEndRow() {
      return endRow;
    }

    public void setEndRow(String endRow) {
      this.endRow = endRow;
    }

    public boolean getFirstIsFull() {
      return firstIsFull;
    }

    public void setFirstIsFull(boolean firstIsFull) {
      this.firstIsFull = firstIsFull;
    }
  }

}

以上就是本文的全部内容,希望对大家学习java程序设计有所帮助。

(0)

相关推荐

  • MySQL切分查询用法分析

    本文实例讲述了MySQL切分查询用法.分享给大家供大家参考,具体如下: 对于大查询有时需要'分而治之',将大查询切分为小查询: 每个查询功能完全一样,但只完成原来的一小部分,每次查询只返回一小部分结果集. 删除旧的数据就是一个很好地例子.定期清理旧数据时,如果一条sql涉及了大量的数据时,可能会一次性锁住多个表或行,耗费了大量的系统资源,却阻塞了其他很多小的但重要的查询.将一个大得DELETE语句切分为较小的查询时,可以尽量减少影响msql的性能,同时减少mysql复制造成的延迟. 例如,每个月

  • 用JavaScript实现字符串切分功能

    字符串切分 function getmulti(oldstr) { var newstr=oldstr;//字段内容 var i=0,j=0,t=1;//判断是否有多个部分内容 var foreindex;//记录前一个分隔符的位置 var index,depchar;//记录当前分隔符及其位置 var linkstr="";//链接方式 var astr = new Array(10); var index1 = newstr.indexOf(","); var

  • PHP explode()函数用法、切分字符串

    复制代码 代码如下: <? // ### 切分字符串 #### function jb51netcut($start,$end,$file){ $content=explode($start,$file); $content=explode($end,$content[1]); return $content[0]; } ?> explode定义和用法 explode() 函数把字符串分割为数组. 语法 explode(separator,string,limit) 参数 描述 separat

  • 15个高级Java多线程面试题及回答

    Java 线程面试问题 在任何Java面试当中多线程和并发方面的问题都是必不可少的一部分.如果你想获得任何股票投资银行的前台资讯职位,那么你应该准备很多关于多线程的问题.在投资银行业务中多线程和并发是一个非常受欢迎的话题,特别是电子交易发展方面相关的.他们会问面试者很多令人混淆的Java线程问题.面试官只是想确信面试者有足够的Java线程与并发方面的知识,因为候选人中有很多只浮于表面.用于直接面向市场交易的高容量和低延时的电子交易系统在本质上是并发的.下面这些是我在不同时间不同地点喜欢问的Jav

  • java基本教程之java线程等待与java唤醒线程 java多线程教程

    本章,会对线程等待/唤醒方法进行介绍.涉及到的内容包括:1. wait(), notify(), notifyAll()等方法介绍2. wait()和notify()3. wait(long timeout)和notify()4. wait() 和 notifyAll()5. 为什么notify(), wait()等函数定义在Object中,而不是Thread中 wait(), notify(), notifyAll()等方法介绍在Object.java中,定义了wait(), notify()

  • Java多线程的用法详解

    1.创建线程 在Java中创建线程有两种方法:使用Thread类和使用Runnable接口.在使用Runnable接口时需要建立一个Thread实例.因此,无论是通过Thread类还是Runnable接口建立线程,都必须建立Thread类或它的子类的实例.Thread构造函数: public Thread( );  public Thread(Runnable target);  public Thread(String name);  public Thread(Runnable target

  • JS字符串的切分用法实例

    本文实例讲述了JS字符串的切分用法.分享给大家供大家参考,具体如下: <script type="text/javascript"> <!-- var str="x:1;y:2;z:3"; var sarry=new Array(str.length,2); var fsa=str.split(";"); for(i=0;i<fsa.length;i++) { var temp=fsa[i].split(":&q

  • 详解Nginx服务器中的Socket切分

    NGINX发布的1.9.1版本引入了一个新的特性:允许使用SO_REUSEPORT套接字选项,该选项在许多操作系统的新版本中是可用的,包括DragonFly BSD和Linux(内核版本3.9及以后).该套接字选项允许多个套接字监听同一IP和端口的组合.内核能够在这些套接字中对传入的连接进行负载均衡. (对于NGINX Plus客户,此功能将在年底发布的版本7中出现) SO_REUSEPORT选项有许多潜在的实际应用.其他服务也可以使用它来简单实现执行中的滚动升级(Nginx已经通过不同的办法支

  • java向多线程中传递参数的三种方法详细介绍

    在传统的同步开发模式下,当我们调用一个函数时,通过这个函数的参数将数据传入,并通过这个函数的返回值来返回最终的计算结果.但在多线程的异步开发模式下,数据的传递和返回和同步开发模式有很大的区别.由于线程的运行和结束是不可预料的,因此,在传递和返回数据时就无法象函数一样通过函数参数和return语句来返回数据.本文就以上原因介绍了几种用于向线程传递数据的方法,在下一篇文章中将介绍从线程中返回数据的方法. 欲先取之,必先予之.一般在使用线程时都需要有一些初始化数据,然后线程利用这些数据进行加工处理,并

  • python实现按行切分文本文件的方法

    本文实例讲述了python实现按行切分文本文件的方法.分享给大家供大家参考,具体如下: python脚本利用shell命令来实现文本的操作, 这些命令大大减少了我们的代码量. 比如按行切分文件并返回切分后得到的文件列表,可以利用内建的split命令进行切分.为了返回得到的文件列表名,可以先将文件切分到自建的子目录中,然后通过os.listdir获取所有文件,再将这些文件移到上一级目录(即函数参数指定的新目录),删除自建子目录,最后返回该文件名列表. 代码如下,如发现问题欢迎指正: # 创建新路径

随机推荐