转载

【每日一博】基于 zbus 的 MySQL 透明代理（<100行）

项目地址 https://git.oschina.net/rushmore/zbus

我们上次讲到 zbus 网络通讯的核心 API ：

Dispatcher -- 负责 -NIO 网络事件 Selector 引擎的管理，对 Selector 引擎负载均衡

IoAdaptor -- 网络事件的处理，服务器与客户端共用，负责读写，消息分包组包等

Session -- 代表网络链接，可以读写消息

实际的应用，我们几乎只需要做 IoAdaptor 的个性化实现就能完成高效的网络通讯服务，今天我们将举例说明如何个性化这个 IoAdaptor 。

我们今天要完成的目标是：实现 MySQL 服务器的透明代理。效果是，你访问代理服务器跟访问目标 MySQL 无差异。

我们在测试环境 10.17.2.30 :3306 这台机器上提供了 MySql ，在我们本地机器上跑起来我们今天基于 zbus .NET 实现的一个代理程序，就能达到下面的效果。

【每日一博】基于 zbus 的 MySQL 透明代理（<100行）

完成大概不到 1 00 行的代码 , Cool ？ Let ’s roll !

首先，我们思考透明 TCP 代理到底在干啥，透明的 TCP 代理的业务逻辑其实非常简单，可以描述为，将来自代理上游（发起请求到代理）的数据转发到目标 TCP 服务器，把目标服务器回来的数据原路返回代理上游客户端。注意这个原路，如何做到原路返回成为关键点。这个示例其实跟 MySQL 没有任何关系，原则上任何 TCP 层面的服务都应该适配。

基于 zbus .NET 怎么来将上面的逻辑在体现出来，也就是如何个性化 IoAdaptor ？直观的讲，我们要处理的几个事件应该包括： 1 ）从上游客户端发起的链接请求 -- 代理服务器的 Accept 事件， 2 ）代理服务器连接目标服务器的 Connect 事件， 3 ）上下游的数据事件 onMessage 。

zbus.NET 的 IoAdaptor 提供的个性化事件如下

【每日一博】基于 zbus 的 MySQL 透明代理（<100行）

基本包括一个链接（客户端或者服务端）的生命周期，与消息的编解码。

我们的代理 IoAdaptor 就是逐一个性化处理。

第一步，编解码：透明代理对消息内容不做理解，所以不需要编解码。

// 透传不需要编解码，简单返回ByteBuffer数据  public IoBuffer encode(Object msg) {   if (msg instanceof IoBuffer) {    IoBuffer buff = (IoBuffer) msg;    return buff;   } else {    throw new RuntimeException("Message Not Support");   }  }   // 透传不需要编解码，简单返回ByteBuffer数据  public Object decode(IoBuffer buff) {   if (buff.remaining() > 0) {    byte[] data = new byte[buff.remaining()];    buff.readBytes(data);    return IoBuffer.wrap(data);   } else {    return null;   }  }

第二步，代理服务接入：

@Override  protected void onSessionAccepted(Session sess) throws IOException {   Session target = null;   Dispatcher dispatcher = sess.getDispatcher();   try {    target = dispatcher.createClientSession(targetAddress, this);   } catch (Exception e) {    sess.asyncClose();    return;   }   sess.chain = target;   target.chain = sess;   dispatcher.registerSession(SelectionKey.OP_CONNECT, target);  }

这里的逻辑思路是，代理服务器每接受到一个请求 -- 通过 onSessionAccepted 表达，我们将同时创建一个到目标服务器的链接，今天的例子是目标 MySQL 服务器，注意上面的处理中把创建目标服务器 Session 过程与真正链接到目标服务分开（ Dispatcher 也提供合并二者的工具方法），是为了能在没有发生链接之前绑定上好上下游关系，通过 Session 的 chain 变量来表达，也就是当前 Session 的关联 Session ，关联好之后启动感兴趣 Connect 事件，逻辑处理完毕。

第三步，链接成功事件（第二步中需要链接到目标服务器）

@Override  public void onSessionConnected(Session sess) throws IOException {     Session chain = sess.chain;   if(chain == null){     sess.asyncClose();    return;    }      if(sess.isActive() && chain.isActive()){     sess.register(SelectionKey.OP_READ);    chain.register(SelectionKey.OP_READ);   }  }

这里的一个核心是当上下游都处于链接正常态，上下游 Session 都启动感兴趣消息读事件（写事件是在读取处理中自动触发），为什么在这里做的原因是一定要等上下游都正常态后才启动双方消息处理，不然会出现字节丢失。

