时间戳

首页 > 技术文章 > 一篇文章让你成为 NIO 大师 － MyCAT通信模型

一篇文章让你成为 NIO 大师 － MyCAT通信模型

geason 原文

这篇文章没有详细介绍 NIO 的概念，对于 NIO 不了解的同学，可根据自己需要，阅读这篇介绍 NIO 的博客 
 
io.mycat.net.NIOAcceptor
NIOAcceptor负责处理客户端（指连接MyCAT以访问数据库的程序）的连接请求。
NIOAcceptor中持有一个Selector字段selector，和一个ServerSocketChannel字段serverChannel。并向selector中注册serverChannel
，并注册感兴趣的事件为 OP_ACCEPT。
NIOAcceptor继承了Thread接口，在run()方法内，生成了一个selector的拷贝tSeletor，并由tSeletor调用select()方法轮询事件是否就绪。
当检测到前段一个连接请求过来时，会调用serverChannel的accept()方法创建一个新的通道，并从工厂获取一个新的前端连接实例（MyCAT和客户端的连接称为前端连接，MyCAT和MySQL的连接称为后端连接），将通道和连接实例绑定。该连接实例由io.mycat.net.NIOProcessor负责管理，NIOProcessor在MyCAT中也是以NIOProcessor池的形式存在的，acceptor从池中拿出一个Processor实例，并将其和连接实例绑定。随后，NIOAcceptor将该连接转交给io.mycat.net.NIOReactor，
NIOAcceptor的工作就结束了。
 
io.mycat.net.NIOReactor
NIOReactor中声明了一个final的内部类 RW，RW继承了Runnable。RW中持有一个Selector字段selector，和一个用来保存AbstractConnection的ConcurrentLinkedQueue队列registerQueue。在RW的run()方法中，调用selector的select()方法（该selector也是由RW的字段selector的拷贝而来，以后如非特别说明，皆是如此），轮询 I/O 事件。
NIOReactor在接到NIOAcceptor发来的前端连接实例后，会将其添加到registerQueue的队尾，并调用selector的wakeup()方法使selector立刻返回。阻塞在select()方法之后的是RW的register方法，调用register(这个方法设计的别出心裁，之后会贴出代码)，RW会从registerQueue中取出队首的连接实例c，从c中获取NIOSocketWR实例，调用NIOSocketWR的register方法，该方法从连接实例中获取通道，向selector中注册该通道，注册感兴趣的事件为OP_READ。随后，调用连接实例c的register()方法，生成认证数据，发送握手数据包，建立TCP连接。连接建立后，调用NIOSocketWR的asynRead()异步读方法（为什么这么做，在贴出代码中会讲述）。
前面讲到RW的run()方法会循环调用selector的select()方法，除了当有新连接加入时，会直接返回之外，selector只有当检测到有注册的 OP_READ 事件就绪时才会返回。当有通道的读事件就绪时，RW会判断该事件是一个“读”事件还是一个“写”事件，并调用连接实例的相应方法处理该读／写事件。为什么还要判断？因为从名字就可以得知，这是一个读写复用的处理类，虽然当前我们处理的是读事件。
asynRead()异步读方法：获取缓冲区，调用通道的read方法，将数据读到缓冲区。随后，调用连接实例的处理方法处理缓冲区的字节流信息。
 
代码（注释是我加的）
/**
 * 当一个连接由 Acceptor 转发过来时,会使得 selector 马上返回,调用该方法,
 * 将该连接及其通道注册到 selector 中,  注册 OP_READ 事件。
 * 随后,建立 TCP 连接,建立 TCP 连接的时候也会调用异步读取数据。
 * 若此时恰好有数据到达,则可直接读取。
 * 若此时没有数据到达,则继续执行 select() 轮询 I/O 事件。
 *
 * 当 selector 轮询到有 I/O 事件就绪,而返回时,
 * 如此时 registerQueue 是一定为空的。
 */
private void register(Selector selector) {
   AbstractConnection c = null;
   if (registerQueue.isEmpty()) {
      return;
   }
   while ((c = registerQueue.poll()) != null) {
      try {
         ((NIOSocketWR) c.getSocketWR()).register(selector);
         c.register();
      } catch (Exception e) {
         c.close("register err" + e.toString());
      }
   }
}
 
io.mycat.net.AbstractConnection
前面反复提到一个词：连接实例，究竟什么是连接实例？客户端发往MyCAT的每一次请求，以及MyCAT发往MySQL的每一次请求，都是一个连接实例。可以把连接实例看作是一次请求事件的主干，我们都知道，NIO是一个同步非阻塞的 I/O 模型。阻塞的线程没有做任何有意义的事情，却依然消耗系统资源，这是我们不能接受的，所谓非阻塞，就是不断的在这条主干上衍生分支，来处理复杂的业务请求，这样主干就不会阻塞。而同步，是指线程不断轮询 IO 事件是否就绪，主干上衍生的这些分支，都维护了一个Selector对象，Selector代替了主干线程来执行这种轮询，包括前面讲到的acceptor和reactor；这些分支线程是以线程池的形式存在的，是可以复用的，从而减少了频繁创建、启动、挂起、析构新线程的开销，大大提升系统的并发效率。
 
