您好,登錄后才能下訂單哦!
這篇文章給大家介紹java中如何實(shí)現(xiàn)異步編程,內(nèi)容非常詳細(xì),感興趣的小伙伴們可以參考借鑒,希望對大家能有所幫助。
很多時(shí)候我們都希望能夠最大的利用資源,比如在進(jìn)行IO操作的時(shí)候盡可能的避免同步阻塞的等待,因?yàn)檫@會(huì)浪費(fèi)CPU的資源。如果在有可讀的數(shù)據(jù)的時(shí)候能夠通知程序執(zhí)行讀操作甚至由操作系統(tǒng)內(nèi)核幫助我們完成數(shù)據(jù)的拷貝,這再好不過了。從NIO到CompletableFuture、Lambda、Fork/Join,java一直在努力讓程序盡可能變的異步甚至擁有更高的并行度,這一點(diǎn)一些函數(shù)式語言做的比較好,因此java也或多或少的借鑒了某些特性。下面介紹一種非常常用的實(shí)現(xiàn)異步操作的方式。
考慮有一個(gè)耗時(shí)的操作,操作完后會(huì)返回一個(gè)結(jié)果(不管是正常結(jié)果還是異常),程序如果想擁有比較好的性能不可能由線程去等待操作的完成,而是應(yīng)該采用listener模式。jdk并發(fā)包里的Future代表了未來的某個(gè)結(jié)果,當(dāng)我們向線程池中提交任務(wù)的時(shí)候會(huì)返回該對象。代碼例子:
/** * jdk1.8之前的Future * * @author Administrator * */ public class JavaFuture { public static void main(String[] args) throws Throwable, ExecutionException { ExecutorService executor = Executors.newFixedThreadPool(1); // Future代表了線程執(zhí)行完以后的結(jié)果,可以通過future獲得執(zhí)行的結(jié)果 // 但是jdk1.8之前的Future有點(diǎn)雞肋,并不能實(shí)現(xiàn)真正的異步,需要阻塞的獲取結(jié)果,或者不斷的輪詢 // 通常我們希望當(dāng)線程執(zhí)行完一些耗時(shí)的任務(wù)后,能夠自動(dòng)的通知我們結(jié)果,很遺憾這在原生jdk1.8之前 // 是不支持的,但是我們可以通過第三方的庫實(shí)現(xiàn)真正的異步回調(diào) 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"); } }
如果想獲得耗時(shí)操作的結(jié)果,可以通過get方法獲取,但是該方法會(huì)阻塞當(dāng)前線程,我們可以在做完剩下的某些工作的時(shí)候調(diào)用get方法試圖去獲取結(jié)果,也可以調(diào)用非阻塞的方法isDone來確定操作是否完成,這種方式有點(diǎn)兒類似下面的過程:
這種方式對流程的控制很混亂,但是在jdk1.8之前只提供了這種笨拙的實(shí)現(xiàn)方式,以至于很多高性能的框架都實(shí)現(xiàn)了自己的一套異步框架,比如Netty和Guava,下面分別介紹下這三種異步的實(shí)現(xiàn)方式(包括jdk1.8)。首先是Guava中的實(shí)現(xiàn)方式:
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); //向線程池中提交一個(gè)任務(wù)后,將會(huì)返回一個(gè)可監(jiān)聽的Future,該Future由Guava框架提供 ListenableFuture<String> lf = listeningExecutor.submit(new Callable<String>() { @Override public String call() throws Exception { System.out.println("task started!"); //模擬耗時(shí)操作 Thread.sleep(3000); System.out.println("task finished!"); return "hello"; } }); //添加回調(diào),回調(diào)由executor中的線程觸發(fā),但也可以指定一個(gè)新的線程 Futures.addCallback(lf, new FutureCallback<String>() { //耗時(shí)任務(wù)執(zhí)行失敗后回調(diào)該方法 @Override public void onFailure(Throwable t) { System.out.println("failure"); } //耗時(shí)任務(wù)執(zhí)行成功后回調(diào)該方法 @Override public void onSuccess(String s) { System.out.println("success " + s); } }); //主線程可以繼續(xù)做其他的工作 System.out.println("main thread is running"); } }
Guava提供了一套完整的異步框架,核心是可監(jiān)聽的Future,通過注冊監(jiān)聽器或者回調(diào)方法實(shí)現(xiàn)及時(shí)獲取操作結(jié)果的能力。需要提一點(diǎn)的是,假設(shè)添加監(jiān)聽的時(shí)候耗時(shí)操作已經(jīng)執(zhí)行完了,此時(shí)回調(diào)方法會(huì)被立即執(zhí)行并不會(huì)丟失。想探究其實(shí)現(xiàn)方式的話可以跟一下源碼,底層的原理并不難。
談到異步編程就不得不提一下Promise,很多函數(shù)式語言比如js原生支持Promise,但是在java界也有一些promise框架,其中就有大名鼎鼎的Netty。從Future、Callback到Promise甚至線程池,Netty實(shí)現(xiàn)了一套完整的異步框架,并且netty代碼中也大量使用了Promise,下面是Netty中的例子:
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); //向線程池中提交任務(wù),并返回Future,該Future是netty自己實(shí)現(xiàn)的future //位于io.netty.util.concurrent包下,此處運(yùn)行時(shí)的類型為PromiseTask Future<?> f = group.submit(new Runnable() { @Override public void run() { System.out.println("任務(wù)正在執(zhí)行"); //模擬耗時(shí)操作,比如IO操作 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("任務(wù)執(zhí)行完畢"); } }); //增加監(jiān)聽 f.addListener( new FutureListener() { @Override public void operationComplete(Future arg0) throws Exception { System.out.println("ok!!!"); } }); System.out.println("main thread is running."); } }
直到j(luò)dk1.8才算真正支持了異步操作,其中借鑒了某些框架的實(shí)現(xiàn)思想,但又有新的功能,同時(shí)在jdk1.8中提供了lambda表達(dá)式,使得java向函數(shù)式語言又靠近了一步。借助jdk原生的CompletableFuture可以實(shí)現(xiàn)異步的操作,同時(shí)結(jié)合lambada表達(dá)式大大簡化了代碼量。代碼例子如下:
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實(shí)現(xiàn)任務(wù)異步處理 * * @author Administrator * */ public class JavaPromise { public static void main(String[] args) throws Throwable, ExecutionException { // 兩個(gè)線程的線程池 ExecutorService executor = Executors.newFixedThreadPool(2); //jdk1.8之前的實(shí)現(xiàn)方式 CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() { @Override public String get() { System.out.println("task started!"); try { //模擬耗時(shí)操作 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "task finished!"; } }, executor); //采用lambada的實(shí)現(xiàn)方式 future.thenAccept(e -> System.out.println(e + " ok")); System.out.println("main thread is running"); } }
關(guān)于java中如何實(shí)現(xiàn)異步編程就分享到這里了,希望以上內(nèi)容可以對大家有一定的幫助,可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò),可以把它分享出去讓更多的人看到。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。