Netty中，通讯的双方建立连接后，会把数据按照ByteBuf的方式进行传输，例如http协议中，就是通过HttpRequestDecoder对ByteBuf数据流进行处理，转换成http的对象。基于这个思路，我自定义一种通讯协议：Server和客户端直接传输java对象。 实现的原理是通过Encoder把java对象转换成ByteBuf流进行传输，通过Decoder把ByteBuf转换成java对象进行处理。 参看：http://blog.csdn.net/u013252773/article/details/21608951
Netty中，通讯的双方建立连接后，会把数据按照ByteBuf的方式进行传输，例如http协议中，就是通过HttpRequestDecoder对ByteBuf数据流进行处理，转换成http的对象。基于这个思路，我自定义一种通讯协议：Server和客户端直接传输java对象。
实现的原理是通过Encoder把java对象转换成ByteBuf流进行传输，通过Decoder把ByteBuf转换成java对象进行处理，处理逻辑如下图所示：
使用的jar包：
使用的log4j.xml文件：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | <?xmlversion="1.0"?>
<!DOCTYPElog4j:configuration SYSTEM "log4j.dtd">
<log4j:configurationxmlns:log4j="http://jakarta.apache.org/log4j/">
<appendername="CONSOLE"class="org.apache.log4j.ConsoleAppender">
<layoutclass="org.apache.log4j.PatternLayout">
<paramname="ConversionPattern"value="[%-5p] [%d] [%t] [%c] %m%n"/>
</layout>
</appender>
<appendername="FILE"class="org.apache.log4j.DailyRollingFileAppender">
<paramname="File"value="./log/netty.log"/>
<layoutclass="org.apache.log4j.PatternLayout">
<paramname="ConversionPattern"value="[%-5p] [%d] [%t] [%c] %m%n"/>
</layout>
</appender>
<appendername="FILE_ERR"class="org.apache.log4j.DailyRollingFileAppender">
<paramname="File"value="./log/netty_err.log"/>
<paramname="Threshold"value="ERROR"/>
<layoutclass="org.apache.log4j.PatternLayout">
<paramname="ConversionPattern"value="[%-5p] [%d] [%t] [%c] %m%n"/>
</layout>
</appender>
<loggername="io.netty"additivity="false">
<levelvalue="INFO,DEBUG"/>
<appender-refref="FILE"/>
<appender-refref="FILE_ERR"/>
<appender-refref="CONSOLE"/>
</logger>
<loggername="com.yao"additivity="false">
<levelvalue="INFO,DEBUG"/>
<appender-refref="FILE"/>
<appender-refref="FILE_ERR"/>
<appender-refref="CONSOLE"/>
</logger>
<root>
<levelvalue="debug"/>
<appender-refref="FILE"/>
<appender-refref="CONSOLE"/>
<appender-refref="FILE_ERR"/>
</root>
</log4j:configuration>
|
传输的java bean为Person：
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | packagecom.yao.nettyobject;
importjava.io.Serializable;
// 必须实现Serializable接口
publicclass Person implementsSerializable{
privatestatic final long serialVersionUID = 1L;
privateString name;
privateString sex;
privateint age;
publicString toString() {
return"name:" + name + " sex:" + sex + " age:" + age;
}
publicString getName() {
returnname;
}
publicvoid setName(String name) {
this.name = name;
}
publicString getSex() {
returnsex;
}
publicvoid setSex(String sex) {
this.sex = sex;
}
publicint getAge() {
returnage;
}
publicvoid setAge(intage) {
this.age = age;
}
}
|
Server端类：Server PersonDecoder BusinessHandler
1、Server：启动netty服务
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 | packagecom.yao.nettyobject;
importio.netty.bootstrap.ServerBootstrap;
importio.netty.channel.ChannelFuture;
importio.netty.channel.ChannelInitializer;
importio.netty.channel.ChannelOption;
importio.netty.channel.EventLoopGroup;
importio.netty.channel.nio.NioEventLoopGroup;
importio.netty.channel.socket.SocketChannel;
importio.netty.channel.socket.nio.NioServerSocketChannel;
publicclass Server {
publicvoid start(intport) throwsException {
EventLoopGroup bossGroup = newNioEventLoopGroup();
EventLoopGroup workerGroup = newNioEventLoopGroup();
try{
ServerBootstrap b = newServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(newChannelInitializer<SocketChannel>() {
@Override
publicvoid initChannel(SocketChannel ch) throwsException {
//解码
ch.pipeline().addLast(newPersonDecoder());
//业务处理
ch.pipeline().addLast(newBusinessHandler());
}
}).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();
}
}
publicstatic void main(String[] args) throwsException {
Server server = newServer();
server.start(8000);
}
}
|
2、PersonDecoder：把ByteBuf流转换成Person对象，其中ByteBufToBytes是读取ButeBuf的工具类，上一篇文章中提到过，在此不在详述。ByteObjConverter是byte和obj的互相转换的工具。
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | packagecom.yao.nettyobject;
importio.netty.buffer.ByteBuf;
importio.netty.channel.ChannelHandlerContext;
importio.netty.handler.codec.ByteToMessageDecoder;
importjava.util.List;
publicclass PersonDecoder extendsByteToMessageDecoder {
@Override
protectedvoid decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throwsException {
ByteBufToBytes read = newByteBufToBytes();
Object obj = ByteObjConverter.byteToObject(read.read(in));
out.add(obj);
}
}
|
3、BusinessHandler 读取Person信息，并打印
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 | packagecom.yao.nettyobject;
importio.netty.channel.ChannelHandlerContext;
importio.netty.channel.ChannelInboundHandlerAdapter;
importorg.apache.commons.logging.Log;
importorg.apache.commons.logging.LogFactory;
publicclass BusinessHandler extendsChannelInboundHandlerAdapter {
privateLog logger = LogFactory.getLog(BusinessHandler.class);
@Override
publicvoid channelRead(ChannelHandlerContext ctx, Object msg) throwsException {
Person person = (Person) msg;
logger.info("BusinessHandler read msg from client :" + person);
}
@Override
publicvoid channelReadComplete(ChannelHandlerContext ctx) throwsException {
ctx.flush();
}
@Override
publicvoid exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throwsException {
}
}
|
Client端的类：Client ClientInitHandler PersonEncoder
1、Client 建立与Server的连接
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 | packagecom.yao.nettyobject;
importio.netty.bootstrap.Bootstrap;
importio.netty.channel.ChannelFuture;
importio.netty.channel.ChannelInitializer;
importio.netty.channel.ChannelOption;
importio.netty.channel.EventLoopGroup;
importio.netty.channel.nio.NioEventLoopGroup;
importio.netty.channel.socket.SocketChannel;
importio.netty.channel.socket.nio.NioSocketChannel;
publicclass Client {
publicvoid connect(String host, intport) throwsException {
EventLoopGroup workerGroup = newNioEventLoopGroup();
try{
Bootstrap b = newBootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE,true);
b.handler(newChannelInitializer<SocketChannel>() {
@Override
publicvoid initChannel(SocketChannel ch) throwsException {
//编码
ch.pipeline().addLast(newPersonEncoder());
//
ch.pipeline().addLast(newClientInitHandler());
}
});
ChannelFuture f = b.connect(host, port).sync();
f.channel().closeFuture().sync();
}finally{
workerGroup.shutdownGracefully();
}
}
publicstatic void main(String[] args) throwsException {
Client client = newClient();
client.connect("127.0.0.1",8000);
}
}
|
2、ClientInitHandler 向Server发送Person对象
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | packagecom.yao.nettyobject;
importio.netty.channel.ChannelHandlerContext;
importio.netty.channel.ChannelInboundHandlerAdapter;
importorg.apache.commons.logging.Log;
importorg.apache.commons.logging.LogFactory;
publicclass ClientInitHandler extendsChannelInboundHandlerAdapter {
privatestatic Log logger = LogFactory.getLog(ClientInitHandler.class);
@Override
publicvoid channelActive(ChannelHandlerContext ctx) throwsException {
logger.info("HelloClientIntHandler.channelActive");
Person person = newPerson();
person.setName("yaokj");
person.setSex("man");
person.setAge(30);
ctx.write(person);
ctx.flush();
}
}
|
3、PersonEncoder 把Person对象转换成ByteBuf进行传送
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | packagecom.yao.nettyobject;
importio.netty.buffer.ByteBuf;
importio.netty.channel.ChannelHandlerContext;
importio.netty.handler.codec.MessageToByteEncoder;
publicclass PersonEncoder extendsMessageToByteEncoder<Person> {
@Override
protectedvoid encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throwsException {
byte[] datas = ByteObjConverter.objectToByte(msg);
out.writeBytes(datas);
ctx.flush();
}
}
|
工具类：ByteObjConverter
?
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | packagecom.yao.nettyobject;
importjava.io.ByteArrayInputStream;
importjava.io.ByteArrayOutputStream;
importjava.io.IOException;
importjava.io.ObjectInputStream;
importjava.io.ObjectOutputStream;
publicclass ByteObjConverter {
publicstatic Object byteToObject(byte[] bytes) {
Object obj = null;
ByteArrayInputStream bi = newByteArrayInputStream(bytes);
ObjectInputStream oi = null;
try{
oi = newObjectInputStream(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();
}
}
returnobj;
}
publicstatic byte[] objectToByte(Object obj) {
byte[] bytes = null;
ByteArrayOutputStream bo = newByteArrayOutputStream();
ObjectOutputStream oo = null;
try{
oo = newObjectOutputStream(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();
}
}
returnbytes;
}
}
|
工具类：ByteBufToBytes
?
| 1 2 3 4 5 6 7 8 9 10 11 12 | packagecom.yao.nettyobject;
importio.netty.buffer.ByteBuf;
publicclass ByteBufToBytes {
publicbyte[] read(ByteBuf datas) {
byte[] bytes = newbyte[datas.readableBytes()];
datas.readBytes(bytes);
returnbytes;
}
}
|
通过上述代码，实现了Server端与Client端直接使用person对象进行通信的目的。基于此，可以构建更为复杂的场景：Server端同时支撑多种协议，不同的协议采用不同的Decoder进行解析，解析结果保持统一，这样业务处理类可以保持接口一致。下一节将编写这样一个案例。
本例中需要注意的事项是：
1、Person对象必须实现Serializable接口，否则不能进行序列化。
2、PersonDecoder读取ByteBuf数据的时候，并没有对多次流式数据进行处理，而是简单的一次性接收，如果数据量大的情况下，可能会出现数据不完整，这个问题会在后续的学习中解决。
