高性能NIO框架Netty-对象传输

http://cxytiandi.com/blog/detail/17403

上篇文章高性能NIO框架Netty入门篇我们对Netty做了一个简单的介绍，并且写了一个入门的Demo,客户端往服务端发送一个字符串的消息，服务端回复一个字符串的消息，今天我们来学习下在Netty中怎么使用对象来传输数据。

上篇文章中传输字符串我们用的是框架自带的StringEncoder，StringDecoder编解码器，现在想要通过对象来传输数据，该怎么弄呢？

既然StringEncoder和StringDecoder可以传输字符串，我们来看看这2个类的源码不就知道它们到底做了一些什么工作。

StringEncoder

1. public class StringEncoder extends MessageToMessageEncoder<CharSequence> {
3. // TODO Use CharsetEncoder instead.
4. private final Charset charset;
6. /**
7. * Creates a new instance with the current system character set.
8. */
9. public StringEncoder() {
10. this(Charset.defaultCharset());
11. }
13. /**
14. * Creates a new instance with the specified character set.
15. */
16. public StringEncoder(Charset charset) {
17. if (charset == null) {
18. throw new NullPointerException("charset");
19. }
20. this.charset = charset;
21. }
23. @Override
24. protected void encode(ChannelHandlerContext ctx, CharSequence msg, List<Object> out) throws Exception {
25. if (msg.length() == 0) {
26. return;
27. }
29. out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset));
30. }
31. }

通过继承MessageToMessageEncoder，重写encode方法来进行编码操作，就是将字符串进行输出即可。

StringDecoder

1. public class StringDecoder extends MessageToMessageDecoder<ByteBuf> {
3. // TODO Use CharsetDecoder instead.
4. private final Charset charset;
6. /**
7. * Creates a new instance with the current system character set.
8. */
9. public StringDecoder() {
10. this(Charset.defaultCharset());
11. }
13. /**
14. * Creates a new instance with the specified character set.
15. */
16. public StringDecoder(Charset charset) {
17. if (charset == null) {
18. throw new NullPointerException("charset");
19. }
20. this.charset = charset;
21. }
23. @Override
24. protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
25. out.add(msg.toString(charset));
26. }
27. }

继承MessageToMessageDecoder，重写decode方法，将ByteBuf数据直接转成字符串进行输出，解码完成。

通过上面的源码分析，我们发现编解码的原理无非就是在数据传输前进行一次处理，接收后进行一次处理，在网络中传输的数据都是字节，我们现在想要传PO对象，那么必然需要进行编码和解码2个步骤，我们可以自定义编解码器来对对象进行序列化，然后通过ByteBuf的形式进行传输, 传输对象需要实现java.io.Serializable接口。

首先我们定义一个传输对象，实现序列化接口，暂时先定义2个字段，一个ID,用来标识客户端，一个内容字段，代码如下：

1. public class Message implements Serializable {
2. private static final long serialVersionUID = -7543514952950971498L;
3. private String id;
4. private String content;
6. public String getId() {
7. return id;
8. }
10. public void setId(String id) {
11. this.id = id;
12. }
14. public String getContent() {
15. return content;
16. }
18. public void setContent(String content) {
19. this.content = content;
20. }
22. }

传输对象定好后，定义对象的编解码器。

对象编码器

将对象序列化成字节，通过ByteBuf形式进行传输，ByteBuf是一个byte存放的缓冲区，提供了读写操作。

1. public class MessageEncoder extends MessageToByteEncoder<Message> {
3. @Override
4. protected void encode(ChannelHandlerContext ctx, Message message, ByteBuf out) throws Exception {
5. byte[] datas = ByteUtils.objectToByte(message);
6. out.writeBytes(datas);
7. ctx.flush();
8. }
10. }

对象解码器

接收ByteBuf数据，将ByteBuf反序列化成对象

1. public class MessageDecoder extends ByteToMessageDecoder {
3. @Override
4. protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
5. Object obj = ByteUtils.byteToObject(ByteUtils.read(in));
6. out.add(obj);
7. }
9. }

将上篇文章中服务端的编解码器改成对象编解码器：

