Netty入门实例

通过代码和代码中的详细注释，可以对Netty有个入门的了解

TimeServer(初始化服务，绑定端口，指定IO handler)

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;

/**
 * 服务端
 *
 * @author zhya
 * @date 2018/9/14
 **/
public class TimeServer {

    public static void main(String[] args) {
        try {
            // 启动服务端，绑定8080端口
            new TimeServer().bind(8080);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化服务端
     *
     * @param port
     * @throws InterruptedException
     */
    public void bind(int port) throws InterruptedException {
        // group 是NIO线程组，用于网络事件，实际就是Reactor线程组
        // bossGroup 用于处理连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // bossGroup 用于处理Socket读写操作
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            // 服务初始化工具，封装初始化服务的复杂代码
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 指定group
            serverBootstrap.group(bossGroup, workerGroup)
                    // 设置channel类型
                    .channel(NioServerSocketChannel.class)
                    // 设置缓存
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    // 在创建NioSocketChannel成功后，将他的ChannelHandler(TimeClientHandler)设置到ChannelPipeline中，用于处理网络IO事件
                    .childHandler(new ChildChannelHandler());

            // 绑定端口（同步，阻塞，等待绑定完成）
            ChannelFuture future = serverBootstrap.bind(port).sync();

            // future接收异步通知回调，阻塞，等待链路关闭后退出
            future.channel().closeFuture().sync();
        } finally {
            // 优雅的关闭线程组
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /**
     * IO事件处理器
     *
     * @author zhya
     * @date 2018/9/14
     **/
    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            // 设置处理IO事件的处理器
            socketChannel.pipeline().addLast(new TimeServerHandler());
        }
    }
}

TimeServerHandler(处理服务端Channel中的读写事件)

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import java.time.LocalDateTime;

/**
 * 对服务端网络事件进行读写操作的处理器
 *
 * @author zhya
 * @date 2018/9/14
 **/
public class TimeServerHandler extends ChannelHandlerAdapter {

    /**
     * 读到部分客户端数据后回调
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 将客户端数据转成ByteBuf
        ByteBuf byteBuf = (ByteBuf) msg;
        byte[] req = new byte[byteBuf.readableBytes()];

        // 从ByteBuf中读取客户端发送的数据
        byteBuf.readBytes(req);

        // 展示客户端发送来的数据
        String body = new String(req, "UTF-8");
        System.out.println("request body : " + body);

        String currentTime = LocalDateTime.now().toString();
        // 回写数据给客户端(write只是把待发送消息放到发送缓存数组中)
        ctx.write(Unpooled.copiedBuffer(currentTime.getBytes()));
    }

    /**
     * 完成读客户端数据后回调
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channel read complete");
        // 刷新，将消息队列中的消息发送给SocketChannel
        ctx.flush();
    }

    /**
     * 遇到错误后回调
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("encounter exception");
        System.out.println(cause.getMessage());
        // 发送异常关闭ChannelHandlerContext，释放资源
        ctx.close();
    }
}

TimeClient（初始化客户端SocketChannel，连接ServerSocket，指定IO handler）

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;

/**
 * 客户端
 *
 * @author zhya
 * @date 2018/9/14
 **/
public class TimeClient {
    public static void main(String[] args) {
        try {
            // 构造100个线程，开启客户端连接服务器
            for (int i = 0; i < 100; i++) {
                new Thread(() -> {
                    try {
                        // 开启客户端连接服务器
                        new TimeClient().connect("127.0.0.1", 8080);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化客户端
     *
     * @param host
     * @param port
     * @throws Exception
     */
    public void connect(String host, int port) throws Exception {
        // 负责处理客户端IO事件的NIO线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 客户端初始化工具类，封装初始化客户端的复杂代码
            Bootstrap bootstrap = new Bootstrap();
            // 指定group
            bootstrap.group(group)
                    // 设置channel类型为NioSocketChannel(socket client)
                    .channel(NioSocketChannel.class)
                    // 设置TCP参数，无延时
                    .option(ChannelOption.TCP_NODELAY, true)
                    // 在创建NioSocketChannel成功后，将他的ChannelHandler(TimeClientHandler)设置到ChannelPipeline中，用于处理网络IO事件
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 设置处理IO事件的处理器
                            socketChannel.pipeline().addLast(new TimeClientHandler());
                        }
                    });
            // 阻塞，指导成功建立连接
            ChannelFuture future = bootstrap.connect(host, port).sync();

            // 阻塞，等待异步回调通知
            future.channel().closeFuture().sync();
        } finally {
            // 优雅的关闭线程组，释放资源
            group.shutdownGracefully();
        }
    }
}

TimeClientHandler(处理客户端channel连接、读写事件)

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

/**
 * 处理客户端IO事件的处理器
 *
 * @author zhya
 * @date 2018/9/14
 **/
public class TimeClientHandler extends ChannelHandlerAdapter {

    /**
     * 定义发送数据的载体/容器
     */
    private final ByteBuf firstMsg;

    /**
     * 初始化要发送的数据和ByteBuf
     */
    public TimeClientHandler() {
        byte[] req = "get time".getBytes();
        firstMsg = Unpooled.buffer(req.length);
        firstMsg.writeBytes(req);
    }

    /**
     * 当channel可用的时候回调
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channel active");
        // 向服务端写入并刷新数据
        ctx.writeAndFlush(firstMsg);
    }

    /**
     * 当读取到数据时回调（并非读取到所有的数据）
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 转换
        ByteBuf byteBuf = (ByteBuf) msg;
        byte[] rep = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(rep);

        // 输出
        String body = new String(rep, "UTF-8");
        System.out.println("reponse body : " + body);
    }

    /**
     * 当遇到异常时回调
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("encounter exception");
        System.out.println(cause.getMessage());

        // 停止ChannelHandlerContext，释放资源
        ctx.close();
    }
}