多线程文本文件排序

网上看到一道题。https://github.com/Skinney/WordSorter
简单的描述就是，对一个已知的文本文件按行字符串自然序进行排序，结果输出到另一个文件。

设计了一系列算法。


先搞一个最简单的。
Allen01_SimpleSort
1 读取数据。
2 排序。
3 写回。
4 单线程完成所有任务。



数据都是字符串，可以使用桶排先预排序一下。
Allen02_String_Bucket
桶排。按照每一行字符串的前2个字符分为26*26个桶。
数据读取和数据入桶线程分开。
写文件线程必须等待数据入桶线程完成任务。并且对桶内数据进行排序。
面向字符串处理。



字符串处理太慢，改为字节处理。
Allen04_Bytes_Bucket
桶排。
数据读取和数据入桶线程分开。
写文件线程必须等待数据入桶线程完成任务。并且对桶内数据进行排序。
面向字节处理。

数据先按照数据块入一个中间queue，入桶线程解析数据块为多个字节数组，入桶。


桶内的数据排序过程可以并行化。
Allen05_Fixed_Sorter
桶排。
数据读取和数据入桶线程分开。
一组桶内排序线程按照桶的数量对桶内数据排序工作做固定划分。
写文件线程必须等待排序线程完成任务。
面向字节处理。


线程桶内数据排序的划分，由于是固定的，有可能导致线程的工作负载不同。
Allen06_Dynamic_Sorter
桶排。
数据读取和数据入桶线程分开。
一组桶内排序线程按照本身排序工作的进度竞争得到下一个需要排序的桶。
写文件线程必须等待排序线程完成任务。
面向字节处理。

这个方案还搞了一个变种，使用26*26*27的桶，效果不太好。


一个桶内数据排序完成，数据就可以写出了，单独有一个线程监控桶内数据排序情况，排好序了就尽量写出。
Allen08_Concurrent_Writer
桶排。
数据读取和数据入桶线程分开。
一组桶内排序线程按照本身排序工作的进度竞争得到下一个需要排序的桶。
写文件线程和排序线程并发运行。
面向字节处理。
写文件线程和排序线程通过queue通信。


原始的桶内数据使用java自带的ArrayList保存，感觉有点重量级。
自己实现了一个简单的list，来保持桶内数据。自行管理数组容量，动态扩展。

/**
 * 简单列表，代替ArrayList。
 * */
public class AllenSimpleList implements AllenList {

    private int      capacity = 16;
    private byte[][] datas    = new byte[capacity][];
    private int      next     = 0;

    public void addData(byte[] data) {
        if (next != capacity) {
            datas[next++] = data;
        } else {
            int new_capacity = capacity << 1;
            byte[][] new_datas = new byte[new_capacity][];
            System.arraycopy(datas, 0, new_datas, 0, capacity);
            new_datas[next++] = data;
            capacity = new_capacity;
            datas = new_datas;
        }
    }

    public int getLength() {
        return next;
    }

    public void mergeOtherList(AllenSimpleList other) {
        if (capacity < next + other.next) {
            int new_capacity = next + other.next;
            byte[][] new_datas = new byte[new_capacity][];
            System.arraycopy(datas, 0, new_datas, 0, next);
            System.arraycopy(other.datas, 0, new_datas, next, other.next);
            capacity = new_capacity;
            datas = new_datas;
            next = capacity;
        } else {
            System.arraycopy(other.datas, 0, datas, next, other.next);
            next += other.next;
        }
    }

    public void sort() {
        AllenSorter.sort(datas, 0, next);
    }

    public void write(OutputStream out) throws Exception {
        for (int i = 0; i < next; i++) {
            out.write(datas[i]);
        }
    }

