Netty Review - 快速上手篇

package com.artisan.netty.client;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
/**
 * @author 小工匠
 * @version 1.0
 * @desc: 通用handler，处理I/O事件
 * @mark: show me the code , change the world
 */
@ChannelHandler.Sharable
public class HandlerClientHello extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        /**
         * @Description 处理接收到的消息
         **/
        System.out.println("接收到的消息：" + byteBuf.toString(CharsetUtil.UTF_8));
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        /**
         * @Description 处理I/O事件的异常
         **/
        cause.printStackTrace();
        ctx.close();
    }
}

package com.artisan.netty.client;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
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;
/**
 * @author 小工匠
 * @version 1.0
 * @desc: 客户端启动类
 * @mark: show me the code , change the world
 */
public class AppClientHello {
    private final String host;
    private final int port;
    public AppClientHello(String host, int port) {
        this.host = host;
        this.port = port;
    }
    /**
     * 配置相应的参数，提供连接远端的方法
     */
    public void run() throws Exception {
        // I/O线程池
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            // 客户端辅助启动类
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    // 实例化一个Channel
                    .channel(NioSocketChannel.class)
                    // 指定远端地址
                    .remoteAddress(new InetSocketAddress(host, port))
                    // 进行通道初始化配置
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //添加我们自定义的Handler
                            socketChannel.pipeline().addLast(new HandlerClientHello());
                        }
                    });
            // 连接到远端，等待完成
            ChannelFuture future = bootstrap.connect().sync();
            // 发送消息到服务器，指定编码格式为UTF-8
            future.channel().writeAndFlush(Unpooled.copiedBuffer("Hello Artisan", CharsetUtil.UTF_8));
            // 阻塞操作，closeFuture()开启了一个channel的监听器（这期间channel在进行各项工作），直到链路断开
            future.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully().sync();
        }
    }
    public static void main(String[] args) throws Exception {
        new AppClientHello("127.0.0.1", 18080).run();
    }
}

package com.artisan.netty.server;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
/**
 * @author 小工匠
 * @version 1.0
 * @desc: 服务器端I/O处理类
 * @mark: show me the code , change the world
 */
@ChannelHandler.Sharable
public class HandlerServerHello extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 处理接收到的数据，并反馈给客户端
        ByteBuf in = (ByteBuf) msg;
        System.out.println("收到客户端发过来的消息: " + in.toString(CharsetUtil.UTF_8));
        // 写入并发送到远端(客户端)
        ctx.channel().writeAndFlush(Unpooled.copiedBuffer("你好，我是Server，已收到消息", CharsetUtil.UTF_8));
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //出现异常的时候执行的动作（打印并关闭通道）
        cause.printStackTrace();
        ctx.close();
    }
}

package com.artisan.netty.server;
import com.artisan.netty.client.AppClientHello;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
 * @author 小工匠
 * @version 1.0
 * @description: 服务器端启动类
 * @mark: show me the code , change the world
 */
public class AppServerHello {
    private int port;
    public AppServerHello(int port) {
        this.port = port;
    }
    public void run() throws InterruptedException {
        // Netty的Reactor线程池，初始化了一个NioEventLoop数组，用来处理I/O操作,如接受新的连接和读/写数据
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 启动NIO服务
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 参数配置
            serverBootstrap.group(group)
                    //通过工厂方法设计模式实例化一个channel
                    .channel(NioServerSocketChannel.class)
                    //设置监听端口
                    .localAddress(port)
                    //配置childHandler来通知一个关于消息处理的InfoServerHandler实例
                    .childHandler(new ChannelInitializer<SocketChannel>() {  //ChannelInitializer是一个特殊的处理类，他的目的是帮助使用者配置一个新的Channel,用于把许多自定义的处理类增加到pipline上来
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new HandlerServerHello());
                        }
                    });
            // 绑定服务器，该实例将提供有关IO操作的结果或状态的信息
            ChannelFuture future = serverBootstrap.bind(port).sync();
            System.out.println("在" + future.channel().localAddress() + "上开启监听");
            //阻塞操作，closeFuture()开启了一个channel的监听器（这期间channel在进行各项工作），直到链路断开
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            group.shutdownGracefully().sync();//关闭EventLoopGroup并释放所有资源，包括所有创建的线程
        }
    }
    public static void main(String[] args) throws InterruptedException {
        new AppServerHello(18080).run();
    }
}