多线程压缩

import java.io.BufferedInputStream;  import java.io.BufferedOutputStream;  import java.io.Closeable;  import java.io.File;  import java.io.FileInputStream;  import java.io.FileOutputStream;  import java.io.IOException;  import java.io.InputStream;  import java.io.OutputStream;  import java.util.Arrays;  import java.util.Queue;  import java.util.concurrent.ArrayBlockingQueue;  import java.util.concurrent.BlockingQueue;  import java.util.concurrent.ConcurrentLinkedQueue;  import java.util.concurrent.atomic.AtomicInteger;  import java.util.zip.GZIPOutputStream;    class Setup {   static BlockingQueue<File> queue = new ArrayBlockingQueue<File>(2);   static Queue<File> fileQueue = new ConcurrentLinkedQueue<File>();   final static String path = "D:/000_test/c";   static AtomicInteger aInt = new AtomicInteger();     public static void main(String[] args) {    File file = new File(path);    if (file.isDirectory()) {     File[] files = file.listFiles();     fileQueue.addAll(Arrays.asList(files));    } else if (file.isFile()) {     fileQueue.add(file);    }    Producer p = new Producer();    Producer p1 = new Producer();    FileTransfer c1 = new FileTransfer();    new Thread(p).start();    new Thread(p1).start();    new Thread(c1).start();   }     static class Producer implements Runnable {    Producer() {    }      public void run() {     try {      while (!fileQueue.isEmpty()) {       aInt.incrementAndGet();       queue.put(compress(fileQueue.poll()));      }      System.out.println("finish");     } catch (InterruptedException ex) {      ex.printStackTrace();     }    }      private File compress(File input) {     InputStream in = null;     OutputStream out = null;     try {      in = new FileInputStream(input);      in = new BufferedInputStream(in);      File output = new File(input.getParentFile().getParent(), input.getName() + ".gz");      System.out.println("produce=" + Thread.currentThread().getName()+" compress1="+output.getAbsolutePath());      if (!output.exists()) {       out = new FileOutputStream(output);       out = new GZIPOutputStream(out);       out = new BufferedOutputStream(out);       int b;       while ((b = in.read()) != -1) {        out.write(b);       }       out.flush();      }     } catch (Exception e) {      e.printStackTrace();     } finally {      close(out);      close(in);     }     return input;    }      private void close(Closeable c) {     if (c != null) {      try {       c.close();      } catch (IOException e) {       e.printStackTrace();      }     }    }   }     static class FileTransfer implements Runnable {      public void run() {     while (true) {      try {       consume(queue.take());       if(aInt.decrementAndGet() == 0 && fileQueue.isEmpty()){        System.out.println("break..");        break;       }      } catch (InterruptedException ex) {      }     }    }      void consume(File x) {     System.out.println("consume=" + Thread.currentThread().getName() + " ==>"+x.getAbsolutePath());    }   }  }