网易视频云：从0到1学系统通信——异步通信的常用技术(二).docx

网易视频云：从0到1学系统通信异步通信的常用技术（二）网易视频云是网易公司旗下的视频云服务产品，以Paas服务模式，向开发者提供音视频编解码SDK和开放API，助力APP接入音视频功能。现在，网易视频云的技术专家给大家分享一篇技术性文章：从0到1学系统通信异步通信的常用技术（二）。上一篇文章介绍了java NIO，异步处理提倡更有效的使用资源，它允许你创建一个任务，当有事件发生时将获得通知并等待事件完成。这样就不会阻塞，不管事件完成与否都会及时返回，资源利用率更高，程序可以利用剩余的资源做一些其他的事情。本章将介绍异步调用中最常用的两种方法。CallBacks(回调)接触过GUI编程的都熟悉这种方法，当你在界面按下一个按钮，会调用回调函数进行处理。回调的思想在GUI的场景下，使得编程变得简单和易于理解。当然回调这种方法使用的地方很多，在了解其原理后，以后我们在看到这种方法的时候，就能加可以理解它是如何运行的了。下面是一个简单的回调代码示例。回调函数接口：public interface CallBack void rightProcess(Request input) throws Exception; void errorProcess(Throwable cause);示例类：public class Activer private Listlisteners = new ArrayList();public void addListener(Request request) Listener listener = new Listener(request);listener.registerListener(new CallBack() Overridepublic void rightProcess(Request input) throws Exception System.out.println(use the right method to process + input);Overridepublic void errorProcess(Throwable cause) System.out.println(use the error method to process + cause.getMessage(););listeners.add(listener);public void activeListen() for (Listener listener : listeners) listener.listen();public static void main(String avg) Activer activer = new Activer();activer.addListener(new Request(1, first to input);/一些应用场景中，可以一个线程注册回调函数activer.activeListen();/另一个线程触发回调函数请求类：public class Request final String id; final String message;public Request(String id, String message) this.id = id; this.message = message;public String toString() return id + : + message;监听类：public class Listener final Request input; private CallBack callback; public Listener(Request input) this.input = input;public void registerListener(CallBack callback) this.callback = callback;public void listen() try callback.rightProcess(input); catch (Exception e) callback.errorProcess(e); 在以上代码示例中解决两个问题：1.注册回调函数和实际调用回调函数时间点的解耦2.触发回调函数的线程可以和实际执行回调函数的线程解耦有了以上两个特性，再回想之前接触的GUI编程中的注册动作的回调函数，其原理一目了然了。在上一章提到的Reactor模式（Reactor释义“反应堆”，是一种事件驱动机制。和普通函数调用的不同之处在于：应用程序不是主动的调用某个API完成处理，而是恰恰相反，Reactor逆置了事件处理流程，应用程序需要提供相应的接口并注册到Reactor上，如果相应的时间发生，Reactor将主动调用应用程序注册的接口，这些接口又称为“回调函数”。）也是采用了这种实现方式。可见回调方法在异步处理中，具有广泛应用，掌握其核心原理和思路，对于理解其他异步通信框架有很大帮助。Futures在没有接触JAVA之前，看到这个future确实有很多困惑。这是一种全新的概念。Futures是一个抽象的概念，它表示一个值，该值可能在某一点变得可用。一个Future要么获得计算完的结果，要么获得计算失败后的异常。Java在java.util.concurrent包中附带了Future接口，它使用Executor异步执行。例如下面的代码，每传递一个Runnable对象到ExecutorService.submit()方法就会得到一个回调的Future，你能使用它检测是否执行完成。public class FutureExample public static void main(String args) throws Exception ExecutorService executor = Executors.newCachedThreadPool();Runnable task1 = new Runnable() Overridepublic void run() /do somethingSystem.out.println(i am task1.);Callabletask2 = new Callable() Overridepublic Integer call() throws Exception /do somethingreturn new Integer(100);Future f1 = executor.submit(task1);Futuref2 = executor.submit(task2);System.out.println(task1 is completed? + f1.isDone();System.out.println(task2 is completed? + f2.isDone(); /waiting task1 completedwhile(f1.isDone()System.out.println(task1 completed.); break; /waiting task2 completedwhile(f2.isDone()System.out.println(return value by task2: + f2.get(); break;查看源码发现，当执行Futuref2 = executor.submit(task2)时，调用的接口为：AbstractExecutorService类public Future submit(Callable task) if (task = null) throw new NullPointerException();RunnableFutureftask = newTaskFor(task);execute(ftask); return ftask;protected RunnableFuture newTaskFor(Callable callable) return new FutureTask(callable);FutureTask类public FutureTask(Callable callable) if (callable = null) throw new NullPointerException(); this.callable = callable; this.state = NEW; / ensure visibility of callable也就是返回的Future是一个futureTask。而public class FutureTask implements RunnableFuture；public interface RunnableFuture extends Runnable, Future 。因此返回的future就是一个含有执行过程中状态state的Runnable对象。由于这个Runnable对象的执行情况未知，但是其状态可以通过对state的读取感知到，其结果可以通过FutureTask的属性outcome得到。FutureTask类中的属性和run（）方法如下：private volatile int state;/* The underlying callable; nulled out after running */private Callable callable;/* The result to return or exception to throw from get() */private Object outcome; / non-volatile, protected by state reads/writes/* The thread running the callable; CASed during run() */private volatile Thread runner;/* Treiber stack of waiting threads */private volatile WaitNode waiters;public void run() if (state != NEW |!UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread() return; try Callablec = callable; if (c != null & state = NEW) V result; boolean ran; try result = c.call();ran = true; catch (Throwable ex) result = null;ran = false;setException(ex); if (ran)set(result); finally / runner must be non-null until state is settled to/ prevent concurrent calls to run()runner = null; / state must be re-read after nulling runner to prevent/ leaked interruptsint s = state; if (s = INTERRUPTING)handlePossibleCancellationInterrupt(s);在JAVA中Future (interface) 的使用是非常简单的，其实Future 本身是一种被广泛运用的并发设计模式，可在很大程度上简化需要数据流同步的并发应用开发。前面讲解java中Future的实现，下面来说明Future设计模式。Future模式中有如下几个概念：1 Client参与者，发送请求方，调用子线程的，它一旦发送请求后，就获取一个VirtualData，作为请求的结果。对应于FutureExample示例中调用executor.submit()2 Host参与者，接受请求者，会为请求者返回一个FutureData对象。并且会启动一个线程来执行真正的任务，最后将任务的返回结果RealData对象赋值给FutureData对象。 对应于javaFuture实现中的AbstractExecut