io.mycat.net.NIOHandler
前面讲到了，当数据读到缓冲区后，调用连接实例的处理方法处理缓冲区的字节流。那么，这里是如何处理的呢？事实上，连接实例会先从将数据流从缓冲区读出来，回想一下，这是一个MySQL的中间件，所有的数据都是SQL语句。所以，接下来就是对字节流形式的SQL解包。不要忘了计算机网络的知识，端与端之间的通信是要按照某种协议的，这就是包头。所以，接下来的工作就是对包头进行解压，分析。经过这一步，MyCAT已经知道了客户端发来的SQL语句是什么类型的语句（登陆、增删改查等）。然后，就会调用 NIOHandler 来处理这一条 SQL 语句。
NIOHandler和客户端相关的实体类有：FrontendAuthenticator、FrontendCommandHandler。从名字就可以看出，一个是负责权限验证的，一个是负责处理命令行的。
在FrontendCommandHandler，会根据解析过的包头，根据不同的SQL语句类型，调用连接实例的相应方法。
 
接下来的事情
接下来就是，在连接实例的细分方法中，将字节流形式的信息转成字符串；然后就是SQL语法分析，生成语法树；展开语法树，计算路由节点；再接下来就是将SQL发给MySQL服务器；然后合并结果集，返回客户端。
我不准备讲语法解析和路由的部分，这不是我们的重点。所以，现在我们假设已经做完了这两步，接下来要做的就是将SQL发给MySQL真正执行。
 
NonBlockingSession、SingleNodeHandler／MultiNodeHandler
这个时候，前端连接实例会调用自己NonBlockingSession类型的session字段的execute方法，execute也只做了一件事情，就是根据返回的RouteResultset是单节点的还是多节点的，决定是调用SingleNodeHandler还是MultiNodeHandler。session中维护了与每个节点的后端连接，在nodeHandler中，会从session中取得需要的后端连接，然后只做了一件事，就是将自己设置为后端连接实例的回调。随后，真正的执行就交给了后端连接。
之所以要经过设置回调这一步，是因为nodeHandler会负责解析由MySQL发回的消息。
 
MySQLConnection
MySQLConnection的execute方法中，将经过解析的SQL语句重新封装成消息包，并将该消息包加入到写队列中。前面出现过一个NIOSocketWR的类，因为前端的连接是NIO的，而MyCAT与MySQL的连接是由AIO实现的，因此，MySQLConnection会将把写队列的缓冲区写到Channel的任务交给了AIOSocketWR，AIOSocketWR负责维护一个AsynchronousSocketChannel类型的channel对象，通过调用AsynchronousSocketChannel的：
write(ByteBuffer src, A attachment, CompletionHandler<Integer,? super A> handler)
 
其中，传入的attachment是AIOSocketWR本身，并实现了一个CompletionHandler类AIOWriteHandler，需要重写两个方法，分别是completed和failed，当写入完成后会回调相应的方法。
 
到这个时候，将客户端的命令发给MySQL的工作就做完了，接下来的就是等待MySQL返回结果了。
 
如何知道MySQL是否返回结果了？也是在NIOReactor（mycat对nio和aio关系处理的有点乱）,nioreactor轮询到有消息过来的时候，就会教给连接实例去执行异步读方法，这个方法中又调用了socketWR的异步读。注意这个时候连接实例已经是后端连接了，所以它会调用AIOSocketWR。异步读和异步写相同，都是使用了Java NIO包封装的类AsynchronousSocketChannel，调用：
read(ByteBuffer src, A attachment, CompletionHandler<Integer,? super A> handler)
 
在completed中，会调用connection的处理方法，而connection，则会交给之前注册过的回调handler，就是SingleNodeHandler或MultiNodeHandler。
因为MySQL返回的包是一行一行的，因此会多次调用异步读方法。
 
SingleNodeHandler和MultiNodeHandler都是继承了ResponseHandler，通过观看源码，可以更加轻松地理解这种回调是如何进行的。
 
public interface ResponseHandler {

  /**
    * 无法获取连接
    *
    * @param e
    * @param conn
    */
  public void connectionError(Throwable e, BackendConnection conn);
 
  /**
    * 已获得有效连接的响应处理
    */
  void connectionAcquired(BackendConnection conn);
 
  /**
    * 收到错误数据包的响应处理
    */
  void errorResponse(byte[] err, BackendConnection conn);
 
  /**
    * 收到OK数据包的响应处理
    */
  void okResponse(byte[] ok, BackendConnection conn);
 
  /**
    * 收到字段数据包结束的响应处理
    */
  void fieldEofResponse(byte[] header, List<byte[]> fields, byte[] eof,
        BackendConnection conn);
 
  /**
    * 收到行数据包的响应处理
    */
  void rowResponse(byte[] row, BackendConnection conn);
 
  /**
    * 收到行数据包结束的响应处理
    */
  void rowEofResponse(byte[] eof, BackendConnection conn);
 
  /**
    * 写队列为空，可以写数据了
    *
    */
  void writeQueueAvailable();
 
  /**
    * on connetion close event
    */
  void connectionClose(BackendConnection conn, String reason);
 
 
}
 

推荐阅读