上一节我们讲到netty启动服务类AbstractBootstrap的doBind的方法&＃xff1a;
private ChannelFuture doBind(final SocketAddress localAddress) {final ChannelFuture regFuture &＃61; initAndRegister();...略}
这里边有一个重要的类ChannelFuture &＃xff1a;
* Sample Usage (Note that the following classes are all* made-up.)* Memory consistency effects: Actions taken by the asynchronous computation* "package-summary.html#MemoryVisibility"> happen-before* actions following the corresponding {&＃64;code Future.get()} in another thread.* 内存一致性影响&＃xff1a;异步计算的动作的完成&＃xff0c;发生在Future.get()之前。* &＃64;see FutureTask* &＃64;see Executor* &＃64;since 1.5* &＃64;author Doug Lea* &＃64;param 最红他们的接口会来到jdk的Future接口&＃xff0c;Future代表了一个异步处理的结果。
/*** A {&＃64;code Future} represents the result of an asynchronous* computation. Methods are provided to check if the computation is* complete, to wait for its completion, and to retrieve the result of* the computation. The result can only be retrieved using method* {&＃64;code get} when the computation has completed, blocking if* necessary until it is ready. Cancellation is performed by the* {&＃64;code cancel} method. Additional methods are provided to* determine if the task completed normally or was cancelled. Once a* computation has completed, the computation cannot be cancelled.* If you would like to use a {&＃64;code Future} for the sake* of cancellability but not provide a usable result, you can* declare types of the form {&＃64;code Future} and* return {&＃64;code null} as a result of the underlying task.译&＃xff1a;
一个Future代表一个一步计算的结果&＃xff0c;它提供了一些方法检查是否计算完毕&＃xff0c;比如等待计算完毕&＃xff0c;获取计算结果的方法。
当计算完毕之后只能通过get方法获取结果&＃xff0c;或者一直阻塞等待计算的完成。取消操作通过cancle方法来取消&＃xff0c;
另外还提供了检测是正常的完成还是被取消的方法&＃xff0c;当一个计算完成后&＃xff0c;不能进行取消操作。如果你想用Future实现取消&＃xff0c;
但是却没有一个可用的结果&＃xff0c;你可以声明很多Future的类型&＃xff0c;然后返回一个null的结果给当前任务。* {&＃64;code* interface ArchiveSearcher { String search(String target); }* class App {* ExecutorService executor &＃61; ...//线程池* ArchiveSearcher searcher &＃61; ...//搜索接口* void showSearch(final String target)* throws InterruptedException {* Future** The {&＃64;link FutureTask} class is an implementation of {&＃64;code Future} that* implements {&＃64;code Runnable}, and so may be executed by an {&＃64;code Executor}.* FutureTask是Future的实现类&＃xff0c;并且实现了Runnable接口&＃xff0c;因此可以被Executor执行。* For example, the above construction with {&＃64;code submit} could be replaced by:* 之前的方式可以被下边的FutureTask的方式替换。* {&＃64;code* FutureTask**
io.netty.util.concurrent.Future对java.util.concurrent.Future进行了扩展&＃xff1a;
public interface Future<V> extends java.util.concurrent.Future<V> {boolean isSuccess();//是否计算成功boolean isCancellable();//可以被取消Throwable cause();//原因Future
}
提到一个监听器GenericFutureListener的封装&＃xff0c;一碰到XXXlistener&＃xff0c;都会用到监听器模式&＃xff1a;
/*** Listens to the result of a {&＃64;link Future}. The result of the asynchronous operation is notified once this listener* is added by calling {&＃64;link Future#addListener(GenericFutureListener)}.* 监听Future的结果&＃xff0c;当一个监听器被注册后&＃xff0c;结果的异步操作会被注册的监听器监听。*/
public interface GenericFutureListener<F extends Future> extends EventListener {/*** Invoked when the operation associated with the {&＃64;link Future} has been completed.** 和Future的完成计算相关的事件&＃xff0c;次方法会被调用。*/void operationComplete(F future) throws Exception;
}
监听器对Future的扩展起到了很灵活的作用&＃xff0c;当某个计算完毕&＃xff0c;会触发相应的时间&＃xff0c;得到Future的结果&＃xff0c;因为jdk的get方法我们知道什么时候去掉&＃xff0c;调早了需要等待&＃xff0c;调晚了浪费了一段时间&＃xff0c;还有isDone里边有2种情况&＃xff0c;无法区分到底是正常的io完毕返回的true还是被取消之后返回的true&＃xff0c;所有到了netty的Future里边加了一个isSuccess()方法&＃xff0c;只有正常的io处理结束isSuccess()才返回true。
接下来我们会走一下ChannelFuture的源码的doc&＃xff1a;
* All I/O operations in Netty are asynchronous. It means any I/O calls will* return immediately with no guarantee that the requested I/O operation has* been completed at the end of the call. Instead, you will be returned with* a {&＃64;link ChannelFuture} instance which gives you the information about the* result or status of the I/O operation.* netty中所有的i/o都是异步的&＃xff0c;意味着很多i/o操作被调用过后会立刻返回&＃xff0c;并且不能保证i/o请求操作被调用后计算已经完毕&＃xff0c;* 替代它的是返回一个当前i/o操作状态和结果信息的ChannelFuture实例。* * A {&＃64;link ChannelFuture} is either uncompleted or completed.* When an I/O operation begins, a new future object is created. The new future* is uncompleted initially - it is neither succeeded, failed, nor cancelled* because the I/O operation is not finished yet. If the I/O operation is* finished either successfully, with failure, or by cancellation, the future is* marked as completed with more specific information, such as the cause of the* failure. Please note that even failure and cancellation belong to the* completed state.* 一个ChannelFuture要么是完成的&＃xff0c;要么是未完成的。当一个i/o操作开始的时候&＃xff0c;会创建一个future 对象&＃xff0c;future 初始化的时候是为完成的状态&＃xff0c;* 既不是是成功的&＃xff0c;或者失败的&＃xff0c;也不是取消的&＃xff0c;因为i/o操作还没有完成&＃xff0c;如果一个i/o不管是成功&＃xff0c;还是失败&＃xff0c;或者被取消&＃xff0c;future 会被标记一些特殊* 的信息&＃xff0c;比如失败的原因&＃xff0c;请注意即使是失败和取消也属于完成状态。* * {&＃64;link #addListener(GenericFutureListener)} is non-blocking. It simply adds* the specified {&＃64;link ChannelFutureListener} to the {&＃64;link ChannelFuture}, and* I/O thread will notify the listeners when the I/O operation associated with* the future is done. {&＃64;link ChannelFutureListener} yields the best* performance and resource utilization because it does not block at all, but* it could be tricky to implement a sequential logic if you are not used to* event-driven programming.* addListener(GenericFutureListener)本身是非阻塞的&＃xff0c;他会添加一个指定的ChannelFutureListener到ChannelFuture* 并且i/o线程在完成对应的操作将会通知监听器&＃xff0c;ChannelFutureListener也会提供最好的性能和资源利用率&＃xff0c;因为他永远不会阻塞&＃xff0c;但是如果* 不是基于事件编程&＃xff0c;他可能在顺序逻辑存在棘手的问题。* * By contrast, {&＃64;link #await()} is a blocking operation. Once called, the* caller thread blocks until the operation is done. It is easier to implement* a sequential logic with {&＃64;link #await()}, but the caller thread blocks* unnecessarily until the I/O operation is done and there&＃39;s relatively* expensive cost of inter-thread notification. Moreover, there&＃39;s a chance of* dead lock in a particular circumstance, which is described below.*相反的&＃xff0c;await()是一个阻塞的操作&＃xff0c;一旦被调用&＃xff0c;调用者线程在操作完成之前是阻塞的&＃xff0c;实现顺序的逻辑比较容易&＃xff0c;但是他让调用者线程等待是没有必要* 的&＃xff0c;会造成资源的消耗&＃xff0c;更多可能性会造成死锁&＃xff0c;接下来会介绍。* * The event handler methods in {&＃64;link ChannelHandler} are usually called by* an I/O thread. If {&＃64;link #await()} is called by an event handler* method, which is called by the I/O thread, the I/O operation it is waiting* for might never complete because {&＃64;link #await()} can block the I/O* operation it is waiting for, which is a dead lock.* ChannelHandler里边的时间处理器通常会被i/o线程调用&＃xff0c;如果await()被一个时间处理方法调用&＃xff0c;并且是一个i/o线程&＃xff0c;那么这个i/o操作将永远不会 * 完成&＃xff0c;因为await()是会阻塞i/o操作&＃xff0c;这是一个死锁。* * In spite of the disadvantages mentioned above, there are certainly the cases* where it is more convenient to call {&＃64;link #await()}. In such a case, please* make sure you do not call {&＃64;link #await()} in an I/O thread. Otherwise,* {&＃64;link BlockingOperationException} will be raised to prevent a dead lock.* 尽管出现了上面提到的这些缺陷&＃xff0c;但是在某些情况下更方便&＃xff0c;在这种情况下&＃xff0c;请确保不要再i/o线程里边调用await()方法&＃xff0c;* 否则会出现BlockingOperationException异常&＃xff0c;导致死锁。*
/*** The result of an asynchronous {&＃64;link Channel} I/O operation.* Channel的异步io操作的结果。* * &＃43;---------------------------&＃43;* | Completed successfully |* &＃43;---------------------------&＃43;* &＃43;----> isDone() &＃61; true |* &＃43;--------------------------&＃43; | | isSuccess() &＃61; true |* | Uncompleted | | &＃43;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃43;* &＃43;--------------------------&＃43; | | Completed with failure |* | isDone() &＃61; false | | &＃43;---------------------------&＃43;* | isSuccess() &＃61; false |----&＃43;----> isDone() &＃61; true |* | isCancelled() &＃61; false | | | cause() &＃61; non-null |* | cause() &＃61; null | | &＃43;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃43;* &＃43;--------------------------&＃43; | | Completed by cancellation |* | &＃43;---------------------------&＃43;* &＃43;----> isDone() &＃61; true |* | isCancelled() &＃61; true |* &＃43;---------------------------&＃43;* ** Various methods are provided to let you check if the I/O operation has been* completed, wait for the completion, and retrieve the result of the I/O* operation. It also allows you to add {&＃64;link ChannelFutureListener}s so you* can get notified when the I/O operation is completed.* ChannelFuture提供了很多方法让你检查i/o操作是否完成、等待完成、获取i/o操作的结果&＃xff0c;他也允许你添加ChannelFutureListener* 因此可以在i/o操作完成的时候被通知。* Prefer {&＃64;link #addListener(GenericFutureListener)} to {&＃64;link #await()}
* 建议使用addListener(GenericFutureListener)&＃xff0c;而不使用await()* It is recommended to prefer {&＃64;link #addListener(GenericFutureListener)} to* {&＃64;link #await()} wherever possible to get notified when an I/O operation is* done and to do any follow-up tasks.* 推荐优先使用addListener(GenericFutureListener)&＃xff0c;不是await()在可能的情况下&＃xff0c;这样就能在i/o操作完成的时候收到通知&＃xff0c;并且可以去做 * 后续的任务处理。* Do not call {&＃64;link #await()} inside {&＃64;link ChannelHandler}
* 不要再ChannelHandler里边调用await()方法* * // BAD - NEVER DO THIS 不推荐的使用方式* {&＃64;code &＃64;Override}* public void channelRead({&＃64;link ChannelHandlerContext} ctx, Object msg) {* {&＃64;link ChannelFuture} future &＃61; ctx.channel().close();* future.awaitUninterruptibly();//不要使用await的 方式* // Perform post-closure operation* // ...* }** // GOOD* {&＃64;code &＃64;Override} //推荐使用的方式* public void channelRead({&＃64;link ChannelHandlerContext} ctx, Object msg) {* {&＃64;link ChannelFuture} future &＃61; ctx.channel().close();* future.addListener(new {&＃64;link ChannelFutureListener}() {//使用时间的方式* public void operationComplete({&＃64;link ChannelFuture} future) {* // Perform post-closure operation* // ...* }* });* }* * Do not confuse I/O timeout and await timeout
*不要将i/o超时和等待超时混淆。* The timeout value you specify with {&＃64;link #await(long)},* {&＃64;link #await(long, TimeUnit)}, {&＃64;link #awaitUninterruptibly(long)}, or* {&＃64;link #awaitUninterruptibly(long, TimeUnit)} are not related with I/O* timeout at all. If an I/O operation times out, the future will be marked as* &＃39;completed with failure,&＃39; as depicted in the diagram above. For example,* connect timeout should be configured via a transport-specific option:* 使用await(long)、await(long, TimeUnit)、awaitUninterruptibly(long)、awaitUninterruptibly(long, TimeUnit)设置的超时时间* 和i/o超时没有任何关系&＃xff0c;如果一个i/o操作超时&＃xff0c;future 将被标记为失败的完成状态&＃xff0c;比如连接超时通过一些选项来配置&＃xff1a;* * // BAD - NEVER DO THIS //不推荐的方式* {&＃64;link Bootstrap} b &＃61; ...;* {&＃64;link ChannelFuture} f &＃61; b.connect(...);* f.awaitUninterruptibly(10, TimeUnit.SECONDS);//不真正确的等待超时* if (f.isCancelled()) {* // Connection attempt cancelled by user* } else if (!f.isSuccess()) {* // You might get a NullPointerException here because the future//不能确保future 的完成。* // might not be completed yet.* f.cause().printStackTrace();* } else {* // Connection established successfully* }** // GOOD//推荐的方式* {&＃64;link Bootstrap} b &＃61; ...;* // Configure the connect timeout option.* b.option({&＃64;link ChannelOption}.CONNECT_TIMEOUT_MILLIS, 10000);</b>//配置连接超时* {&＃64;link ChannelFuture} f &＃61; b.connect(...);* f.awaitUninterruptibly();** // Now we are sure the future is completed.//确保future 一定是完成了。* assert f.isDone();** if (f.isCancelled()) {* // Connection attempt cancelled by user* } else if (!f.isSuccess()) {* f.cause().printStackTrace();* } else {* // Connection established successfully* }* */
public interface ChannelFuture extends Future
}
下一接介绍initAndRegister( )方法。
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有 