    @Override
    public void addData(int threadIndex, byte[] data) {
        addData(data);
    }

同时，原有排序线程和写出线程使用queue通信，可以配置为忙查询。
Allen08_Concurrent_Writer_Need_Strategy
使用了AllenSimpleList。
需要一个Strategy来控制程序。Strategy可以控制
是否设置线程的优先级。
使用何种策略进行排序线程和写文件线程的通信,queue还是轮询.
写文件线程是否会Yield。



数据入桶也可以并行化。
有可能导致桶内数据并发修改时不一致。
第1个方案，每一个线程一个单独的数据桶。

/**
 * 多个simpleList，按照线程号分流。
 * */
public class AllenSafeSimpleList implements AllenList {

    private AllenSimpleList[] list;

    public AllenSafeSimpleList(int threadCount) {

        list = new AllenSimpleList[threadCount];
        for (int i = 0; i < list.length; i++) {
            list[i] = new AllenSimpleList();
        }
    }

    public void addData(int threadIndex, byte[] data) {
        list[threadIndex].addData(data);
    }

    @Override
    public void sort() {
        for (int i = 1; i < list.length; i++) {
            if (list[i].getLength() > 0) {
                list[0].mergeOtherList(list[i]);
            }
        }
        list[0].sort();
    }

    @Override
    public void write(OutputStream out) throws Exception {
        list[0].write(out);
    }
}

第2个方案，主备方案。

/**
 * 2个simpleList，主备分流。
 * */
public class AllenSafeSimpleList2 implements AllenList {

    private AllenSimpleList mainList;
    private AllenSimpleList bakList;

    private AtomicBoolean   main = new AtomicBoolean(false);
    private AtomicBoolean   bak  = new AtomicBoolean(false);

    public AllenSafeSimpleList2() {
        mainList = new AllenSimpleList();
        bakList = new AllenSimpleList();
    }

    public void addData(int threadIndex, byte[] data) {
        if (main.compareAndSet(false, true)) {
            mainList.addData(data);
            main.set(false);
            return;
        } else {
            while (bak.compareAndSet(false, true)) {
                bakList.addData(data);
                bak.set(false);
                return;
            }
        }
    }

    @Override
    public void sort() {
        if (bakList.getLength() != 0) {
            mainList.mergeOtherList(bakList);
        }
        mainList.sort();
    }

    @Override
    public void write(OutputStream out) throws Exception {
        mainList.write(out);
    }
}

Allen12_Concurrent_DataWorker_Need_Strategy

桶排。
数据读取和数据入桶线程分开。
一组桶内排序线程按照本身排序工作的进度竞争得到下一个需要排序的桶。
写文件线程和排序线程并发运行。
面向字节处理。
入桶线程变成多线程。

需要一个Strategy来控制程序。Strategy可以控制
是否设置线程的优先级。
使用何种策略进行排序线程和写文件线程的通信,queue还是轮询.
写文件线程是否会Yield。
使用AllenSafeSimpleList还是AllenSafeSimpleList2。


bug fix:

AllenSafeSimpleList2

    @Override
    public void addData(int threadIndex, byte[] data) {
        if (main.compareAndSet(false, true)) {
            mainList.addData(data);
            main.set(false);
            return;
        } else {
            while (true) {
                if (bak.compareAndSet(false, true)) {
                    bakList.addData(data);
                    bak.set(false);
                    return;
                }
            }
        }
    }

评论

5 楼 zhang_xzhi_xjtu 2012-09-26  

大家有建议可以交流啊。

4 楼 zhongliangjun1 2012-09-26  

jss 写道

lvwenwen 写道

zjhlht 写道

这个比较有用，研究一下

这个比较有用，研究一下

这个比较有用，研究一下

我是来破坏队形的

3 楼 jss 2012-09-26  

lvwenwen 写道

zjhlht 写道

这个比较有用，研究一下

这个比较有用，研究一下

这个比较有用，研究一下

2 楼 lvwenwen 2012-09-26  

zjhlht 写道

这个比较有用，研究一下

这个比较有用，研究一下

1 楼 zjhlht 2012-09-26  

这个比较有用，研究一下