zoukankan      html  css  js  c++  java
  • AIO系列文档(2)----TIO使用

    AIO系列文档(1)----图解ByteBuffer中介绍了ByteBuffer用法,下面通过介绍t-io介绍如何使用:

      1. hello world例子简介

        本例子演示的是一个典型的TCP长连接应用,代码位于example/helloworld目录中。

        • 服务端和客户端的消息协议比较简单,消息头为4个字节,用以表示消息体的长度,消息体为一个字符串的byte[]
        • 服务端先启动,监听6789端口
        • 客户端连接到服务端后,会主动向服务器发送一条消息
        • 服务器收到消息后会回应一条消息
        • 之后,框架层会自动从客户端发心跳到服务器,服务器也会检测心跳有没有超时
        • 框架层会在断链后自动重连(对t-io来说,只需多一行代码便拥有自动重连功能
      2. 公共模块代码

        在pom.xml文件中引入tio-core
        <dependency>
            <groupId>org.t-io</groupId>
            <artifactId>tio-core</artifactId>
            <version>2.0.1.v20171015-RELEASE</version>
        </dependency>

        定义Packet(注:有时候服务器和客户端的业务消息包结构不一样,这种情况下,消息包的定义就不要放在公共模块中,而是在服务端和客户端分别定义)

        package org.tio.examples.helloworld.common;
        
        import org.tio.core.intf.Packet;
        
        /**
         * @author tanyaowu
         */
        public class HelloPacket extends Packet {
        	private static final long serialVersionUID = -172060606924066412L;
        	public static final int HEADER_LENGHT = 4;//消息头的长度
        	public static final String CHARSET = "utf-8";
        	private byte[] body;
        
        	/**
        	 * @return the body
        	 */
        	public byte[] getBody() {
        		return body;
        	}
        
        	/**
        	 * @param body the body to set
        	 */
        	public void setBody(byte[] body) {
        		this.body = body;
        	}
        }
        

        定义服务器端和客户端都用得到的常量

        package org.tio.examples.helloworld.common;
        
        /**
         *
         * @author tanyaowu
         * 2017年3月30日 下午7:05:54
         */
        public interface Const {
        	/**
        	 * 服务器地址
        	 */
        	public static final String SERVER = "127.0.0.1";
        	
        	/**
        	 * 监听端口
        	 */
        	public static final int PORT = 6789;
        
        	/**
        	 * 心跳超时时间
        	 */
        	public static final int TIMEOUT = 5000;
        }
        
      3. 服务端代码

        实现org.tio.server.intf.ServerAioHandler

        package org.tio.examples.helloworld.server;
        
        import java.nio.ByteBuffer;
        
        import org.tio.core.Aio;
        import org.tio.core.ChannelContext;
        import org.tio.core.GroupContext;
        import org.tio.core.exception.AioDecodeException;
        import org.tio.core.intf.Packet;
        import org.tio.examples.helloworld.common.HelloPacket;
        import org.tio.server.intf.ServerAioHandler;
        
        /**
         * @author tanyaowu
         */
        public class HelloServerAioHandler implements ServerAioHandler {
        
        	/**
        	 * 解码:把接收到的ByteBuffer,解码成应用可以识别的业务消息包
        	 * 总的消息结构:消息头 + 消息体
        	 * 消息头结构:    4个字节,存储消息体的长度
        	 * 消息体结构:   对象的json串的byte[]
        	 */
        	@Override
        	public HelloPacket decode(ByteBuffer buffer, ChannelContext channelContext) throws AioDecodeException {
        		int readableLength = buffer.limit() - buffer.position();
        		//收到的数据组不了业务包,则返回null以告诉框架数据不够
        		if (readableLength < HelloPacket.HEADER_LENGHT) {
        			return null;
        		}
        
        		//读取消息体的长度
        		int bodyLength = buffer.getInt();
        
        		//数据不正确,则抛出AioDecodeException异常
        		if (bodyLength < 0) {
        			throw new AioDecodeException("bodyLength [" + bodyLength + "] is not right, remote:" + channelContext.getClientNode());
        		}
        
        		//计算本次需要的数据长度
        		int neededLength = HelloPacket.HEADER_LENGHT + bodyLength;
        		//收到的数据是否足够组包
        		int isDataEnough = readableLength - neededLength;
        		// 不够消息体长度(剩下的buffe组不了消息体)
        		if (isDataEnough < 0) {
        			return null;
        		} else //组包成功
        		{
        			HelloPacket imPacket = new HelloPacket();
        			if (bodyLength > 0) {
        				byte[] dst = new byte[bodyLength];
        				buffer.get(dst);
        				imPacket.setBody(dst);
        			}
        			return imPacket;
        		}
        	}
        
        	/**
        	 * 编码:把业务消息包编码为可以发送的ByteBuffer
        	 * 总的消息结构:消息头 + 消息体
        	 * 消息头结构:    4个字节,存储消息体的长度
        	 * 消息体结构:   对象的json串的byte[]
        	 */
        	@Override
        	public ByteBuffer encode(Packet packet, GroupContext groupContext, ChannelContext channelContext) {
        		HelloPacket helloPacket = (HelloPacket) packet;
        		byte[] body = helloPacket.getBody();
        		int bodyLen = 0;
        		if (body != null) {
        			bodyLen = body.length;
        		}
        
        		//bytebuffer的总长度是 = 消息头的长度 + 消息体的长度
        		int allLen = HelloPacket.HEADER_LENGHT + bodyLen;
        		//创建一个新的bytebuffer
        		ByteBuffer buffer = ByteBuffer.allocate(allLen);
        		//设置字节序
        		buffer.order(groupContext.getByteOrder());
        
        		//写入消息头----消息头的内容就是消息体的长度
        		buffer.putInt(bodyLen);
        
        		//写入消息体
        		if (body != null) {
        			buffer.put(body);
        		}
        		return buffer;
        	}
        
        	
        	/**
        	 * 处理消息
        	 */
        	@Override
        	public void handler(Packet packet, ChannelContext channelContext) throws Exception {
        		HelloPacket helloPacket = (HelloPacket) packet;
        		byte[] body = helloPacket.getBody();
        		if (body != null) {
        			String str = new String(body, HelloPacket.CHARSET);
        			System.out.println("收到消息:" + str);
        
        			HelloPacket resppacket = new HelloPacket();
        			resppacket.setBody(("收到了你的消息,你的消息是:" + str).getBytes(HelloPacket.CHARSET));
        			Aio.send(channelContext, resppacket);
        		}
        		return;
        	}
        }
        
        package org.tio.examples.helloworld.server;
        
        import java.io.IOException;
        
        import org.tio.examples.helloworld.common.Const;
        import org.tio.server.AioServer;
        import org.tio.server.ServerGroupContext;
        import org.tio.server.intf.ServerAioHandler;
        import org.tio.server.intf.ServerAioListener;
        
        /**
         *
         * @author tanyaowu
         * 2017年4月4日 下午12:22:58
         */
        public class HelloServerStarter {
        	//handler, 包括编码、解码、消息处理
        	public static ServerAioHandler aioHandler = new HelloServerAioHandler();
        
        	//事件监听器,可以为null,但建议自己实现该接口,可以参考showcase了解些接口
        	public static ServerAioListener aioListener = null;
        
        	//一组连接共用的上下文对象
        	public static ServerGroupContext serverGroupContext = new ServerGroupContext(aioHandler, aioListener);
        
        	//aioServer对象
        	public static AioServer aioServer = new AioServer(serverGroupContext);
        
        	//有时候需要绑定ip,不需要则null
        	public static String serverIp = null;
        
        	//监听的端口
        	public static int serverPort = Const.PORT;
        
        	/**
        	 * 启动程序入口
        	 */
        	public static void main(String[] args) throws IOException {
        		serverGroupContext.setHeartbeatTimeout(org.tio.examples.helloworld.common.Const.TIMEOUT);
        
        		aioServer.start(serverIp, serverPort);
        	}
        }
      4. 客户端代码

        实现org.tio.client.intf.ClientAioHandler

        package org.tio.examples.helloworld.client;
        
        import java.nio.ByteBuffer;
        
        import org.tio.client.intf.ClientAioHandler;
        import org.tio.core.ChannelContext;
        import org.tio.core.GroupContext;
        import org.tio.core.exception.AioDecodeException;
        import org.tio.core.intf.Packet;
        import org.tio.examples.helloworld.common.HelloPacket;
        
        /**
         * 
         * @author tanyaowu
         */
        public class HelloClientAioHandler implements ClientAioHandler {
        	private static HelloPacket heartbeatPacket = new HelloPacket();
        
        
        	/**
        	 * 解码:把接收到的ByteBuffer,解码成应用可以识别的业务消息包
        	 * 总的消息结构:消息头 + 消息体
        	 * 消息头结构:    4个字节,存储消息体的长度
        	 * 消息体结构:   对象的json串的byte[]
        	 */
        	@Override
        	public HelloPacket decode(ByteBuffer buffer, ChannelContext channelContext) throws AioDecodeException {
        		int readableLength = buffer.limit() - buffer.position();
        		//收到的数据组不了业务包,则返回null以告诉框架数据不够
        		if (readableLength < HelloPacket.HEADER_LENGHT) {
        			return null;
        		}
        
        		//读取消息体的长度
        		int bodyLength = buffer.getInt();
        
        		//数据不正确,则抛出AioDecodeException异常
        		if (bodyLength < 0) {
        			throw new AioDecodeException("bodyLength [" + bodyLength + "] is not right, remote:" + channelContext.getClientNode());
        		}
        
        		//计算本次需要的数据长度
        		int neededLength = HelloPacket.HEADER_LENGHT + bodyLength;
        		//收到的数据是否足够组包
        		int isDataEnough = readableLength - neededLength;
        		// 不够消息体长度(剩下的buffe组不了消息体)
        		if (isDataEnough < 0) {
        			return null;
        		} else //组包成功
        		{
        			HelloPacket imPacket = new HelloPacket();
        			if (bodyLength > 0) {
        				byte[] dst = new byte[bodyLength];
        				buffer.get(dst);
        				imPacket.setBody(dst);
        			}
        			return imPacket;
        		}
        	}
        
        	/**
        	 * 编码:把业务消息包编码为可以发送的ByteBuffer
        	 * 总的消息结构:消息头 + 消息体
        	 * 消息头结构:    4个字节,存储消息体的长度
        	 * 消息体结构:   对象的json串的byte[]
        	 */
        	@Override
        	public ByteBuffer encode(Packet packet, GroupContext groupContext, ChannelContext channelContext) {
        		HelloPacket helloPacket = (HelloPacket) packet;
        		byte[] body = helloPacket.getBody();
        		int bodyLen = 0;
        		if (body != null) {
        			bodyLen = body.length;
        		}
        
        		//bytebuffer的总长度是 = 消息头的长度 + 消息体的长度
        		int allLen = HelloPacket.HEADER_LENGHT + bodyLen;
        		//创建一个新的bytebuffer
        		ByteBuffer buffer = ByteBuffer.allocate(allLen);
        		//设置字节序
        		buffer.order(groupContext.getByteOrder());
        
        		//写入消息头----消息头的内容就是消息体的长度
        		buffer.putInt(bodyLen);
        
        		//写入消息体
        		if (body != null) {
        			buffer.put(body);
        		}
        		return buffer;
        	}
        	
        	/**
        	 * 处理消息
        	 */
        	@Override
        	public void handler(Packet packet, ChannelContext channelContext) throws Exception {
        		HelloPacket helloPacket = (HelloPacket) packet;
        		byte[] body = helloPacket.getBody();
        		if (body != null) {
        			String str = new String(body, HelloPacket.CHARSET);
        			System.out.println("收到消息:" + str);
        		}
        
        		return;
        	}
        
        	/**
        	 * 此方法如果返回null,框架层面则不会发心跳;如果返回非null,框架层面会定时发本方法返回的消息包
        	 */
        	@Override
        	public HelloPacket heartbeatPacket() {
        		return heartbeatPacket;
        	}
        }
        
        package org.tio.examples.helloworld.client;
        
        import org.tio.client.AioClient;
        import org.tio.client.ClientChannelContext;
        import org.tio.client.ClientGroupContext;
        import org.tio.client.ReconnConf;
        import org.tio.client.intf.ClientAioHandler;
        import org.tio.client.intf.ClientAioListener;
        import org.tio.core.Aio;
        import org.tio.core.Node;
        import org.tio.examples.helloworld.common.Const;
        import org.tio.examples.helloworld.common.HelloPacket;
        
        /**
         *
         * @author tanyaowu
         *
         */
        public class HelloClientStarter {
        	//服务器节点
        	public static Node serverNode = new Node(Const.SERVER, Const.PORT);
        
        	//handler, 包括编码、解码、消息处理
        	public static ClientAioHandler aioClientHandler = new HelloClientAioHandler();
        
        	//事件监听器,可以为null,但建议自己实现该接口,可以参考showcase了解些接口
        	public static ClientAioListener aioListener = null;
        
        	//断链后自动连接的,不想自动连接请设为null
        	private static ReconnConf reconnConf = new ReconnConf(5000L);
        
        	//一组连接共用的上下文对象
        	public static ClientGroupContext clientGroupContext = new ClientGroupContext(aioClientHandler, aioListener, reconnConf);
        
        	public static AioClient aioClient = null;
        	public static ClientChannelContext clientChannelContext = null;
        
        	/**
        	 * 启动程序入口
        	 */
        	public static void main(String[] args) throws Exception {
        		clientGroupContext.setHeartbeatTimeout(Const.TIMEOUT);
        		aioClient = new AioClient(clientGroupContext);
        		clientChannelContext = aioClient.connect(serverNode);
        		//连上后,发条消息玩玩
        		send();
        	}
        
        	private static void send() throws Exception {
        		HelloPacket packet = new HelloPacket();
        		packet.setBody("hello world".getBytes(HelloPacket.CHARSET));
        		Aio.send(clientChannelContext, packet);
        	}
        }
        
      5. 运行程序

        运行服务器:org.tio.examples.helloworld.server.HelloServerStarter.main(String[])
        运行客户端:org.tio.examples.helloworld.client.HelloClientStarter.main(String[])
      6. 下一步

        https://gitee.com/tywo45/t-io下载源代码及例子,里面的showcase例子是专门为学习t-io而写的,其设计也是准生产级别的,可以直接拿来做您项目的手脚架。下载完成后,请按下面步骤导入到eclipse中

  • 相关阅读:
    微信多业务
    jps命令
    三白话经典算法系列 Shell排序实现
    无法识别的属性“targetFramework”。请注意,属性名是大写和小写。错误的解决方案
    Timus 1777. Anindilyakwa 奇怪的问题计数
    tokumx经营报表
    Quartz CronTrigger应用
    HDU 3681 BFS&amp;像缩进DP&amp;二分法
    01背包问题
    Web模板
  • 原文地址:https://www.cnblogs.com/barrywxx/p/8546713.html
Copyright © 2011-2022 走看看