网站设计公司南京,响应式布局的概念,优化网络培训,网站建设课后感项目地址 https://git.oschina.net/rushmore/zbus 我们上次讲到zbus网络通讯的核心API#xff1a; Dispatcher -- 负责-NIO网络事件Selector引擎的管理#xff0c;对Selector引擎负载均衡 IoAdaptor -- 网络事件的处理#xff0c;服务器与客户端共用#xff0c;负责读写 Dispatcher -- 负责-NIO网络事件Selector引擎的管理对Selector引擎负载均衡 IoAdaptor -- 网络事件的处理服务器与客户端共用负责读写消息分包组包等 Session -- 代表网络链接可以读写消息 实际的应用我们几乎只需要做IoAdaptor的个性化实现就能完成高效的网络通讯服务今天我们将举例说明如何个性化这个IoAdaptor。 我们今天要完成的目标是实现MySQL服务器的透明代理。效果是你访问代理服务器跟访问目标MySQL无差异。 我们在测试环境10.17.2.30:3306 这台机器上提供了MySql在我们本地机器上跑起来我们今天基于zbus.NET实现的一个代理程序就能达到下面的效果。 完成大概不到100 行的代码, CoolLet’s roll! 首先我们思考透明TCP代理到底在干啥透明的TCP代理的业务逻辑其实非常简单可以描述为将来自代理上游发起请求到代理的数据转发到目标TCP服务器把目标服务器回来的数据原路返回代理上游客户端。 注意这个原路如何做到原路返回成为关键点。这个示例其实跟MySQL没有任何关系原则上任何TCP层面的服务都应该适配。基于zbus.NET怎么来将上面的逻辑在体现出来也就是如何个性化IoAdaptor直观的讲我们要处理的几个事件应该包括1从上游客户端发起的链接请求--代理服务器的Accept事件2代理服务器连接目标服务器的Connect事件3上下游的数据事件onMessage。 zbus.NET的IoAdaptor提供的个性化事件如下 基本包括一个链接客户端或者服务端的生命周期与消息的编解码。 我们的代理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;}} 第二步代理服务接入 Overrideprotected 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事件逻辑处理完毕。 第三步链接成功事件第二步中需要链接到目标服务器 Overridepublic 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都启动感兴趣消息读事件写事件是在读取处理中自动触发为什么在这里做的原因是一定要等上下游都正常态后才启动双方消息处理不然会出现字节丢失。 第四步处理上下游数据事件 Overrideprotected void onMessage(Object msg, Session sess) throws IOException { Session chain sess.chain;if(chain null){sess.asyncClose(); return;} chain.write(msg); } 是不是非常简单类似pipeline从一端的数据写到另外一端。 原则上面4步结束整个透明代理就完成了但是为了处理链接异常清理我们增加了Session清理处理如下 Overridepublic 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你可以规模压力测。看看你是否在本地代理出来了你的目标服务MySQLgl,hf, gogogo. 完整代码可运行代码如下,也可直接到zbus示例代码库中找到 https://git.oschina.net/rushmore/zbus/blob/master/src/test/java/org/zbus/net/TcpProxyAdaptor.java?dir0filepathsrc%2Ftest%2Fjava%2Forg%2Fzbus%2Fnet%2FTcpProxyAdaptor.javaoid08abff381d93519485e1c0ee2c35f1d4f8d1814cshaa29272ed99a8f21ec19a14b403ebee53a385e9a4 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;}}Overrideprotected 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);}Overridepublic 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);}}Overrideprotected void onMessage(Object msg, Session sess) throws IOException { Session chain sess.chain;if(chain null){sess.asyncClose(); return;} chain.write(msg); }Overridepublic 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();}
} 文章转载自 开源中国社区[https://www.oschina.net]
