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.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.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.CompletableFuture;
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");
	}
}