netty基础知识2

 本文所使用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>() {
                     @Override
                     protected 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();
    }
 
    @Sharable
    public 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 ) );