MessagePack在Netty中的应用
前面使用Netty通信时，传输的都是字符串对象，因为在进行远程过程调用时，更多的是传输pojo对象，这时就需要对pojo对象进行序列化与反序列化（编码与解码），因为Java序列化技术本身的局限性，所以往往会使用第三方的编解码框架，如这里使用的MessagePack。
在使用MessagePack时，需要注意下面两点：
- MessagePack编码后的结果是一个List对象；
- 传输的pojo对象一定要加上@Message注解，否则无法使用MessagePack进行编码；
上面两点确实非常重要，我第一次在Netty中使用MessagePack，因为没有注意上面两点，写的Netty程序运行没有报错，客户端连接服务端也没有问题，但就是不能输出传输的pojo对象，原因就是上面的这两个问题，所以一定要先知道这两点原理，否则后面在测试Netty程序时会有很多问题，并且排错debug过程也不太容易。
下面就直接给出demo的代码，因为在代码中都加了很多注释，所以这里不再详细进行说明。
编码器与解码器
MsgpackEncoder.java
package cn.xpleaf.msgpack;import org.msgpack.MessagePack;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.MessageToByteEncoder;/** * MsgpackEncoder继承自Netty中的MessageToByteEncoder类， * 并重写抽象方法encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) * 它负责将Object类型的POJO对象编码为byte数组，然后写入到ByteBuf中 * @author yeyonghao * */public class MsgpackEncoder extends MessageToByteEncoder<Object> { ???@Override ???protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { ???????// 创建MessagePack对象 ???????MessagePack msgpack = new MessagePack(); ???????// 将对象编码为MessagePack格式的字节数组 ???????byte[] raw = msgpack.write(msg); ???????// 将字节数组写入到ByteBuf中 ???????out.writeBytes(raw); ???}}MsgpackDecoder
package cn.xpleaf.msgpack;import java.util.List;import org.msgpack.MessagePack;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.ByteToMessageDecoder;import io.netty.handler.codec.MessageToMessageDecoder;/** * MsgpackDecoder继承自Netty中的MessageToMessageDecoder类， * 并重写抽象方法decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) * 首先从数据报msg（数据类型取决于继承MessageToMessageDecoder时填写的泛型类型）中获取需要解码的byte数组 * 然后调用MessagePack的read方法将其反序列化（解码）为Object对象 * 将解码后的对象加入到解码列表out中，这样就完成了MessagePack的解码操作 * @author yeyonghao * */public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf> { ???@Override ???protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception { ???????// 从数据报msg中（这里的数据类型为ByteBuf，因为Netty的通信基于ByteBuf对象） ???????final byte[] array; ???????final int length = msg.readableBytes(); ???????array = new byte[length]; ???????/** ????????* 这里使用的是ByteBuf的getBytes方法来将ByteBuf对象转换为字节数组，前面是使用readBytes，直接传入一个接收的字节数组参数即可 ????????* 这里的参数比较多，第一个参数是index，关于readerIndex，说明如下： ????????* ByteBuf是通过readerIndex跟writerIndex两个位置指针来协助缓冲区的读写操作的，具体原理等到Netty源码分析时再详细学习一下 ????????* 第二个参数是接收的字节数组 ????????* 第三个参数是dstIndex the first index of the destination ????????* 第四个参数是length ??the number of bytes to transfer ????????*/ ???????msg.getBytes(msg.readerIndex(), array, 0, length); ???????// 创建一个MessagePack对象 ???????MessagePack msgpack = new MessagePack(); ???????// 解码并添加到解码列表out中 ???????out.add(msgpack.read(array)); ???}}服务端
PojoServer.java
package cn.xpleaf.basic;import cn.xpleaf.msgpack.MsgpackDecoder;import cn.xpleaf.msgpack.MsgpackEncoder;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 PojoServer { ???public void bind(int port) throws Exception { ???????EventLoopGroup bossGroup = new NioEventLoopGroup(); ???????EventLoopGroup workerGroup = new NioEventLoopGroup(); ???????try { ???????????ServerBootstrap b = new ServerBootstrap(); ???????????b.group(bossGroup, workerGroup) ???????????????.channel(NioServerSocketChannel.class) ???????????????.option(ChannelOption.SO_BACKLOG, 1024) ???????????????.childHandler(new ChannelInitializer<SocketChannel>() { ???????????????????@Override ???????????????????protected void initChannel(SocketChannel ch) throws Exception { ???????????????????????// 添加MesspagePack解码器 ???????????????????????ch.pipeline().addLast("msgpack decoder", new MsgpackDecoder()); ???????????????????????// 添加MessagePack编码器 ???????????????????????ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder()); ???????????????????????// 添加业务处理handler ???????????????????????ch.pipeline().addLast(new PojoServerHandler()); ???????????????????} ???????????????}); ???????????// 绑定端口，同步等待成功，该方法是同步阻塞的，绑定成功后返回一个ChannelFuture ???????????ChannelFuture f = b.bind(port).sync(); ???????????// 等待服务端监听端口关闭，阻塞，等待服务端链路关闭之后main函数才退出 ???????????f.channel().closeFuture().sync(); ???????} finally { ???????????// 优雅退出，释放线程池资源 ???????????bossGroup.shutdownGracefully(); ???????????workerGroup.shutdownGracefully(); ???????} ???} ???public static void main(String[] args) throws Exception { ???????int port = 8080; ???????if(args != null && args.length > 0) { ???????????try { ???????????????port = Integer.valueOf(port); ???????????} catch (NumberFormatException e) { ???????????????// TODO: handle exception ???????????} ???????} ???????new PojoServer().bind(port); ???}}PojoServerHandler.java
package cn.xpleaf.basic;import java.util.List;import cn.xpleaf.pojo.User;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;public class PojoServerHandler extends ChannelInboundHandlerAdapter { ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????// 注意msg为List，而不是User类型，这点尤其需要注意 ???????// 否则程序人执行，不会报错，但没有任何输出 ???????@SuppressWarnings("unchecked") ???????List<Object> list = (List<Object>) msg; ???????System.out.println("Pojo from client : " + list); ???????// 遍历List，输出的是pojo对象中的属性 ???????for (Object obj : list) { ???????????System.out.println(obj); ???????} ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???????ctx.flush(); ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ???????ctx.close(); ???}}客户端
PojoClient.java
package cn.xpleaf.basic;import cn.xpleaf.msgpack.MsgpackDecoder;import cn.xpleaf.msgpack.MsgpackEncoder;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 PojoClient { ???public void connect(int port, String host) throws Exception { ???????// 配置客户端NIO线程组 ???????EventLoopGroup group = new NioEventLoopGroup(); ???????try { ???????????Bootstrap b = new Bootstrap(); ???????????b.group(group).channel(NioSocketChannel.class) ???????????????.option(ChannelOption.TCP_NODELAY, true) ???????????????// 设置TCP连接超时时间 ???????????????.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) ???????????????.handler(new ChannelInitializer<SocketChannel>() { ???????????????????@Override ???????????????????protected void initChannel(SocketChannel ch) throws Exception { ???????????????????????// 添加MesspagePack解码器 ???????????????????????ch.pipeline().addLast("msgpack decoder", new MsgpackDecoder()); ???????????????????????// 添加MessagePack编码器 ???????????????????????ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder()); ???????????????????????// 添加业务处理handler ???????????????????????ch.pipeline().addLast(new PojoClientHandler()); ???????????????????} ???????????????}); ???????????// 发起异步连接操作（注意服务端是bind，客户端则需要connect） ???????????ChannelFuture f = b.connect(host, port).sync(); ???????????// 等待客户端链路关闭 ???????????f.channel().closeFuture().sync(); ???????} finally { ???????????// 优雅退出，释放NIO线程组 ???????????group.shutdownGracefully(); ???????} ???} ???public static void main(String[] args) throws Exception { ???????int port = 8080; ???????if(args != null && args.length > 0) { ???????????try { ???????????????port = Integer.valueOf(port); ???????????} catch (NumberFormatException e) { ???????????????// 采用默认值 ???????????} ???????} ???????new PojoClient().connect(port, "localhost"); ???}}PojoClientHandler.java
package cn.xpleaf.basic;import cn.xpleaf.pojo.User;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;public class PojoClientHandler extends ChannelInboundHandlerAdapter { ???@Override ???public void channelActive(ChannelHandlerContext ctx) { ???????User user = new User(); ???????user.setName("client"); ???????user.setAge(10);// ?????for(int i = 0; i < 10; i++) {// ?????????ctx.write(user);// ?????}// ?????ctx.flush(); ???????ctx.writeAndFlush(user); ???} ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???????ctx.flush(); ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????ctx.close(); ???}}POJO
User.java
package cn.xpleaf.pojo;import org.msgpack.annotation.Message;@Messagepublic class User { ???private String name; ???private int age; ???public String getName() { ???????return name; ???} ???public void setName(String name) { ???????this.name = name; ???} ???public int getAge() { ???????return age; ???} ???public void setAge(int age) { ???????this.age = age; ???} ???@Override ???public String toString() { ???????return "User [name=" + name + ", age=" + age + "]"; ???}}测试
运行服务端，再运行客户端，服务端的输出结果如下：
Pojo from client : ["client",10]"client"10MessagePack在Netty中的应用
原文地址：http://blog.51cto.com/xpleaf/2071686