多線程壓縮

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());
        }
    }
}