Netty - 异步与回调

snoopy日志 2019-09-19

764

开始之前

异步与回调，一个有趣的话题。我们总是对异步编程抱有很大的兴趣，因为正确的运用异步，可以使你的程序变快。但我们又想在异步完成时获得通知，怎么办？在java的concurrent包中提供了Future接口，可以获取异步调用的返回结果，但这个获取是同步的，显然破坏了异步的调用形式。那么本文将会阐述如何使用回调机制让程序真正的异步。

回调

回调是一种双向调用：在类A的a()方法中调用了类B的b()方法，b()方法执行完之后调用A的callback来告知b()方法已经执行完毕，那么如何才能调用到callback()呢，或者说如何做到通知A呢，很显然b()方法执行时要持有A的引用，所以B的b方法应该是：

b(A ref)。类似下面的伪代码：

class A{
  void a(B ref){
    ref.b();
  };
  void callback();
}
class B{
  void b(A ref){
    ...
    ref.callback();
  }
  main(){
    new A().a(new B());
  }
}

上面这种是同步回调，即回调放在同一个线程内。多数情况下我们更关注文章开篇提到的异步回调。

异步回调

两大要素：

回调接口(Callback)
回调监听(CallbackListener)

被调用方返回一个Callback给调用方，调用方在Callback上注册监听。被调用方在程序最后执行监听。这里有一个注意点：当调用方在放置监听的时候发现被调用方已经完成了自己代码，则调用方需要立即执行监听。

下面是作者简单实现的异步回调例子：

public interface Callback{
  public void addListener(CallbackListener listener);//注册监听
  public void notifyListener();//唤起监听
  public void setComplete(boolean b);//标记被调用方已完成
}


public interface CallbackListener{
  public void operationComplete(Callback callBack);
}


public class DefaultCallback implements Callback {
    private List<CallbackListener> ls = new ArrayList<>();
    private boolean completed = false;
    @Override
    public void addListener(CallbackListener listener) {
        if(completed){//如果已完成，则立即执行
            listener.operationComplete(this);
        }else{
            ls.add(listener);
        }
    }
    @Override
    public void notifyListener() {
        for(CallbackListener l : ls){
            l.operationComplete(this);
        }
    }
    @Override
    public void setComplete(boolean b) {
        this.completed = b;
    }
}

上面是基本组件，下面我们定义一个被调用方的例子

public class A{
  public Callback invoke(){
  Callback callback = new DefaultCallback();
    //用线程实现异步
    new Thread(new Runnable(){
      public void run(){
        System.out.println("do something...");
        callback.setComplete(true);
        callback.notifyListener();
      }
    }).start();
    return callback;
  }
}

写一个测试程序

public class Test {
    public static void main(String[] args) {
        A a = new A();
        Callback callback = a.invoke();
        callback.addListener(new CallbackListener() {
            @Override
            public void operationComplete(Callback callBack) {
                System.out.println("do callback...");
            }
        });
    }
}

如果大家用这几段代码做了测试就能体会到，异步回调除了可以异步通知之外，还能保证代码在逻辑上的同步。

Netty中的实现

Netty中实现回调或者future通知的机制与上面的相似。

Netty中对Channel上IO操作都会立马返回并得到一个ChannelFuture引用，我们可以在ChannelFuture中注册监听 - ChannelFutureListener，这样我们就能在操作完成时获得通知。如下利用此机制实现在成功连接后的一些操作：

Channe channel = ...;
ChannelFuture future = 
channel.connect(new InetSocketAddress("localhost",8848));
future.addListener(new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
          if(future.isSucess()){
            System.out.println("do something...");
          }else{
            ...
          }
      }
 });

结束语

本文的重点在于中篇的异步回调，如果能明白了这个模式，也就明白了Netty中完全异步是怎么实现的。建议大家可以结合最后netty的例子调式下，这样才会有更深的理解。

下篇将会继续介绍读写事件在netty中是如何优雅处理的。

数据库

文章转载自snoopy日志，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

Netty - 异步与回调

评论