java不解压读取linux服务器上的zip以及tar.gz压缩文件

悠悠 2022-07-17 16:07 221阅读 0赞

本周接到一个活让读取linux服务器上大文件，按行读取后再拆分放入Redis中。由于是大文件，想考虑到效率的问题，计划是想使用多线程读取，但是查资料，读取文件I/O的效率只跟磁盘有关，故多线程无效，但是考虑到I/O太多会死，故还是使用了线程池，当线程大于某一个值时，停一会，在读取。  
1、首先是读取zip压缩文件  
由于直接使用jdk的java.util.zip.ZipFile以及java.util.zip.ZipEntry读取时，可能由于服务器上zip压缩文件的编码问题，导致读取失败。故使用了第三方jar包[ant.jar][]

public void readZipFile(String file) {
            FileWriterWithEncoding writer=null;
            try {
                writer=new FileWriterWithEncoding(new File("/home/www/test1/gz/zipfile.txt"),"utf-8",true);
            } catch (IOException e2) {
                // TODO Auto-generated catch block
                e2.printStackTrace();
            }
            String fileName = "";
            try {
                ZipFile zf = new ZipFile(new File(file));
                long timestamp = 0;
                LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
                final int maxsize = 10;
                ExecutorService executor = new ThreadPoolExecutor(maxsize, maxsize, 0l, TimeUnit.MILLISECONDS, queue);
                for (Enumeration<?> entries = zf.getEntries(); entries.hasMoreElements();) {
                    ZipEntry ze = ((ZipEntry) entries.nextElement());
                    if (ze.isDirectory()) {
                    } else {
                        fileName = ze.getName();
                        long size = ze.getSize();
                        timestamp = ze.getTime();
                        System.err.println("fileName>>"+fileName);
                        if (size > 0 ) {
                            BufferedReader br = new BufferedReader(new InputStreamReader(zf.getInputStream(ze), "utf-8"));
                            String line;
                            // sohu_地域_电视剧-日本_563.txt
                            // media_taggroup_tagname_tagId.txt
                            String[] names = fileName.split("_");
                            while ((line = br.readLine()) != null) {
                                //System.out.println(fileName + "--zip>>>" + line);
                                writer.write(line);
                                while (queue.size() >= maxsize) {
                                    try {
                                        Thread.sleep(10);
                                    } catch (InterruptedException e) {
                                        Log.OutException(e);
                                    }
                                }
                                Callable<String> worker = new MyCallable(id, tagid, media, album, chanel, video, tagname,
                                        timestamp);//此处为线程类
                                executor.submit(worker);
                                // list.add(submit);
    
                            }
                            updataFileStatus(fileName, file, 1);
                            br.close();
                        }
                    }
    
                }
                executor.shutdown();
    
                try {
                    executor.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
                } catch (InterruptedException e) {
                    Log.OutException(e);
                }
            } catch (IOException e1) {
                // TODO Auto-generated catch block
                updataFileStatus(fileName, file, 0);
                e1.printStackTrace();
            }
        }

2、读取tar.gz文件  
由于tar.gz是一层层的往里面读取，不像zip那样直接拿到压缩内文件的全路径，所以需要一层层的遍历后拿到文件的全路径，然后读取。支持的jar包为 [commons-compress-1.12.jar ][commons-compress-1.12.jar]

