java异步编程详解

/**

  * jdk1.8之前的future

  *

  * @author administrator

  *

  */

public class javafuture {

     public static void main(string[] args) throws throwable, executionexception {

         executorservice executor = executors.newfixedthreadpool( 1 );

         // future代表了线程执行完以后的结果，可以通过future获得执行的结果

         // 但是jdk1.8之前的future有点鸡肋，并不能实现真正的异步，需要阻塞的获取结果，或者不断的轮询

         // 通常我们希望当线程执行完一些耗时的任务后，能够自动的通知我们结果，很遗憾这在原生jdk1.8之前

         // 是不支持的，但是我们可以通过第三方的库实现真正的异步回调

         future<string> f = executor.submit( new callable<string>() {

             @override

             public string call() throws exception {

                 system.out.println( "task started!" );

                 thread.sleep( 3000 );

                 system.out.println( "task finished!" );

                 return "hello" ;

             }

         });

         //此处阻塞main线程

         system.out.println(f.get());

         system.out.println( "main thread is blocked" );

     }

}

package guava;

import java.util.concurrent.callable;

import java.util.concurrent.executorservice;

import java.util.concurrent.executors;

import com.google测试数据mon.util.concurrent.futurecallback;

import com.google测试数据mon.util.concurrent.futures;

import com.google测试数据mon.util.concurrent.listenablefuture;

import com.google测试数据mon.util.concurrent.listeningexecutorservice;

import com.google测试数据mon.util.concurrent.moreexecutors;

/**

  * guava中的future

  *

  * @author administrator

  *

  */

public class guavafuture {

     public static void main(string[] args) {

         executorservice executor = executors.newfixedthreadpool( 1 );

         // 使用guava提供的moreexecutors工具类包装原始的线程池

         listeningexecutorservice listeningexecutor = moreexecutors.listeningdecorator(executor);

         //向线程池中提交一个任务后，将会返回一个可监听的future，该future由guava框架提供

         listenablefuture<string> lf = listeningexecutor.submit( new callable<string>() {

             @override

             public string call() throws exception {

                 system.out.println( "task started!" );

                 //模拟耗时操作

                 thread.sleep( 3000 );

                 system.out.println( "task finished!" );

                 return "hello" ;

             }

         });

         //添加回调，回调由executor中的线程触发，但也可以指定一个新的线程

         futures.addcallback(lf, new futurecallback<string>() {

             //耗时任务执行失败后回调该方法

             @override

             public void onfailure(throwable t) {

                 system.out.println( "failure" );

             }

             //耗时任务执行成功后回调该方法

             @override

             public void onsuccess(string s) {

                 system.out.println( "success " + s);

             }

         });

         //主线程可以继续做其他的工作

         system.out.println( "main thread is running" );

     }

}

package netty_promise;

import io.netty.util.concurrent.defaulteventexecutorgroup;

import io.netty.util.concurrent.eventexecutorgroup;

import io.netty.util.concurrent.future;

import io.netty.util.concurrent.futurelistener;

/**

  * netty中的promise

  *

  * @author administrator

  *

  */

public class promisetest {

     @suppresswarnings ({ "unchecked" , "rawtypes" })

     public static void main(string[] args) throws throwable {

         //线程池

         eventexecutorgroup group = new defaulteventexecutorgroup( 1 );

         //向线程池中提交任务，并返回future，该future是netty自己实现的future

         //位于io.netty.util.concurrent包下，此处运行时的类型为promisetask

         future<?> f = group.submit( new runnable() {

             @override

             public void run() {

                 system.out.println( "任务正在执行" );

                 //模拟耗时操作，比如io操作

                 try {

                     thread.sleep( 1000 );

                 } catch (interruptedexception e) {

                     e.printstacktrace();

                 }

                 system.out.println( "任务执行完毕" );

             }

         });

         //增加监听

         f.addlistener( new futurelistener() {

             @override

             public void operationcomplete(future arg0) throws exception {

                 system.out.println( "ok!!!" );

             }

         });

         system.out.println( "main thread is running." );

     }

}

package netty_promise;

import java.util.concurrent测试数据pletablefuture;

import java.util.concurrent.executionexception;

import java.util.concurrent.executorservice;

import java.util.concurrent.executors;

import java.util.function.supplier;

/**

  * 基于jdk1.8实现任务异步处理

  *

  * @author administrator

  *

  */

public class javapromise {

     public static void main(string[] args) throws throwable, executionexception {

         // 两个线程的线程池

         executorservice executor = executors.newfixedthreadpool( 2 );

         //jdk1.8之前的实现方式

         completablefuture<string> future = completablefuture.supplyasync( new supplier<string>() {

             @override

             public string get() {

                 system.out.println( "task started!" );

                 try {

                     //模拟耗时操作

                     thread.sleep( 2000 );

                 } catch (interruptedexception e) {

                     e.printstacktrace();

                 }

                 return "task finished!" ;

             }

         }, executor);

         //采用lambada的实现方式

         future.thenaccept(e -> system.out.println(e + " ok" ));

         system.out.println( "main thread is running" );

     }

}