汉辰攻略

The palest ink is better than the best memory.

  BlogJava :: 首页 :: 联系 :: 聚合  :: 管理
  8 Posts :: 1 Stories :: 3 Comments :: 0 Trackbacks
Fork-join framework是Java 7并行库的新内容,基于divide-and-conquer算法来处理大数据量计算。DnQ是处理微粒度并行计算(数据量大,单位运行时间短)的理想模式。数据量在达到一个预定义的门槛之前,被分割成多个任务被Worker threads执行。因为现代Java虚拟机都把Java thread映射到系统线程或LWP(Light-weight process) ,同时Worker数量一般设定等同于CPU个数,这样在多核的硬件系统中能充分利用多个CPU的计算能力。

写了一个MergeSort的测试例子,最终的排序用的是Java Collection Framework 自带的Arrays.sort()。在自己双核机器试了试,发现提升不是特别明显。Arrays.sort 本身很高效,Framework有thread之间协作和管理worker pool的开销,所以必须选择一个适合的数据量阚值。下面是运行结果:

java -Xms64m -Xmx128m -cp C;/forkjoin/jsr166y.zip;C:/workspace/java.tij forkjoin.SortTask

Number of processor 2
=================Sequential ===================
Sorting takes 2617701971 to complete
=================ForkJoin ====================
Sorting takes 2284940405 to complete

找不到更多核的机器,有条件的同学可以测试一把。另外,Brain Goetz (Java Concurrency in Practice作者) 的文章可参考,他的测试例子显示了不错的性能提升(最高17倍在32cpu系统),一般4核或8核的能达到3倍或5倍的SPEEDUP

Java thread and practice: Stick a fork in it Part 1 - http://www.ibm.com/developerworks/java/library/j-jtp11137.html

package forkjoin;

import jsr166y.forkjoin.RecursiveAction;
import jsr166y.forkjoin.ForkJoinPool;
import java.util.Random;
import java.util.Arrays;

public class SortTask extends RecursiveAction {

    
final static int ARRAY_LENGTH = 10000000;

    
final static int THRESHOLD = 3000000;

    
final int[] array;

    
final int lo;

    
final int hi;

    
public SortTask(int[] array, int lo, int hi) {
        
this.array = array;
        
this.lo = lo;
        
this.hi = hi;
    }


    
private void sequentiallySort(int[] array, int lo, int hi) {
        
int[] units = new int[hi - lo + 1];
        
for (int i = lo; i <= hi; i++)
            units[i 
- lo] = array[i];
        Arrays.sort(units);
        
for (int i = lo; i <= hi; i++)
            array[i] 
= units[i - lo];
    }


    
private void merge(int[] array, int lo, int mid, int hi) {

        
int[] units = new int[hi - lo + 1];
        
int i = lo;
        
int j = mid + 1;

        
for (int k = 0; k < units.length; k++{
            
if (array[i] <= array[j])
                units[k] 
= array[i++];
            
else if (array[i] > array[j])
                units[k] 
= array[j++];

            
if (i > mid)
                
for (int m = j; m <= hi; m++)
                    units[
++k] = array[m];
            
else if (j > hi)
                
for (int m = i; m <= mid; m++)
                    units[
++k] = array[m];
        }


        
for (int k = lo; k <= hi; k++)
            array[k] 
= units[k - lo];

    }


    
protected void compute() {
        
try {
            
if (hi - lo < THRESHOLD)
                sequentiallySort(array, lo, hi);
            
else {
                
int mid = (lo + hi) >>> 1;
                
//System.out.println(mid);
                forkJoin(new SortTask(array, lo, mid), new SortTask(array, mid + 1, hi));

                merge(array, lo, mid, hi);
            }

        }
 catch (Throwable t) {
            t.printStackTrace();
        }

    }


    
/**
     * 
@param args
     
*/

    
public static void main(String[] args) {
        
int[] sample = new int[ARRAY_LENGTH];

        System.out.println(
"Number of processor"
                
+ Runtime.getRuntime().availableProcessors());
        
        Random seed 
= new Random(47);

        
for (int i = 0; i < sample.length; i++{
            sample[i] 
= seed.nextInt();
        }


        
long start = System.nanoTime();
        Arrays.sort(sample);
        
long duration = System.nanoTime() - start;


        System.out.println(
"===============Sequential==================");
        System.out.println(
"Sorting takes " + duration + " to compelte");

        
int[] sample2 = new int[ARRAY_LENGTH];

        
for (int i = 0; i < sample2.length; i++{
            sample2[i] 
= seed.nextInt();
        }


        ForkJoinPool pool 
= new ForkJoinPool(Runtime.getRuntime()
                .availableProcessors());
        SortTask st 
= new SortTask(sample2, 0, sample2.length - 1);

        start 
= System.nanoTime();
        pool.execute(st);
        
while (!st.isDone()) {
        }

        duration 
= System.nanoTime() - start;

        System.out.println(
"===============ForkJoin==================");
        System.out.println(
"Sorting takes " + duration + " to compelte");
        
    }


}


posted on 2008-06-26 10:11 汉辰 阅读(1314) 评论(2)  编辑  收藏

Feedback

# re: Fork-join Framework 2009-03-30 17:03 f
请问这个方法在什么地方
forkJoin();
谢谢.  回复  更多评论
  

# re: Fork-join Framework 2009-03-31 09:52 汉辰
@f
是SortTask所继承的上层抽象类中的一个方法  回复  更多评论
  


只有注册用户登录后才能发表评论。


网站导航: