碼上歡樂
相關內容    簡體    繁體

Netty中的Future

本文轉載自   查看原文   2015-04-08 10:55   2859    編程語言

先看下Future的整個繼承體系，還有一個ChannelFuture不在里面；

    在並發編程中，我們通常會用到一組非阻塞的模型：Promise，Future 和 Callback。其中的 Future 表示一個可能還沒有實際完成的異步任務的結果，針對這個結果可以添加 Callback 以便在任務執行成功或失敗后做出對應的操作，而 Promise 交由任務執行者，任務執行者通過 Promise 可以標記任務完成或者失敗。 可以說這一套模型是很多異步非阻塞架構的基礎。
    這一套經典的模型在 Scala、C# 中得到了原生的支持，但 JDK 中暫時還只有無 Callback 的 Future 出現，當然也並非在 JAVA 界就沒有發展了，比如 Guava 就提供了ListenableFuture 接口，而 Netty 4+ 更是提供了完整的 Promise、Future 和 Listener 機制，在 Netty 的官方文檔 Using as a generic library 中也介紹了將 Netty 作為一個 lib 包依賴，並且使用 Listenable futures 的示例。在實際的項目使用中，發現 Netty 的 EventLoop 機制不一定適用其他場景，因此想去除對 EventLoop 的依賴，實現一個簡化版本。
    Netty自己實現了一套並發庫，Future是其中的一塊，下篇文章講下他的並發庫的線程池實現。Netty的Future的特性
  • Future<V>的V為異步結果的返回類型
  • getNow 是無阻塞調用，返回異步執行結果，如果未完成那么返回null
  • await 是阻塞調用,等到異步執行完成
  • isSuccess 執行成功是否成功
  • sync  阻塞調用，等待這個future直到isDone（可能由於正常終止、異常或取消而完成）返回true; 如果該future失敗，重新拋出失敗的原因。 和await區別就是返回結果不同，它返回一個Future對象，通過這個Future知道任務執行結果。
  • 添加GenericFutureListener， 執行完成(future可能由於正常終止、異常或取消而完成)后調用該監聽器。
如果我實現這個Future怎么實現：1：任務執行器，這樣我們可以控制任務的執行了。2：執行結果，用於返回。3：監聽器集合。4：執行結果狀態
5：觸發監聽器的函數。那么 怎么執行完成后自定觸發監聽器，應該在Future里面又啟動了一個線程去執行這個觸發機制。這些都是我的猜想。看下他的子類怎么實現的：
 
AbstractFuture實現了JDK中的get方法，調用netty中future的方法，一個模板模式出現了。所有的Future都會繼承該類
Promise：任務執行者可以標記任務執行成功或失敗，添加了setFailure(Throwable)可以知道Promise的子類需要有個成員變量來保存異常，添加了setSuccess(V) 方法，setUncancellable()方法， 和tryFailure和trySuccess方法。這樣所有的操作都可以返回一個Promise，你自己檢測是否執行成功，好處是啥，接口統一嗎？
ScheduledFuture：一個定時任務的執行結果
ProgressiveFuture：可以跟蹤任務的執行進度。
 
下面看下幾個類的具體實現：
DefaultPromise：成員變量如下：
  1. privatefinalEventExecutor executor; //任務執行器
  2. privatevolatileObject result;//不僅僅是結果，也有可能是異常
  3. * 一個或多個監聽器，可能是GenericFutureListener或者DefaultFutureListeners。如果是NULL有兩種可能
    * 1：沒有添加觸發器
    * 2：已經出發了
  4. privateObject listeners;
  5. privateLateListeners lateListeners;
  6. privateshort waiters;
看來異常也是結果。
看一個重要方法的實現：
isDone: 可能由於正常終止、異常或取消而完成
  1. privatestaticboolean isDone0(Object result){
  2. return result !=null&& result != UNCANCELLABLE;
  3. }
isSuccess: 任務執行成功
  1. publicboolean isSuccess(){
  2. Object result =this.result;
  3. if(result ==null|| result == UNCANCELLABLE){
  4. returnfalse;
  5. }
  6. return!(result instanceofCauseHolder);
  7. }
getNow:返回執行結果
  1. public V getNow(){
  2. Object result =this.result;
  3. if(result instanceofCauseHolder|| result == SUCCESS){
  4. returnnull;
  5. }
  6. return(V) result;
  7. }
sync:同步阻塞方法
  1. @Override
  2. publicPromise<V> sync()throwsInterruptedException{
  3. await();
  4. rethrowIfFailed();
  5. returnthis;
  6. }
看來比await多了拋出異常。
await：等待任務執行完成
  1. @Override
  2. publicPromise<V> await()throwsInterruptedException{
  3. if(isDone()){
  4. returnthis;
  5. }
  7. if(Thread.interrupted()){
  8. thrownewInterruptedException(toString());
  9. }
  11. synchronized(this){
  12. while(!isDone()){
  13. checkDeadLock();//判斷當前線程是否是執行線程。如果是拋出異常。
  14. incWaiters();//添加等待個數
  15. try{
  16. wait();//釋放鎖，等待喚醒，阻塞該線程
  17. }finally{
  18. decWaiters();
  19. }
  20. }
  21. }
  22. returnthis;
  23. }
執行器所在的線程不能調用await(),只能是調用者所在的線程才可以，waiters有什么用呢？
cancle方法使用：
  1. @Override
  2. publicboolean cancel(boolean mayInterruptIfRunning){
  3. Object result =this.result;
  4. if(isDone0(result)|| result == UNCANCELLABLE){
  5. returnfalse;
  6. }
  8. synchronized(this){
  9. // Allow only once.
  10. result =this.result;
  11. if(isDone0(result)|| result == UNCANCELLABLE){
  12. returnfalse;
  13. }
  15. this.result = CANCELLATION_CAUSE_HOLDER;
  16. if(hasWaiters()){
  17. notifyAll();
  18. }
  19. }
  21. notifyListeners();
  22. returntrue;
  23. }
當我們修改DefaultPromise的狀態時，要觸發監聽器。
notifyListeners:
  1. /**
  2. * 該方法不需要異步，為啥呢
  3. * 1：這個方法在同步代碼塊里面調用，因此任何監聽器列表的改變都happens-before該方法
  4. * 2：該方法只有isDone==true的時候調用，一但 isDone==true 那么監聽器列表將不會改變
  5. */
  6. privatevoid notifyListeners(){
  8. Object listeners =this.listeners;
  9. if(listeners ==null){
  10. return;
  11. }
  13. EventExecutor executor = executor();
  14. if(executor.inEventLoop()){
  15. finalInternalThreadLocalMap threadLocals =InternalThreadLocalMap.get();
  16. finalint stackDepth = threadLocals.futureListenerStackDepth();
  17. if(stackDepth < MAX_LISTENER_STACK_DEPTH){
  18. threadLocals.setFutureListenerStackDepth(stackDepth +1);
  19. try{
  20. if(listeners instanceofDefaultFutureListeners){
  21. notifyListeners0(this,(DefaultFutureListeners) listeners);
  22. }else{
  23. finalGenericFutureListener<?extendsFuture<V>> l =
  24. (GenericFutureListener<?extendsFuture<V>>) listeners;
  25. notifyListener0(this, l);
  26. }
  27. }finally{
  28. this.listeners =null;
  29. threadLocals.setFutureListenerStackDepth(stackDepth);
  30. }
  31. return;
  32. }
  33. }
  35. if(listeners instanceofDefaultFutureListeners){
  36. finalDefaultFutureListeners dfl =(DefaultFutureListeners) listeners;
  37. execute(executor,newRunnable(){
  38. @Override
  39. publicvoid run(){
  40. notifyListeners0(DefaultPromise.this, dfl);
  41. DefaultPromise.this.listeners =null;
  42. }
  43. });
  44. }else{
  45. finalGenericFutureListener<?extendsFuture<V>> l =
  46. (GenericFutureListener<?extendsFuture<V>>) listeners;
  47. execute(executor,newRunnable(){
  48. @Override
  49. publicvoid run(){
  50. notifyListener0(DefaultPromise.this, l);
  51. DefaultPromise.this.listeners =null;
  52. }
  53. });
  54. }
  55. }
任務永遠不能在主線程中執行，需要放到執行器所在的線程執行。 DefaultFutureListeners和 GenericFutureListener，一個是容器，一個是元素
還有幾個修改狀態的方法：
  1. @Override
  2. publicboolean setUncancellable(){
  3. Object result =this.result;
  4. if(isDone0(result)){
  5. return!isCancelled0(result);
  6. }
  8. synchronized(this){
  9. // Allow only once.
  10. result =this.result;
  11. if(isDone0(result)){
  12. return!isCancelled0(result);
  13. }
  15. this.result = UNCANCELLABLE;
  16. }
  17. returntrue;
  18. }
  20. privateboolean setFailure0(Throwable cause){
  21. if(cause ==null){
  22. thrownewNullPointerException("cause");
  23. }
  25. if(isDone()){
  26. returnfalse;
  27. }
  29. synchronized(this){
  30. // Allow only once.
  31. if(isDone()){
  32. returnfalse;
  33. }
  35. result =newCauseHolder(cause);
  36. if(hasWaiters()){
  37. notifyAll();
  38. }
  39. }
  40. returntrue;
  41. }
  43. privateboolean setSuccess0(V result){
  44. if(isDone()){
  45. returnfalse;
  46. }
  48. synchronized(this){
  49. // Allow only once.
  50. if(isDone()){
  51. returnfalse;
  52. }
  53. if(result ==null){
  54. this.result = SUCCESS;
  55. }else{
  56. this.result = result;
  57. }
  58. if(hasWaiters()){
  59. notifyAll();
  60. }
  61. }
  62. returntrue;
  63. }
