Netty实现自定义简单的编解码器（二）

关于编解码器请参见：http://my.oschina.net/xinxingegeya/blog/282878

主要思路：

* 实现了一个IntegerEncoder编码器和一个IntegerDecoder解码器

* 服务器端发送一个数字，此时客户端解码器会编码为一个字符串，发送到服务器端

* 服务器先用解码器解码字符串为整形，打印出来。然后服务器端通过服务器端编码器编码后向客户端发送数字

* 客户端通过客户端解码器解码后接收数据，打印出来

编码器

package codec2.encoder;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;import java.util.List;public class IntegerEncoder extends MessageToMessageEncoder<Integer> {/*** encode Integer messages to String messages** @param ctx* @param msg* @param out* @throws Exception*/@Overridepublic void encode(ChannelHandlerContext ctx, Integer msg,List<Object> out) throws Exception {System.err.println("IntegerEncoder encode msg is " + msg);//把int转换为字节数组byte[] result = new byte[4];result[0] = (byte) (msg.intValue() >>> 24);//取最高8位放到0下标result[1] = (byte) (msg.intValue() >>> 16);//取次高8为放到1下标result[2] = (byte) (msg.intValue() >>> 8); //取次低8位放到2下标result[3] = (byte) (msg.intValue());      //取最低8位放到3下标ByteBuf encoded = ctx.alloc().buffer(Integer.BYTES);encoded.writeBytes(result);out.add(encoded);}
}

解码器

package codec2.decoder;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;import java.util.List;public class IntegerDecoder extends MessageToMessageDecoder<ByteBuf> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf msg,List<Object> out) throws Exception {System.err.println("IntegerDecoder decode msg is " + msg);out.add(msg.readInt());}
}

服务器端程序

package codec2;import codec2.decoder.IntegerDecoder;
import codec2.encoder.IntegerEncoder;
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;/*** 主要思路：* 实现了一个IntegerEncoder编码器和一个IntegerDecoder解码器* 服务器端发送一个数字，此时客户端解码器会编码为一个字符串，发送到服务器端* 服务器先用解码器解码字符串为整形，打印出来。然后服务器端通过服务器端编码器编码后向客户端发送数字* 客户端通过客户端解码器解码后接收数据，打印出来*/
public class HelloServer {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).option(ChannelOption.SO_BACKLOG, 128).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {// 注册handlerch.pipeline().addLast(new IntegerEncoder(),new IntegerDecoder(),new HelloServerInHandler());}});ChannelFuture f = b.bind(port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {HelloServer server = new HelloServer();server.start(12345);}
}

package codec2;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;// 该handler是InboundHandler类型
public class HelloServerInHandler extends ChannelInboundHandlerAdapter {@Overridepublic boolean isSharable() {System.out.println("==============handler-sharable==============");return super.isSharable();}@Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel-register==============");}@Overridepublic void channelUnregistered(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel-unregister==============");}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel-active==============");}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel-inactive==============");}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("==============channel-read==============");System.out.println("the msg type is " + msg.getClass().getName());Integer integer = (Integer) msg;System.out.println("服务器端接收到的客户端的数字是" + integer);System.out.println("服务器向客户端写入整型数字2001");ctx.writeAndFlush(new Integer(2001));ctx.close();}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel-read-complete==============");ctx.flush();}
}

客户端程序

package codec2;import codec2.decoder.IntegerDecoder;
import codec2.encoder.IntegerEncoder;
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;/*** 主要思路：* 实现了一个IntegerToStringEncoder编码器和一个StringToIntegerDecoder解码器* 服务器端发送一个数字2000，此时客户端解码器会编码为一个字符串，发送到服务器端* 服务器先用解码器解码字符串为整形，打印出来。然后服务器端通过服务器端编码器编码后向客户端发送数字* 客户端通过客户端解码器解码后接收数据，打印出来*/
public class HelloClient {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.AUTO_READ, true);b.handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new IntegerEncoder(),new IntegerDecoder(),new HelloClientIntHandler());}});ChannelFuture f = b.connect(host, port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {HelloClient client = new HelloClient();client.connect("192.168.0.102", 12345);}
}

package codec2;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;//InboundHandler类型
public class HelloClientIntHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRegistered(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel--register==============");}@Overridepublic void channelUnregistered(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel--unregistered==============");}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel--inactive==============");}// 连接成功后，向server发送消息@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("==============channel--active==============");System.out.println("向服务器端写入2000数字");ctx.writeAndFlush(new Integer(2000));}// 接收server端的消息，并打印出来@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("==============channel--read==============");System.out.println("the msg type is " + msg.getClass().getName());Integer result = (Integer) msg;System.out.println("接收到服务器数据，该数字是" + result);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}
}

