dictsort.java

用java实现的多线程字典排序

import java.io.*;
import java.io.Reader.*;
//import java.util.Arrays;

public class DictSort {
	private final static int THREADNUM = 4;
	private final static String TMPPREFIX = "hhtmp";
	private short flag = 0;
	private final static short THREADFINISHED = 0X0F;

	private String fileOpen = null;
	
	public DictSort(String strDict) throws Exception
	{
		File f = new File(strDict);
		if(!f.isFile() || f.length() < 1024)
		{
			throw new Exception(strDict + " is not a valid file" 
					+ "or file length < 1024bytes");
		}
		fileOpen = strDict;
	}
	
	/**
	 * Sort to dest file.
	 * @param fileSave: dest file path
	 */
	public void sortTo(String fileSave)throws IOException
	{
		try
		{
			generateBlocks();
		}
		catch(Exception e)
		{
			System.out.println("Error occurs in generate Blocks.");
			System.out.println(e);
		}
	
		while(!isFinished())
		{
			//if not finished, wait 0.5s
			try
			{
				java.lang.Thread.sleep(500);
			}
			catch(InterruptedException e)
			{
				System.out.println("Thread sleep fucn error.");
				System.out.println(e);
			}
		}
		//if all blocks are prepared,merge them
		mergeBlocks(fileSave);
		//clean up
		cleanTmpFile();

	}
	
	/**
	 * Merge
	 */
	protected void mergeBlocks(String fileSave)throws IOException
	{
		int i;
		String buff[] = new String[THREADNUM];
		BufferedReader br[] = new BufferedReader[THREADNUM];
		for(i =0;i < THREADNUM; i++)
		{
			br[i] = new BufferedReader(new FileReader(TMPPREFIX + (i+1)));
			buff[i] = br[i].readLine(); 
		}
		
		BufferedWriter bw = new BufferedWriter(new FileWriter(fileSave));
		int who = 0;//who should be read next
		int blockReadFinished = 0;
		for(;;)
		{
			if(blockReadFinished == THREADFINISHED)
				break;
			for(i = 0;
				i < THREADNUM;// (blockReadFinished & (1<<i)) == 0;
				i++
				)
			{
				if((blockReadFinished & (1<<i)) == 0)
				{
					who = i;
					break;
				}
			}

			for(i = who + 1; i < THREADNUM; i++)
			{
				if(
						(blockReadFinished & (1<<i)) == 0
						&& buff[i].compareTo(buff[who]) < 0 
				   )
					who = i;
			}
			
			bw.write(buff[who]);
			bw.newLine();
			buff[who] = br[who].readLine();
			if(buff[who] == null)
			{
				blockReadFinished |= (1<<who);
				br[who].close();
			}
		}
		bw.close();
	}
	/**
	 * Using threads to generate THREADNUM sorted blocks
	 */
	protected void generateBlocks()throws Exception
	{
		BufferedReader rd = new BufferedReader(new FileReader(fileOpen));
		
		//first, get the dictionary records number
		int lines;
		try
		{
			lines = Integer.parseInt(rd.readLine());
		}
		catch(Exception e)
		{
			throw new Exception("Dictionary number is invalid.");
		}
		if(lines < THREADNUM)
		{
			throw new Exception("Dictionary number is invalid.");
		}
		
		//decide how many records that each thread should deal with
		int howMany = lines / THREADNUM;
		if(lines % THREADNUM != 0) howMany++;
		
		for(int cnt =0, sum = lines, i=1;;cnt = 0, i++)
		{	
			String[] buff = null;
			if(sum < howMany)
				buff = new String[sum];
			else
				buff = new String[howMany];
			while(cnt < howMany)
			{
				buff[cnt++] = rd.readLine(); 
			}
			//sort each block and save to tmp files
			new SortThread(buff, i).start();
				
			sum -= howMany;
			if(sum <= 0) break;
		}
	}
	
	/**
	 * all threads are finished?
	 */
	protected boolean isFinished()
	{
		return (flag == THREADFINISHED);
	}
	/**
	 * clean the generated temp files...
	 */
	protected void cleanTmpFile()
	{
		for(int i = 1; i <= THREADNUM; i++)
		{
			File f = new File(TMPPREFIX + i);
			if(f.exists())
			{
				f.delete();
			}
		}
	}
	
		
	/**
	 * Thread sort innner class
	 */
	class SortThread extends Thread
	{
		String[] array = null;
		//i denotes block no
		int i;
		
		
		SortThread(String[] strArray, int blockNo)
		{
			array = strArray;
			i = blockNo;
		}
		public void run()
		{
			java.util.Arrays.sort(array);
			BufferedWriter bw;
			try
			{
				bw = new BufferedWriter(new FileWriter(TMPPREFIX+i));
				for(int j=0; j < array.length; j++)
				{
					bw.write(array[j]);
					bw.newLine();
				}
				bw.close();
			}
			catch(IOException e)
			{
				System.out.println("Write file " + TMPPREFIX + i + " Error.");
				System.out.println(e);
			}
			//set thread finish flag
			flag |= (1 << (i-1));
		}
	}
}