本文所使用JDK7和netty5.0为基础。Netty5.0

Netty 服务端开发

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;/*** Created by lianpeng on 2016/12/6.*/
public class NettyService {private static final int PORT = 30088;public static void start() {//事件循环组将网络编程的线程处理部分很好的封装起来EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workGroup = new NioEventLoopGroup();//启动服务端必须实例化ServerBootstrap bootstrap = new ServerBootstrap();//指定使用的事件循环组，该组用于接受连接、接收/发送数据，等等//可以指定两个事件循环组，分别用于接受连接、读写数据bootstrap.group( bossGroup, workGroup )//使用非阻塞的套接字传输，必须指定此服务器通道类型，以监听并处理客户端连接.channel( NioServerSocketChannel.class )//当连接被接受后，会创建NioServerSocketChannel的子通道SocketChannel//childHandler方法用于指定子通道的处理器，通常为ChannelInitializer.handler( new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel( SocketChannel ch ) throws Exception {//管道（Pipeline）持有某个通道的全部处理器ChannelPipeline pipeline = ch.pipeline();//添加一个处理器pipeline.addLast();}} );try {//绑定端口，等待直到成功ChannelFuture f = bootstrap.bind( PORT ).sync();//等待服务器通道被关闭f.channel().closeFuture().sync();} catch ( InterruptedException e ) {e.printStackTrace();} finally {//关闭事件循环，释放相关资源（包括创建的线程）bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}}class nettyServHandle extends ChannelHandlerAdapter {/*** 当从通道读取到数据后，会执行该回调* 注意：数据可能碎片化，分若干次读取，这种情况下，该回调会被执行多次*/public void channelRead( ChannelHandlerContext ctx, Object msg ) {ByteBuf buf = (ByteBuf) msg;//注意这里不一定能收到完整的消息//将接收到的数据写回去，注意这里还没有将数据刷空以发送到对端（peer）//当前方法不使用channelRead0的原因：write可能在channelRead返回前尚未完成（因为异步）//如果使用channelRead0，那么msg对应的ByteBuf将被自动释放（release）ctx.write( msg );}/*** 当读取数据完毕（没有更多数据可读）后，会执行该回调*/public void channelReadComplete( ChannelHandlerContext ctx ) {//刷空所有数据，并在执行完毕后，关闭通道ctx.writeAndFlush( Unpooled.EMPTY_BUFFER ).addListener( ChannelFutureListener.CLOSE );}/*** 当发生任何异常时，执行该回调* 至少应当有一个通道处理器覆盖此方法，以实现必要的异常处理*/public void exceptionCaught( ChannelHandlerContext ctx, Throwable cause ) {//关闭通道ctx.close();}
}

Netty 客户端开发

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;import java.net.InetSocketAddress;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import cc.gmem.study.network.basic.Helper;public class NettyClient
{private static final Logger LOGGER = LoggerFactory.getLogger( NettyClient.class );private final String        host;private final int           port;public NettyClient( String host, int port ){this.host = host;this.port = port;}public void start() throws Exception{EventLoopGroup group = new NioEventLoopGroup();try{//要启动客户端，必须实例化下面这个实例Bootstrap b = new Bootstrap();b//指定事件循环组.group( group )//指定非阻塞的套接字通道类.channel( NioSocketChannel.class )//指定需要连接的服务器.remoteAddress( new InetSocketAddress( host, port ) )//指定通道的处理器，一旦连接成功就会调用此处理器.handler( new ChannelInitializer() {public void initChannel( SocketChannel ch ) throws Exception{//在通道的尾部添加一个处理器ch.pipeline().addLast( new EchoClientHandler() );}} );//连接、等待直到连接成功ChannelFuture f = b.connect().sync();//等待直到客户端通道被关闭f.channel().closeFuture().sync();}finally{//关闭事件循环，释放相关资源（包括创建的线程）group.shutdownGracefully().sync();}}public static void main( String[] args ) throws Exception{new EchoClient( Helper.DEFAULT_HOST, Helper.DEFAULT_PORT ).start();}@Sharablepublic class NettyClientHandler extends SimpleChannelInboundHandler{/*** 一旦与服务器的连接建立，即调用此回调*/public void channelActive( ChannelHandlerContext ctx ){//发送一个简单的消息给服务器//Unpooled是一个工具类，可以分配新的缓冲，或者包装已经存在的字节数组、字节缓冲、字符串//copiedBuffer方法可以根据指定的字符串和编码方式，创建一个大端（big-endian）的字节缓冲//这里的几个汉字和标点共计21字节ByteBuf msg = Unpooled.copiedBuffer( "服务器，你好！", CharsetUtil.UTF_8 );//写入数据并刷空，如果不刷空，数据可能存在于本地缓冲，不发送给服务器ctx.writeAndFlush( msg );}/*** 从服务器接收到数据后，调用此回调。channelRead0会自动的释放（减少引用计数，如果为0则解构并回收）ByteBuf对象* 需要注意的是：接收到的字节可能是碎片化的（channelRead0），虽然上面发送了21字节给服务器* 但是可能分多次读取，例如：第一次读取10字节，第二次读取11字节。因此该回调可能被调用多次* 唯一能保证的是：在使用TCP或其它面向流的协议的情况下，数据被读取的顺序有保证*/public void channelRead0( ChannelHandlerContext ctx, ByteBuf in ){int count = in.readableBytes();//可以获取本次可读的字节数ByteBuf buf = in.readBytes( count ); //读取为字节缓冲Object msg = buf.toString( CharsetUtil.UTF_8 );//转换为字符串形式LOGGER.debug( "Client received: {}", msg );}public void exceptionCaught( ChannelHandlerContext ctx, Throwable cause ){ctx.close();}}
}

Netty 解决tcp粘包和分包

TCP以流的方式进行数据传输，上层的应用程序为了对消息进行区分，往往采样一下此种方式

1.消息长度固定，累计读取到长度总和为LEN的报文后，就认为读取到了一个完整的消息；讲计数器，讲计数器职位重新开始读下一个数据包；

2.将回车符作为消息的结束符，例如：FTP协议，这种方式使用比较广泛

3.将特殊字符作为消息的结束符；

4.通过消息头定义长度字段来标示消息的总长度。

Netty对上面四种应用做了统一的抽象，提供了四种解码器来解决对应问题。

1. StringDecoder     将消息解码成String类型方便读取；

2.LineBasedFrameDecoder    以换行符为结束标记的解码器，工作原理是依次遍历ByteBuf中的可读字节，判读是否是“\n”或“\r\n”;

pipeline.addLast(new LineBasedFrameDecoder( 1024 ) );

3.DelimiterBasedFrameDecoder   以特殊字符为消息结束符的解码器；

ByteBuf delimiter = Unpooled.copiedBuffer( "$_".getBytes() );
//第一个参数表示消息的最大长度，当消息读取到最大长度后任然没找见分割符，就抛出
//TooLongFrameException异常，防止异常码流缺失导致的内存溢出
pipeline.addLast( "framer", new DelimiterBasedFrameDecoder( Integer.MAX_VALUE, Delimiters.nulDelimiter() ) );

4.FixedBasedFrameDecoder   以固定长度进行解码,不管一次接受多少数据包，他都是按设定的长度进行解码；

pipeline.addLast( new FixedBasedFrameDecoder( 1024 ) );