利用JAVA多线程来提高数据处理效率

Posted 2020-10-12 风轻云淡

　　肿瘤大数据挖掘中经常需要处理上百亿行的文本文件，这些文件往往高达数百GB，假如文件结构简单统一，那么用sed和awk 处理是非常方便和快速的。但有时候会遇到逻辑较为复杂的处理流程，这样我一般会用JAVA来处理。但由于JAVA是单线程的，因此对于实验室多核服务器来说，能充分有效的利用起每个核会方便不少，那么这个时候就推荐用多线程来并发（并行）处理任务，从而达到运算速度倍速的提升。

　　这里举一个并行计算的例子。例子比较简单，主要是对三个数进行累加，最后输出结果。我们分别用单线程和多线程来执行，其中单线程是顺序执行而多线程则同时启动三个线程来并行（服务器CPU数大于三，所以这里是并行而不是并发）执行。

 

　　首先是单线程的运行结果：

public class Nothreading
{
    public static void main(String[] args)
    {    
        long startTime = System.currentTimeMillis();
        int sum_i = 0;
        int sum_j = 0;
        int sum_k = 0;
        for(int i = 0; i < 10000; i++)
        {    
            sum_i += 1;
　　　　　　　/* 增加程序运行时间， 后面同理 */
            for(int a = 0 ; a < 100000 ; a ++)
            {
                String s = "To cost some time";
                String[] ss = s.split(" ");
            }
        }
        for(int j = 0; j < 10000; j++)
        {    
            sum_j += 2;
            for(int a = 0 ; a < 100000 ; a ++)
            {
                String s = "To cost some time";
                String[] ss = s.split(" ");
            }
        }
        for(int k = 0; k < 10000; k++)
        {    
            sum_k += 3;
            for(int a = 0 ; a < 100000 ; a ++)
            {
                String s = "To cost some time";
                String[] ss = s.split(" ");
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println(sum_i + "\\t" + sum_j + "\\t" + sum_k);
        System.out.println("run time:"+(endTime-startTime)+"ms");
    }
}

　　运行结果：

10000    20000    30000
run time:663587ms

　　图片是该程序运行时的CPU资源利用状态： 可以看到仅有一个CPU的利用率达到100%.

 

　　下面是多线程：

class Count_i
{    
    public int sum_i = 0;
    public synchronized void count()
    {
        for(int i = 0 ; i < 10000; i++)
        {
            sum_i += 1;
            /* 增加运行时间 后面同理*/
            for(int a = 0 ; a < 100000; a ++)
            {
                String s = "To cost some time";
                String[] ss = s.split(" ");
            }
        }
    }
}

class Count_j
{    
    public int sum_j = 0;
    public synchronized void count()
    {    
        for(int j = 0 ; j < 10000; j++)
        {
            sum_j += 2;
            for(int a = 0 ; a < 100000; a ++)
            {
                String s = "To cost some time";
                String[] ss = s.split(" ");
            }
        }
    }
}

class Count_k
{    
    public int sum_k = 0;
    public synchronized void count()
    {
        for(int k = 0 ; k < 10000; k++)
        {
            sum_k += 3;
            for(int a = 0 ; a < 100000; a ++)
            {
                String s = "To cost some time";
                String[] ss = s.split(" ");
            }
        }
    }
}

class Mul_thread_i extends Thread
{
    public Count_i c_i;
    public Mul_thread_i(Count_i acc)
    {
        this.c_i = acc;
    }
    public void run()
    {
        c_i.count();
    }
}

class Mul_thread_j extends Thread
{
    public Count_j c_j;
    public Mul_thread_j(Count_j acc)
    {
        this.c_j = acc;
    }
    public void run()
    {
        c_j.count();
    }
}

class Mul_thread_k extends Thread
{
    public Count_k c_k;
    public Mul_thread_k(Count_k acc)
    {
        this.c_k = acc;
    }
    public void run()
    {
        c_k.count();
    }
}


public class Threethreading_save
{    
    public static void main(String[] args) throws InterruptedException 
    {    
        long startTime = System.currentTimeMillis();
        Count_i ci = new Count_i();
        Count_j cj = new Count_j();
        Count_k ck = new Count_k();
        Mul_thread_i aa = new Mul_thread_i(ci);
        Mul_thread_j bb = new Mul_thread_j(cj);
        Mul_thread_k cc = new Mul_thread_k(ck);

        aa.start();
        bb.start();
        cc.start();
        aa.join();
        bb.join();
        cc.join();
        
    
        System.out.println(ci.sum_i);
        System.out.println(cj.sum_j);
        System.out.println(ck.sum_k);
        long endTime = System.currentTimeMillis();
        System.out.println("run time:"+(endTime-startTime)+"ms");
    }
}

　　下面是运行结果：

10000
20000
30000
run time:221227ms

　　CPU状态：可以看到有三个CPU的利用率达到100%.

 

 

空闲时的状态：

 

 

　　总结一些，当我们处理的任务量很大的时候，如果计算机有多个CPU，可以将待处理的任务合理的分为几个部分，然后开几个线程同时进行运算，等这些子任务都完成以后再交给主线程后续的处理,可以看到效率成倍的提升。当然线程安全是一个需要注意的问题，由于时间关系后面将详细介绍。