第四步，处理上下游数据事件

@Override  protected void onMessage(Object msg, Session sess) throws IOException {     Session chain = sess.chain;   if(chain == null){    sess.asyncClose();     return;   }    chain.write(msg);   }

是不是非常简单，类似 pipeline ，从一端的数据写到另外一端。

原则上面 4 步结束，整个透明代理就完成了，但是为了处理链接异常清理，我们增加了 Session 清理处理，如下

@Override  public void onSessionToDestroy(Session sess) throws IOException {      try {    sess.close();   } catch (IOException e) { //ignore   }    if (sess.chain == null) return;    try {     sess.chain.close();     sess.chain.chain = null;    sess.chain = null;   } catch (IOException e) {    }  }

工作就是解决上下游链接清理链接。

至此为止我们的 IoAdaptor 个性化就完成了，是不是非常简单，现在我们要跑起来测试了，下面的代码就是上一次讲到重复的设置，没有新意。

public static void main(String[] args) throws Exception {      Dispatcher dispatcher = new Dispatcher();    IoAdaptor ioAdaptor = new TcpProxyAdaptor("10.17.2.30:3306");    final Server server = new Server(dispatcher, ioAdaptor, 3306);    server.start();  }

骚年，包括渣渣 import和少许注释加起来折腾了不到 100 行，该跑一跑了，还是那句话，不是 HelloWorld ，你可以规模压力测。看看你是否在本地代理出来了你的目标服务 MySQL ， gl,hf, gogogo.

完整代码可运行代码如下,也可直接到zbus示例代码库中找到

https://git.oschina.net/rushmore/zbus/blob/master/src/test/java/org/zbus/net/TcpProxyAdaptor.java?dir=0&filepath=src%2Ftest%2Fjava%2Forg%2Fzbus%2Fnet%2FTcpProxyAdaptor.java&oid=08abff381d93519485e1c0ee2c35f1d4f8d1814c&sha=a29272ed99a8f21ec19a14b403ebee53a385e9a4

package org.zbus.net; import java.io.IOException; import java.nio.channels.SelectionKey; import org.zbus.net.core.Dispatcher; import org.zbus.net.core.IoAdaptor; import org.zbus.net.core.IoBuffer; import org.zbus.net.core.Session;   public class TcpProxyAdaptor extends IoAdaptor {  private String targetAddress;   public TcpProxyAdaptor(String targetAddress) {   this.targetAddress = targetAddress;  }   // 透传不需要编解码，简单返回ByteBuffer数据  public IoBuffer encode(Object msg) {   if (msg instanceof IoBuffer) {    IoBuffer buff = (IoBuffer) msg;    return buff;   } else {    throw new RuntimeException("Message Not Support");   }  }   // 透传不需要编解码，简单返回ByteBuffer数据  public Object decode(IoBuffer buff) {   if (buff.remaining() > 0) {    byte[] data = new byte[buff.remaining()];    buff.readBytes(data);    return IoBuffer.wrap(data);   } else {    return null;   }  }   @Override  protected void onSessionAccepted(Session sess) throws IOException {   Session target = null;   Dispatcher dispatcher = sess.getDispatcher();   try {    target = dispatcher.createClientSession(targetAddress, this);   } catch (Exception e) {    sess.asyncClose();    return;   }   sess.chain = target;   target.chain = sess;   dispatcher.registerSession(SelectionKey.OP_CONNECT, target);  }    @Override  public void onSessionConnected(Session sess) throws IOException {     Session chain = sess.chain;   if(chain == null){     sess.asyncClose();    return;    }      if(sess.isActive() && chain.isActive()){     sess.register(SelectionKey.OP_READ);    chain.register(SelectionKey.OP_READ);   }  }   @Override  protected void onMessage(Object msg, Session sess) throws IOException {     Session chain = sess.chain;   if(chain == null){    sess.asyncClose();     return;   }    chain.write(msg);   }    @Override  public void onSessionToDestroy(Session sess) throws IOException {      try {    sess.close();   } catch (IOException e) { //ignore   }    if (sess.chain == null) return;    try {     sess.chain.close();     sess.chain.chain = null;    sess.chain = null;   } catch (IOException e) {    }  }    @SuppressWarnings("resource")  public static void main(String[] args) throws Exception {      Dispatcher dispatcher = new Dispatcher();    IoAdaptor ioAdaptor = new TcpProxyAdaptor("10.17.2.30:3306");    final Server server = new Server(dispatcher, ioAdaptor, 3306);   server.setServerName("TcpProxyServer");   server.start();  } }

正文到此结束