Netty学习——实战篇3 BIO与NIO零拷贝和 Netty入门实战

1 BIO拷贝

BIOServer.java

public class BIOServer {
    public static void main(String[] args) throws Exception{
        ServerSocket serverSocket = new ServerSocket(8000);
        while(true){
            Socket socket = serverSocket.accept();
            DataInputStream inputStream = new DataInputStream(socket.getInputStream());
            try {
                byte[] bytes = new byte[4096];
                while(true){
                    int read = inputStream.read(bytes, 0, bytes.length);
                    if(-1 == read){
                        break;
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

BIOClient.java

public class BIOClient {
    public static void main(String[] args) throws Exception{
        Socket socket = new Socket("localhost", 8000);
        String fileName = "";
        InputStream inputStream = new FileInputStream(fileName);
        DataOutputStream outputStream = new DataOutputStream(socket.getOutputStream());
        byte[] bytes = new byte[4096];
        long readCount;
        long total = 0L;
        long beginTime = System.currentTimeMillis();
        while((readCount = inputStream.read(bytes)) >= 0){
            total += readCount;
            outputStream.write(bytes);
        }
        System.out.println("发送总字节数:" + total + ",耗时：" + (System.currentTimeMillis() - beginTime));
        outputStream.close();
        socket.close();
        inputStream.close();
    }
}

2 NIO零拷贝

NIOServer.java

public class NIOServer {
    public static void main(String[] args) throws Exception{
        //1 创建InetSocketAddress,并设置端口号 8000
        InetSocketAddress inetSocketAddress = new InetSocketAddress(8000);

        //2 创建ServerSocketChannel
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //3 创建ServerSocket
        ServerSocket serverSocket = serverSocketChannel.socket();
        //4 ServerSocket绑定address
        serverSocket.bind(inetSocketAddress);
        //5 创建Buffer
        ByteBuffer buffer = ByteBuffer.allocate(4096);
        //6 等待客户端连接
        while(true){
            SocketChannel socketChannel = serverSocketChannel.accept();
            int readCount = 0;
            //7 循环读取
            while( -1 != readCount ){
                try {
                    readCount = socketChannel.read(buffer);
                }catch (Exception ex){
                    break;
                    //ex.printStackTrace();
                }
                //8 复位
                buffer.rewind();
            }
        }




    }
}

NIOClient.java

public class NIOClient {
    public static void main(String[] args) throws Exception{
        //1 创建 SocketChannel
        SocketChannel socketChannel = SocketChannel.open();
        String fileName = "";
        //2 socketchannel 连接服务器
        socketChannel.connect(new InetSocketAddress("localhost",8000));
        //3 得到一个文件channel
        FileChannel fileChannel = new FileInputStream(fileName).getChannel();
        long beginTime = System.currentTimeMillis();
        //4 发送
        long transferCount = fileChannel.transferTo(0, fileChannel.size(), socketChannel);
        System.out.println("发送总字节数是:" + transferCount + ",总耗时:" + (System.currentTimeMillis() - beginTime));
        //5 关闭通道
        fileChannel.close();
    }
}

3 Netty入门实战

3.1 NettyServer.java

@Slf4j
public class NettyServer {
    public static void main(String[] args) {
        //1 创建bossGroup和workerGroup
        /*
            说明：
                1、创建两个线程组 bossGroup和workerGroup
                2、bossGroup只处理连接请求，workerGroup负责处理客户端业务
                3、两个都是无限循环
                4、bossGroup和workerGroup含有的子线程个数，默认是CPU 核数 * 2
         */
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //2 创建服务端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //3 使用链式变成来进行设置
            bootstrap.group(bossGroup,workerGroup) //3.1 设置两个线程组
                    .channel(NioServerSocketChannel.class)  //3.2 使用NIOServerSocketChannel作为服务端的通道
                    .option(ChannelOption.SO_BACKLOG,128) //3.3 设置线程队列得到链接个数
                    .childOption(ChannelOption.SO_KEEPALIVE,true) //3.4 设置保持活动连接状态
                    .handler(new ChannelInitializer<SocketChannel>() {   //3.5 创建一个通道初始化对象(匿名对象)
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            System.out.println("客户socketchannel hashcode=" + ch.hashCode());
                            ch.pipeline().addLast(new MyNettyServerHandler());
                        }
                    });

            //4 绑定一个端口并且同步，生成一个ChannelFuture对象，启动服务器
            ChannelFuture channelFuture = bootstrap.bind(8000).sync();
            //5 为channelfuture注册监听器，监听关心的事件
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if(channelFuture.isSuccess()){
                        log.info("监听端口8000成功");
                    }else {
                        log.info("监听端口8000失败");
                    }
                }
            });
            //6 对关闭通道事件进行监听
            channelFuture.channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            //关闭 bossGroup 和 workerGroup
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}

3.2 MyNettyServerHandler.java

@Slf4j
public class MyNettyServerHandler extends ChannelInboundHandlerAdapter {
    //读取实际的数据
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("ctx:{}",ctx);
        log.info("channel:{}",ctx.channel());
        ByteBuf buf = (ByteBuf) msg;
        log.info("向客户端发送的信息是:{}", buf.toString(CharsetUtil.UTF_8));
        log.info("客户端地址是:{}",ctx.channel().remoteAddress());
    }

    //数据读取完毕，将数据写入缓存，并刷新
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端~(>^ω^<)喵1",CharsetUtil.UTF_8));
    }

    //异常处理，一般是关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }
}

3.3 NettyClient.java

@Slf4j
public class NettyClient {
    public static void main(String[] args) throws Exception{
        //1 创建一个事件循环组
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            //2 创建客户端启动对象
            Bootstrap bootstrap = new Bootstrap();
            //3 设置相关参数
            bootstrap.group(eventLoopGroup)  //3.1 设置线程组
                    .channel(SocketChannel.class) //3.2 设置客户端通道的实现类
                    .handler(new ChannelInitializer<SocketChannel>() {  //3.4 设置自定义handler
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new MyNettyClientHandler());
                        }

                    });
            log.info("客户端就绪");
            //4 启动客户端，连接服务器
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8000);
            //5 监听 关闭通道事件
            channelFuture.channel().closeFuture().sync();
        }finally {
            //6 关闭线程组
            eventLoopGroup.shutdownGracefully();
        }
    }
}

3.4 MyNettyClientHandler.java

@Slf4j
public class MyNettyClientHandler extends ChannelInboundHandlerAdapter {

    //当通道就绪就会触发该方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端：{}",ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 服务端: (>^ω^<)喵", CharsetUtil.UTF_8));
    }

    //当通道有读取事件时，会触发该方法
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        log.info("服务器回复的信息是:{}",buf.toString(CharsetUtil.UTF_8));
    }

    //发生异常时会触发该方法,关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.channel().close();
    }
}

分别启动 NettyServer.java 和 NettyClient.java，运行结果如下图：