1. public class ImServer {
3. public void run(int port) {
4. EventLoopGroup bossGroup = new NioEventLoopGroup();
5. EventLoopGroup workerGroup = new NioEventLoopGroup();
7. ServerBootstrap bootstrap = new ServerBootstrap();
8. bootstrap.group(bossGroup, workerGroup)
9. .channel(NioServerSocketChannel.class)
10. .childHandler(new ChannelInitializer<SocketChannel>() {
11. @Override
12. public void initChannel(SocketChannel ch) throws Exception {
13. //实体类传输数据，jdk序列化
14. ch.pipeline().addLast("decoder", new MessageDecoder());
15. ch.pipeline().addLast("encoder", new MessageEncoder());
16. ch.pipeline().addLast(new ServerPoHandler());
17. //字符串传输数据
18. /*ch.pipeline().addLast("decoder", new StringDecoder());
19. ch.pipeline().addLast("encoder", new StringEncoder());
20. ch.pipeline().addLast(new ServerStringHandler());*/
21. }
22. })
23. .option(ChannelOption.SO_BACKLOG, 128)
24. .childOption(ChannelOption.SO_KEEPALIVE, true);
26. try {
27. ChannelFuture f = bootstrap.bind(port).sync();
28. f.channel().closeFuture().sync();
29. } catch (InterruptedException e) {
30. e.printStackTrace();
31. } finally {
32. workerGroup.shutdownGracefully();
33. bossGroup.shutdownGracefully();
34. }
35. }
37. }

接下来编写服务端的消息处理类：

1. public class ServerPoHandler extends ChannelInboundHandlerAdapter {
3. @Override
4. public void channelRead(ChannelHandlerContext ctx, Object msg) {
5. Message message = (Message) msg;
6. System.err.println("server:" + message.getId());
7. ctx.writeAndFlush(message);
8. }
10. @Override
11. public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
12. cause.printStackTrace();
13. ctx.close();
14. }
16. }

服务端改造好了之后，就要改造客户端了，同样的道理，客户端和服务端的编解码器都要一致才行。

客户端连接时指定对象编解码器和对象消息处理类，代码如下：

1. public class ImConnection {
3. private Channel channel;
5. public Channel connect(String host, int port) {
6. doConnect(host, port);
7. return this.channel;
8. }
10. private void doConnect(String host, int port) {
11. EventLoopGroup workerGroup = new NioEventLoopGroup();
12. try {
13. Bootstrap b = new Bootstrap();
14. b.group(workerGroup);
15. b.channel(NioSocketChannel.class);
16. b.option(ChannelOption.SO_KEEPALIVE, true);
17. b.handler(new ChannelInitializer<SocketChannel>() {
18. @Override
19. public void initChannel(SocketChannel ch) throws Exception {
20. //实体类传输数据，jdk序列化
21. ch.pipeline().addLast("decoder", new MessageDecoder());
22. ch.pipeline().addLast("encoder", new MessageEncoder());
23. ch.pipeline().addLast(new ClientPoHandler());
25. //字符串传输数据
26. /*ch.pipeline().addLast("decoder", new StringDecoder());
27. ch.pipeline().addLast("encoder", new StringEncoder());
28. ch.pipeline().addLast(new ClientStringHandler());*/
29. }
30. });
32. ChannelFuture f = b.connect(host, port).sync();
33. channel = f.channel();
34. } catch(Exception e) {
35. e.printStackTrace();
36. }
37. }
39. }

客户端消息处理类：

1. /**
2. * 当编解码器为实体对象时时用来接收数据
3. * @author yinjihuan
4. *
5. */
6. public class ClientPoHandler extends ChannelInboundHandlerAdapter {
8. @Override
9. public void channelRead(ChannelHandlerContext ctx, Object msg) {
10. Message message = (Message) msg;
11. System.out.println("client:" + message.getContent());
12. }
14. @Override
15. public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
16. cause.printStackTrace();
17. ctx.close();
18. }
20. }

客户端启动类也需要改造，将发送字符串的消息变成对象消息

1. public class ImClientApp {
2. public static void main(String[] args) {
3. String host = "127.0.0.1";
4. int port = 2222;
5. Channel channel = new ImConnection().connect(host, port);
6. //对象传输数据
7. Message message = new Message();
8. message.setId(UUID.randomUUID().toString().replaceAll("-", ""));
9. message.setContent("hello yinjihuan");
10. channel.writeAndFlush(message);
11. //字符串传输数据
12. //channel.writeAndFlush("yinjihuan");
13. }
14. }

源码参考：https://github.com/yinjihuan/netty-im