public  void readGzipFile(String targzFile){
    
            FileWriterWithEncoding writer=null;
            try {
                writer=new FileWriterWithEncoding(new File("/home/www/test1/gz/gzfile.txt"),"utf-8",true);
            } catch (IOException e2) {
                // TODO Auto-generated catch block
                e2.printStackTrace();
            }
    
               FileInputStream fis = null;
               ArchiveInputStream in = null;
               BufferedInputStream bufferedInputStream = null;
               String f="";
               try {
                   fis = new FileInputStream(targzFile);
                   GZIPInputStream is = new GZIPInputStream(new BufferedInputStream(fis));
                   in = new ArchiveStreamFactory().createArchiveInputStream("tar", is);
                   bufferedInputStream = new BufferedInputStream(in);
    
               TarArchiveEntry entry = null;
               LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
                final int maxsize = 10;
                ExecutorService executor = new ThreadPoolExecutor(maxsize, maxsize, 0l, TimeUnit.MILLISECONDS, queue);
               entry = (TarArchiveEntry) in.getNextEntry();
               long timestamp = 0;
               while (entry != null) {
                   long size =0;
                      String name = entry.getName();
                      String[] names = name.split("/");
                      String fileName = targzFile;
    
                      for(int i = 0;i<names.length;i++){
                          String str = names[i];
                          fileName = fileName + File.separator + str;
                          f=f+File.separator + str;
                      }
                      int time_stamp = 0;
                      if(!fileName.endsWith("/")){
                        // System.err.println("f>>>"+f+"<<<fileName>>>"+fileName);
                         size = entry.getSize();
                         timestamp = (entry.getModTime()).getTime()/1000;
    
                      }
    
                if (size > 0 ) {
                StringBuffer line = new StringBuffer(); 
                byte[] b = new byte[(int) entry.getSize()];
                int len = 0; 
                while ((len = in.read(b)) != -1){  
                    line.append(new String(b, 0, len, "utf-8"));  
                  }  
                writer.write(line.toString());
               // System.err.println("gzip>>>>"+line);
                            while (queue.size() >= maxsize) {
                                try {
                                    Thread.sleep(10);
                                } catch (InterruptedException e) {
                                    Log.OutException(e);
                                }
                            }
                            Callable<String> worker = new AiRuiUser(userId, gender, age, province, city,timestamp);//此处为线程类
                            executor.submit(worker);
                      }
                  }
                  updataFileStatus(f, targzFile, 1);
                  f="";
               }else{
                   f="";
               }
                entry = (TarArchiveEntry) in.getNextEntry();
              }
               in.close();
               fis.close();
               bufferedInputStream.close();
               //fileIn.close();
               executor.shutdown();
              }catch (Exception e1) {
                    // TODO Auto-generated catch block
                    updataFileStatus(f, targzFile, 0);
                    e1.printStackTrace();
              }
        }

两个多线程的类为：

class MyCallable implements Callable<String> {
        private String id;
        private String tagid;//
        private String media;
        private String album;// 专辑
        private String chanel;// 频道
        private String video;// 视频
        private String tagname;
        private long timestamp;
    
        public MyCallable(String id, String tagid, String media, String album, String chanel, String video, String tagname,
                long timestamp) {
            this.id = id;
            this.tagid = tagid;
            this.media = media;
            this.album = album;
            this.chanel = chanel;
            this.video = video;
            this.tagname = tagname;
            this.timestamp = timestamp;
        }
    
        @Override
        public String call() throws Exception {
            // TODO Auto-generated method stub
            AiRuiBean aiRuiBean = new AiRuiBean(tagid, media, album, chanel, video, tagname);
    
            // AiRuiSave.doSave(id, aiRuiBean);
             AiRuiSave.doSave(id,timestamp,aiRuiBean);//调用redis存储的方法
            // AiRuiSave.doSave(id, album,timestamp,aiRuiBean);
    
            return null;
        }
    }
    
    class AiRuiUser implements Callable<String>{
    
        private String userId;
        private String gender;//性别
        private String age;//年龄
        private String province;//省份
        private String city;//城市
        private long timestamp;
        public AiRuiUser(String userId, String gender, String age, String province, String city,long timestamp) {
            super();
            this.userId = userId;
            this.gender = gender;
            this.age = age;
            this.province = province;
            this.city = city;
            this.timestamp=timestamp;
        }
    
        @Override
        public String call() throws Exception {
            // TODO Auto-generated method stub
            AiRuiUserBean aiRuiUserBean=new AiRuiUserBean(gender, age, province, city);
            AiRuiSave.iwtuserSave(userId, timestamp, aiRuiUserBean);//调用redis存储的方法
            return null;
        }
    }

当然最后的程序是要放在服务器中执行。

[ant.jar]: http://download.csdn.net/detail/zhaojw_420/9618944
[commons-compress-1.12.jar]: http://download.csdn.net/detail/zhaojw_420/9618948