Netty中,通讯的双方建立连接后,会把数据按照ByteBuf的方式进行传输,例如http协议中,就是通过HttpRequestDecoder对ByteBuf数据流进行处理,转换成http的对象。基于这个思路,我自定义一种通讯协议:Server和客户端直接传输Java对象。
实现的原理是通过Encoder把java对象转换成ByteBuf流进行传输,通过Decoder把ByteBuf转换成java对象进行处理,处理逻辑如下图所示:
传输的Java bean为Person.java
package com.bijian.netty.dto; import java.io.Serializable; //必须实现Serializable接口 public class Person implements Serializable { private static final long serialVersionUID = 1L; private String name; private String sex; private int age; public String toString() { return "name:" + name + " sex:" + sex + " age:" + age; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getSex() { return sex; } public void setSex(String sex) { this.sex = sex; } public int getAge() { return age; } public void setAge(int age) { this.age = age; } }
Server端类:Server.java、PersonDecoder.java、BusinessHandler.java
1.Server.java启动netty服务
package com.bijian.netty.server; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; public class Server { public void start(int port) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new PersonDecoder()); ch.pipeline().addLast(new BusinessHandler()); } }).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true); ChannelFuture f = b.bind(port).sync(); f.channel().closeFuture().sync(); } finally { workerGroup.shutdownGracefully(); bossGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { Server server = new Server(); server.start(8000); } }
2.PersonDecoder.java把ByteBuf流转换成Person对象,其中ByteBufToBytes是读取ButeBuf的工具类,上一篇文章中提到过,在此不在详述。ByteObjConverter是byte和obj的互相转换的工具。
package com.bijian.netty.server; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import java.util.List; import com.bijian.netty.util.ByteBufToBytes; import com.bijian.netty.util.ByteObjConverter; public class PersonDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ByteBufToBytes read = new ByteBufToBytes(); Object obj = ByteObjConverter.ByteToObject(read.read(in)); out.add(obj); } }
3.BusinessHandler.java读取Person信息,并打印
package com.bijian.netty.server; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.bijian.netty.dto.Person; public class BusinessHandler extends ChannelInboundHandlerAdapter { private Logger logger = LoggerFactory.getLogger(BusinessHandler.class); @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { Person person = (Person) msg; logger.info("BusinessHandler read msg from client :" + person); ctx.write("ok"); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { } }
Client端的类:Client ClientInitHandler PersonEncoder
1.Client.java建立与Server的连接
package com.bijian.netty.client; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; public class Client { public void connect(String host, int port) throws Exception { EventLoopGroup workerGroup = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(workerGroup); b.channel(NioSocketChannel.class); b.option(ChannelOption.SO_KEEPALIVE, true); b.handler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new PersonEncoder()); ch.pipeline().addLast(new ClientInitHandler()); } }); ChannelFuture f = b.connect(host, port).sync(); f.channel().closeFuture().sync(); } finally { workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { Client client = new Client(); client.connect("127.0.0.1", 8000); } }
2.ClientInitHandler.java向Server发送Person对象
package com.bijian.netty.client; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.bijian.netty.dto.Person; public class ClientInitHandler extends ChannelInboundHandlerAdapter { private static Logger logger = LoggerFactory.getLogger(ClientInitHandler.class); @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { logger.info("ClientInitHandler.channelActive"); Person person = new Person(); person.setName("test"); person.setSex("man"); person.setAge(30); ctx.write(person); ctx.flush(); } }
3.PersonEncoder.java把Person对象转换成ByteBuf进行传送
package com.bijian.netty.client; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; import com.bijian.netty.dto.Person; import com.bijian.netty.util.ByteObjConverter; public class PersonEncoder extends MessageToByteEncoder<Person> { @Override protected void encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throws Exception { byte[] datas = ByteObjConverter.ObjectToByte(msg); out.writeBytes(datas); ctx.flush(); } }
工具类:ByteObjConverter.java
package com.bijian.netty.util; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; public class ByteObjConverter { public static Object ByteToObject(byte[] bytes) { Object obj = null; ByteArrayInputStream bi = new ByteArrayInputStream(bytes); ObjectInputStream oi = null; try { oi = new ObjectInputStream(bi); obj = oi.readObject(); } catch (Exception e) { e.printStackTrace(); } finally { try { bi.close(); } catch (IOException e) { e.printStackTrace(); } try { oi.close(); } catch (IOException e) { e.printStackTrace(); } } return obj; } public static byte[] ObjectToByte(Object obj) { byte[] bytes = null; ByteArrayOutputStream bo = new ByteArrayOutputStream(); ObjectOutputStream oo = null; try { oo = new ObjectOutputStream(bo); oo.writeObject(obj); bytes = bo.toByteArray(); } catch (Exception e) { e.printStackTrace(); } finally { try { bo.close(); } catch (IOException e) { e.printStackTrace(); } try { oo.close(); } catch (IOException e) { e.printStackTrace(); } } return (bytes); } }
工具类:ByteBufToBytes.java
package com.bijian.netty.util; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; public class ByteBufToBytes { private ByteBuf temp; private boolean end = true; public ByteBufToBytes() { } public ByteBufToBytes(int length) { temp = Unpooled.buffer(length); } public void reading(ByteBuf datas) { datas.readBytes(temp, datas.readableBytes()); if (this.temp.writableBytes() != 0) { end = false; } else { end = true; } } public boolean isEnd() { return end; } public byte[] readFull() { if (end) { byte[] contentByte = new byte[this.temp.readableBytes()]; this.temp.readBytes(contentByte); this.temp.release(); return contentByte; } else { return null; } } public byte[] read(ByteBuf datas) { byte[] bytes = new byte[datas.readableBytes()]; datas.readBytes(bytes); return bytes; } }
通过上述代码,实现了Server端与Client端直接使用person对象进行通信的目的。基于此,可以构建更为复杂的场景:Server端同时支撑多种协议,不同的协议采用不同的Decoder进行解析,解析结果保持统一,这样业务处理类可以保持接口一致。下一节将编写这样一个案例。
本例中需要注意的事项是:
1.Person对象必须实现Serializable接口,否则不能进行序列化。
2.PersonDecoder读取ByteBuf数据的时候,并没有对多次流式数据进行处理,而是简单的一次性接收,如果数据量大的情况下,可能会出现数据不完整,这个问题会在后续的学习中解决。
文章来源:http://blog.csdn.net/u013252773/article/details/21608951
相关推荐
NULL 博文链接:https://bijian1013.iteye.com/blog/2340636
NULL 博文链接:https://bijian1013.iteye.com/blog/2340639
NULL 博文链接:https://bijian1013.iteye.com/blog/2340631
NULL 博文链接:https://bijian1013.iteye.com/blog/2340632
NULL 博文链接:https://bijian1013.iteye.com/blog/2340634
Netty学习笔记_Springboot实现自定义协议.docx Netty学习笔记_Springboot实现自定义协议.docx Netty学习笔记_Springboot实现自定义协议.docx
Netty (netty-netty-4.0.56.Final.tar.gz)是一个 NIO 客户端服务器框架,可以快速轻松地开发协议服务器和客户端等网络应用程序。它极大地简化和流线了网络编程,例如 TCP 和 UDP 套接字服务器。 “快速和简单”并...
Netty4.0全部jar包.开发时候只需要倒入总的哪一个netty4.0.jar就行了 后缀为resources.jar的全部是源码。 简单的代码例子在netty-example-resources.jar里面。
NIO socket开发,netty4.0工具包。
Springboot2.0.8集成 netty4 ,使用protobuf作为ping的数据交换,比json更加的小巧,占用数据量更小,可用于任何第三方应用做心跳监控。 已完成功能: - 客户端授权验证(基于protoBuff) - 心跳检测(基于protoBuff) ...
netty4.0 关于buffe
描述文档请看我的个人博客:www.mesoftware.cn
Netty4.0实现http服务,客户端通过http请求,服务器进行业务处理,返回响应,交互采用json格式
Netty4.0.54英文版API文档,与官网中文档内容一致,方便用户在离线环境下,开发Netty
netty4.0.45jar包socket和http工具包,公司最近开发一个保险项目,用到socket,经过公司10个以上项目验证,该版本的netty4.0.45相当稳定,可以在生产环境上使用。
Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序dsf。...netty, 4.0.28, Final, jar包, 含源码
赠送jar包:netty-codec-dns-4.1.65.Final.jar; 赠送原API文档:netty-codec-dns-4.1.65.Final-javadoc.jar; 赠送源代码:netty-codec-dns-4.1.65.Final-sources.jar; 赠送Maven依赖信息文件:netty-codec-dns-...
赠送jar包:netty-resolver-dns-4.1.68.Final.jar; 赠送原API文档:netty-resolver-dns-4.1.68.Final-javadoc.jar; 赠送源代码:netty-resolver-dns-4.1.68.Final-sources.jar; 赠送Maven依赖信息文件:netty-...
包含javadoc,jar 两个文件夹。 doc中是文档资料。 jar是源码及常用例子,均好源码,好使。