运行结果：

服务器端

==============handler-sharable==============

==============channel-register==============

==============channel-active==============

IntegerDecoder decode msg is SimpleLeakAwareByteBuf(UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 4, cap: 1024))

IntegerEncoder encode msg is 2001

==============channel-read==============

the msg type is java.lang.Integer

服务器端接收到的客户端的数字是2000

服务器向客户端写入整型数字2001

==============channel-read-complete==============

==============channel-inactive==============

==============channel-unregister==============

客户端

==============channel--register==============

IntegerEncoder encode msg is 2000

==============channel--active==============

向服务器端写入2000数字

==============channel--read==============

the msg type is java.lang.Integer

接收到服务器数据，该数字是2001

==============channel--inactive==============

==============channel--unregistered==============

IntegerDecoder decode msg is UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 4, cap: 1024)

Process finished with exit code 0

后记：

我想到netty中有两个现成的编解码器是StringEncoder和StringDecoder，看代码：

/** Copyright 2012 The Netty Project** The Netty Project licenses this file to you under the Apache License,* version 2.0 (the "License"); you may not use this file except in compliance* with the License. You may obtain a copy of the License at:**   http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the* License for the specific language governing permissions and limitations* under the License.*/
package io.netty.handler.codec.string;import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.util.List;/*** Encodes the requested {@link String} into a {@link ByteBuf}.* A typical setup for a text-based line protocol in a TCP/IP socket would be:* <pre>* {@link ChannelPipeline} pipeline = ...;** // Decoders* pipeline.addLast("frameDecoder", new {@link LineBasedFrameDecoder}(80));* pipeline.addLast("stringDecoder", new {@link StringDecoder}(CharsetUtil.UTF_8));** // Encoder* pipeline.addLast("stringEncoder", new {@link StringEncoder}(CharsetUtil.UTF_8));* </pre>* and then you can use a {@link String} instead of a {@link ByteBuf}* as a message:* <pre>* void channelRead({@link ChannelHandlerContext} ctx, {@link String} msg) {*     ch.write("Did you say '" + msg + "'?\n");* }* </pre>*/
@Sharable
public class StringEncoder extends MessageToMessageEncoder<CharSequence> {// TODO Use CharsetEncoder instead.private final Charset charset;/*** Creates a new instance with the current system character set.*/public StringEncoder() {this(Charset.defaultCharset());}/*** Creates a new instance with the specified character set.*/public StringEncoder(Charset charset) {if (charset == null) {throw new NullPointerException("charset");}this.charset = charset;}@Overrideprotected void encode(ChannelHandlerContext ctx, CharSequence msg, List<Object> out) throws Exception {if (msg.length() == 0) {return;}out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset));}
}

/** Copyright 2012 The Netty Project** The Netty Project licenses this file to you under the Apache License,* version 2.0 (the "License"); you may not use this file except in compliance* with the License. You may obtain a copy of the License at:**   http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the* License for the specific language governing permissions and limitations* under the License.*/
package io.netty.handler.codec.string;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.MessageToMessageDecoder;import java.nio.charset.Charset;
import java.util.List;/*** Decodes a received {@link ByteBuf} into a {@link String}.  Please* note that this decoder must be used with a proper {@link ByteToMessageDecoder}* such as {@link DelimiterBasedFrameDecoder} or {@link LineBasedFrameDecoder}* if you are using a stream-based transport such as TCP/IP.  A typical setup for a* text-based line protocol in a TCP/IP socket would be:* <pre>* {@link ChannelPipeline} pipeline = ...;** // Decoders* pipeline.addLast("frameDecoder", new {@link LineBasedFrameDecoder}(80));* pipeline.addLast("stringDecoder", new {@link StringDecoder}(CharsetUtil.UTF_8));** // Encoder* pipeline.addLast("stringEncoder", new {@link StringEncoder}(CharsetUtil.UTF_8));* </pre>* and then you can use a {@link String} instead of a {@link ByteBuf}* as a message:* <pre>* void channelRead({@link ChannelHandlerContext} ctx, {@link String} msg) {*     ch.write("Did you say '" + msg + "'?\n");* }* </pre>*/
@Sharable
public class StringDecoder extends MessageToMessageDecoder<ByteBuf> {// TODO Use CharsetDecoder instead.private final Charset charset;/*** Creates a new instance with the current system character set.*/public StringDecoder() {this(Charset.defaultCharset());}/*** Creates a new instance with the specified character set.*/public StringDecoder(Charset charset) {if (charset == null) {throw new NullPointerException("charset");}this.charset = charset;}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {out.add(msg.toString(charset));}
}

你看吧，传输String还得有编解码器。

所以就根据这对编解码器实现了上述编解码器。

===========END===========