CompleteFuture的幾個子類是狀態Promise
 
PromiseTask:該類繼承了RunnableFuture接口，該類表示異步操作的結果也可以異步獲得，類似JDK中的FutureTask，實例化該對象時候需要傳一個Callable的對象，如果沒有該對象可以傳遞一個Runnable和一個Result構造一個Callable對象。
  1. privatestaticfinalclassRunnableAdapter<T>implementsCallable<T>{
  2. finalRunnable task;
  3. final T result;
  5. RunnableAdapter(Runnable task, T result){
  6. this.task = task;
  7. this.result = result;
  8. }
  10. @Override
  11. public T call(){
  12. task.run();
  13. return result;
  14. }
  16. @Override
  17. publicString toString(){
  18. return"Callable(task: "+ task +", result: "+ result +')';
  19. }
  20. }
看下他的run方法。
  1. @Override
  2. publicvoid run(){
  3. try{
  4. if(setUncancellableInternal()){
  5. V result = task.call();
  6. setSuccessInternal(result);
  7. }
  8. }catch(Throwable e){
  9. setFailureInternal(e);
  10. }
該類的setFailure,setSuccess等方法都會拋出一異常，而加了internal的方法會成功執行，他們是protected，子類或者同一個package中可以調用。
ScheduledFutureTask:該類是定時任務返回的 ChannelFuture是這個結構中最重要的類，從名稱可以知道是通道異步執行的結果：在netty中所有的IO操作都是異步的。這意味這所有的IO調用都會立即返回，且不保證IO操作完成。 
IO調用會返回一個ChannelFuture的實例，通過該實例可以查看IO操作的結果和狀態，
ChannelFuture有完成和未完成兩種狀態，當IO操作開始，就會創建一個ChannelFuture的實例，該實例初始是未完成狀態，它不是成功，失敗，或者取消，因為IO操作還沒有完成，如果IO操作完成了那么將會有成功，失敗，和取消狀態， 
* +---------------------------+
* | Completed successfully |
* +---------------------------+
* +----> isDone() = <b>true</b> |
* +--------------------------+ | | isSuccess() = <b>true</b> |
* | Uncompleted | | +===========================+
* +--------------------------+ | | Completed with failure |
* | isDone() = <b>false</b> | | +---------------------------+
* | isSuccess() = false |----+----> isDone() = <b>true</b> |
* | isCancelled() = false | | | cause() = <b>non-null</b> |
* | cause() = null | | +===========================+
* +--------------------------+ | | Completed by cancellation |
* | +---------------------------+
* +----> isDone() = <b>true</b> |
* | isCancelled() = <b>true</b> |
* +---------------------------+
 該類提供了很多方法用來檢查IO操作是否完成，等待完成，和接受IO操作的結果。還可以添加ChannelFutureListener的監聽器，這樣IO操作完成時就可以得到提醒 * 強烈建議使用addListener而不是await。 * addListener是非阻塞的，它簡單的添加指定的ChannelFutureListener到ChannelFuture中， * IO線程將在當綁定在這個future的IO操作完成時，觸發這個觸發器，優點是提高效率和資源的利用率 * await()是一個阻塞方法，一旦調用，調用線程將會阻塞直到IO操作完成。優點是容易實現順序邏輯

 
 
 
 
 
 
 
 
 






免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 netty中的Future、ChannelFuture與ChannelPromise詳解 Netty 中的異步編程 Future 和 Promise Netty核心概念(9)之Future Netty概念之 Future 和 Promise Future Promise 模式（netty源碼9） Netty 源碼解析（三）: Netty 的 Future 和 Promise netty源碼分析之-Future、ChannelFuture與ChannelPromise詳解 你了解Java中的Future嗎？ java中Future的使用 Http-Netty-Rpc-Service系統改造：不要在springboot的啟動主線程中阻塞，nettyserver的postconstruct注解的init方法中不要寫future.channel().closeFuture().sync();
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM