Netty 系列一（核心组件和实例）.

@ChannelHandler.Sharable //标识一个Channel-Handler 可以被多个Channel安全的共享
public class EchoServerHandler extends ChannelHandlerAdapter {


    /**
     * 对于每个传入的消息都要调用
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        System.out.println("Server received：" + in.toString(CharsetUtil.UTF_8));
        //将接收到的消息写给发送者，而不冲刷出站消息
        //ChannelHandlerContext 发送消息。导致消息向下一个ChannelHandler流动
        //Channel 发送消息将会导致消息从 ChannelPipeline的尾端开始流动
        ctx.write(in);
    }

    /**
     * 通知 ChannelHandlerAdapter 最后一次对channel-Read()的调用是当前批量读取中的最后一条消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //暂存于ChannelOutboundBuffer中的消息，在下一次调用flush()或者writeAndFlush()方法时将会尝试写出到套接字
        //将这份暂存消息冲刷到远程节点，并且关闭该Channel
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE);
    }

    /**
     * 在读取操作期间，有异常抛出时会调用
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

}

@ChannelHandler.Sharable //标记该类的实例可以被多个Channel共享
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    /**
     * 当从服务器接收到一条消息时被调用
     *
     * @param ctx
     * @param msg ByteBuf (Netty 的字节容器) 作为一个面向流的协议，TCP 保证了字节数组将会按照服务器发送它们的顺序接收
     * @throws Exception
     */
    @Override
    protected void messageReceived(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("Client" + ctx.channel().remoteAddress() + "connected");
        System.out.println(msg.toString(CharsetUtil.UTF_8));
    }

    /**
     * 在到服务器的连接已经建立之后将被调用
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx)  {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rock!", CharsetUtil.UTF_8));
    }


    /**
     * 在处理过程中引发异常时被调用
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

}

public class NettyServer {

    public static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    private static Integer LISTENER_PORT = PropertiesLoader.getResourcesLoader().getInteger("nettyPort");



    private int port;
    EventLoopGroup boss = null;
    EventLoopGroup worker = null;
    ServerBootstrap serverBootstrap = null;

    public static NettyServer nettyServer = null;

    public static NettyServer getInstance() {
        if (nettyServer == null) {
            synchronized (NettyServer.class) {
                if (nettyServer == null) {
                    nettyServer = new NettyServer(LISTENER_PORT==null?9999:LISTENER_PORT);
                }
            }
        }
        return nettyServer;
    }

    /**
     * 构造函数
     *
     * @param port 端口
     */
    private NettyServer(int port) {
        this.port = port;

    }

    /**
     * 绑定
     *
     * @throws InterruptedException
     */
    public void init() throws InterruptedException {
        try {

            //创建两个线程池
            //目前服务器CPU为单核8线程，调整线程为8
            boss = new NioEventLoopGroup(8);
            worker = new NioEventLoopGroup(8);

            serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(boss, worker);//两个工作线程
            serverBootstrap.channel(NioServerSocketChannel.class);
            //重用缓冲区
            serverBootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            //自动调整下一次缓冲区建立时分配的空间大小，避免内存的浪费
            serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT);
            //当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度,默认值50。
            serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            //用于启用或关于Nagle算法。如果要求高实时性，有数据发送时就马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数减少网络交互，就设置为false等累积一定大小后再发送。默认为false。
            serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
            //是否启用心跳保活机制
            serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            //支持tcp协议
            //bootstrap.childHandler(new TcpChannelInitializer());

            //支持webSocket协议
            serverBootstrap.childHandler(new WebSocketChannelInitializer());
            ChannelFuture f = serverBootstrap.bind(port).sync();
            if (f.isSuccess()) {
                logger.info("netty server start...");
            }
            //等到服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            //优雅释放线程资源
            boss.shutdownGracefully().sync();
            worker.shutdownGracefully().sync();
        }
    }

    /**
     * 销毁netty相关资源
     */
    public void destroy() {
        try {
            if (boss != null) {
                boss.shutdownGracefully();
            }
            if (worker != null) {
                worker.shutdownGracefully();
            }
            if (serverBootstrap != null) {
                serverBootstrap = null;
            }
        } catch (Exception e) {
            logger.error("netty close err:" + e.getMessage(), e);
        }
    }
}