Java7 ForkJoin入门实例

本文转自：http://www.xiaoyaochong.net/wordpress/?p=314

Java7引入了Fork Join的概念，来更好的支持并行运算。顾名思义，Fork Join类似与流程语言的分支，合并的概念。也就是说Java7 SE原生支持了在一个主线程中开辟多个分支线程，并且根据分支线程的逻辑来等待(或者不等待)汇集，当然你也可以fork的某一个分支线程中再开辟Fork Join，这也就可以实现Fork Join的嵌套。

有两个核心类ForkJoinPool和ForkJoinTask。

ForkJoinPool实现了ExecutorService接口，起到线程池的作用。所以他的用法和Executor框架的使用时一样的，当然Fork Join本身就是Executor框架的扩展。ForkJoinPool有3个关键的方法，来启动线程，execute(...),invoke(...),submit(...)。具体描述如下：

	客户端非fork/join调用	内部调用fork/join
异步执行	execute(ForkJoinTask)	ForkJoinTask.fork
等待获取结果	invoke(ForkJoinTask)	ForkJoinTask.invoke
执行，获取Futrue	submit(ForkJoinTask)	ForkJoinTask.fork(ForkJoinTasks are Futures)

 

 ForkJoinTask是分支合并的执行任何，分支合并的业务逻辑使用者可以再继承了这个抽先类之后，在抽象方法exec()中实现。其中exec()的返回结果和ForkJoinPool的执行调用方（execute(...),invoke(...),submit(...)），共同决定着线程是否阻塞，具体请看下面的测试用例。

首先，用户需要创建一个自己的ForkJoinTask。代码如下：

public class MyForkJoinTask<V> extends ForkJoinTask<V> {

    /**
     * 
     */
    private static final long serialVersionUID = 1L;
    
    private V value;
    
    private boolean success = false;
    
    @Override
    public V getRawResult() {
        return value;
    }

    @Override
    protected void setRawResult(V value) {
        this.value = value;
    }

    @Override
    protected boolean exec() {
        System.out.println("exec");
        return this.success;
    }

    public boolean isSuccess() {
        return success;
    }

    public void setSuccess(boolean isSuccess) {
        this.success = isSuccess;
    }

}

 

测试ForkJoinPool.invoke(...)：

@Test
    public void testForkJoinInvoke() throws InterruptedException, ExecutionException {
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        task.setSuccess(true);
        task.setRawResult("test");
        String invokeResult = forkJoinPool.invoke(task);
        assertEquals(invokeResult, "test");
    }
    
    @Test
    public void testForkJoinInvoke2() throws InterruptedException, ExecutionException {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        final MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        new Thread(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                
                task.complete("test");
            }
        }).start();
        
        // exec()返回值是false，此处阻塞，直到另一个线程调用了task.complete(...)
        String result = forkJoinPool.invoke(task);
        System.out.println(result);
    }
    
    @Test
    public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        final MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        
        task.setSuccess(true); // 是否在此任务运行完毕后结束阻塞
        ForkJoinTask<String> result = forkJoinPool.submit(task);
        result.get(); // 如果exec()返回值是false，在此处会阻塞，直到调用complete
    }

 

测试ForkJoinPool.submit(...)：

@Test
    public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        final MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        
        task.setSuccess(true); // 是否在此任务运行完毕后结束阻塞
        ForkJoinTask<String> result = forkJoinPool.submit(task);
        result.get(); // 如果exec()返回值是false，在此处会阻塞，直到调用complete
    }
    
    @Test
    public void testForkJoinSubmit2() throws InterruptedException, ExecutionException {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        final MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        
        forkJoinPool.submit(task);
        Thread.sleep(1000);
    }
    
    @Test
    public void testForkJoinSubmit3() throws InterruptedException, ExecutionException {
        final ForkJoinPool forkJoinPool = new ForkJoinPool();
        final MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        new Thread(new Runnable() {
            public void run() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
                
                task.complete("test");
            }
        }).start();
        
        ForkJoinTask<String> result = forkJoinPool.submit(task);
        // exec()返回值是false，此处阻塞，直到另一个线程调用了task.complete(...)
        result.get(); 
        Thread.sleep(1000);
    }

 

测试ForkJoinPool.execute(...)：

 @Test
    public void testForkJoinExecute() throws InterruptedException, ExecutionException {
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        MyForkJoinTask<String> task = new MyForkJoinTask<String>();
        forkJoinPool.execute(task); // 异步执行，无视task.exec()返回值。
    }