java线程池的应用浅析

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * 保存每次主线程读上来的字符串数组
 * @author winkey
 *
 */
public class BatchDataArrHolder {

    private static final int QUEUE_SIZE = 1000000;

    private static Queue<String[]> DATA_QUEUE = new ArrayBlockingQueue<String[]>(QUEUE_SIZE);

    public static void push(String[] data) {
        DATA_QUEUE.add(data);
    }

    public static String[] take(){
        try {
            return ((ArrayBlockingQueue<String[]>)DATA_QUEUE).take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
    
    public static boolean isEmpty(){
        return DATA_QUEUE.isEmpty();
    }
}

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 多线程 字符流
 * @author winkey
 *
 */
public class MyThread2 extends Thread{
    public static SortArray sortArray = new SortArray(10000) ;
    
    
    public void run(){  
        //这个方法这么写对吗？？？
        while(true){
            String[] batchDataArr = BatchDataArrHolder.take();
            if (null == batchDataArr) continue;
            if (null != batchDataArr){
                for(int i=0;i<batchDataArr.length;i++){
                    String str = batchDataArr[i];
                    Integer numAsInt = Integer.valueOf(str);
                    System.out.println(Thread.currentThread().getName()+"__"+numAsInt);
　　　　　　　　　　　　//setBit方法定义在排序类里面，我在setBit方法上加了关键字synchronized，这样对吗？？
                    sortArray.setBit(numAsInt, 1);
                }
            }
        }
    } 
    
    public static void main(String[] args) throws Exception {    
        BufferedReader in = new BufferedReader(new FileReader("E:\\b.txt"));
        BufferedWriter out = new BufferedWriter(new FileWriter("E:\\cc.txt"));
        BatchDataArrHolder holder = new BatchDataArrHolder();
        int readLength = 0;
        char[] chars = new char[5];
        String tail=null;
        String batchData;
        
        try{
            //创建一个可重用固定线程数的线程池
            ExecutorService pool = Executors.newFixedThreadPool(2);//两个子线程
            MyThread2 mt1 = new MyThread2();
            MyThread2 mt2 = new MyThread2();
　　　　　　　//主线程读数据前，子线程就启动，分析数据，这样写对吗？
            pool.execute(mt1);
            pool.execute(mt2);
             while(true){
                readLength = in.read(chars, 0, chars.length);
                //没有读上来数据说明上一次读取数据已读完，不再处理
                if(readLength == 0)
                    break;
                boolean isEnd = false;
                //读上来的数据少于数组长度，说明这一次已读完，处理完这次后不再继续读取
                if(readLength < chars.length){
                    //System.out.println(String.valueOf(chars).substring(0, readLength));
                    batchData = String.valueOf(chars).substring(0, readLength);
                    isEnd = true;
                }else{
                    //System.out.println(String.valueOf(chars));
                    batchData = String.valueOf(chars);
                }
                //接上前一次的剩余数据
                if(tail != null){
                    batchData = ""+tail+batchData;
                }
                //截取本次的剩余数据，留到下次用
                tail = batchData.substring(batchData.lastIndexOf(",")+1,batchData.length());
                if(tail.length()==0){
                    tail = null;
                }
                batchData = batchData.substring(0,batchData.lastIndexOf(","));
                String[] batchDataArr = new String[batchData.split(",").length];//多线程处理这个东西！！！
                batchDataArr = batchData.split(",");
                holder.push(batchDataArr);//主线程将每次读上来的数据保存到 
                
                if(isEnd==true){
                    break;
                }
            }
             //关闭线程用，这么关闭子线程对吗？？？
             while(true){
                 if(BatchDataArrHolder.isEmpty()){
                     pool.shutdown(); 
                     break;
                 }
             }
            //写数据
            Integer sortUnit = sortArray.getMaxNum();
            System.out.println(sortUnit);
            for(int i = 0;i<=sortUnit;i++){
                if(sortArray.getBit(i)==1){
                    StringBuffer buf = new StringBuffer(""+i+"");
                    buf.append(",");
                    Integer num = sortArray.repeatingData.get(i);
                    if(num!=null && num>=2){
                        for(int j=2;j<=num;j++){
                            buf.append(""+i+"").append(",");
                        }
                    }
                    out.write(buf.toString());
                }                
            }
            out.flush();
        }finally{
            if(in!=null){
                 try{
                     in.close();
                 }catch(IOException e){
                     e.printStackTrace();
                 }
            }
             if(out!=null){
                 try{
                     out.close();
                 }catch(IOException e){
                     e.printStackTrace();
                 }
            }
        }
    }
}