Netty学习--传输

传输迁移

• 未使用Netty 的阻塞网络编程

public class PlainOioServer {
    public void serve(int port) throws IOException {
        // 将服务器绑定到指定端口
        final ServerSocket socket = new ServerSocket(port);
        try {
            for (;;) {
                final Socket clientSocket = socket.accept(); // 接受连接
                System.out.println("Accepted connection from " + clientSocket);
                new Thread(new Runnable() { // 创建一个新的线程来处理该连接
                    @Override
                    public void run() {
                        OutputStream out;
                        try {
                            out = clientSocket.getOutputStream();
                            out.write("Hi!\r\n".getBytes(Charset.forName("UTF-8"))); // 将消息写给已连接的客户端
                            out.flush();
                            clientSocket.close(); // 关闭连接
                        }
                        catch (IOException e) {
                            e.printStackTrace();
                        }
                        finally {
                            try {
                                clientSocket.close();
                            }
                            catch (IOException ex) {
                                // ignore on close
                            }
                        }
                    }
                }).start();
            }
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }
}

• 未使用Netty 的异步网络编程

public void serve(int port) throws IOException {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        ServerSocket ssocket = serverChannel.socket();
        InetSocketAddress address = new InetSocketAddress(port);
        ssocket.bind(address); // 将服务器绑定到选定的端口
        Selector selector = Selector.open(); // 打开Selector来处理 Channel
        serverChannel.register(selector, SelectionKey.OP_ACCEPT); // 将ServerSocket注册到Selector以接受连接
        final ByteBuffer msg = ByteBuffer.wrap("Hi!\r\n".getBytes());
        for (;;) {
            try {
                selector.select(); // 等待需要处理的新事件；阻塞将一直持续到下一个传入事件
            } catch (IOException ex) {
                ex.printStackTrace();
                // handle exception
                break;
            }
            Set<SelectionKey> readyKeys = selector.selectedKeys(); // 获取所有接收事件的Selection-Key实例
            Iterator<SelectionKey> iterator = readyKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                try {
                    if (key.isAcceptable()) { // 检查事件是否是一个新的已经就绪可以被接受的连接
                        ServerSocketChannel server =
                                (ServerSocketChannel)key.channel();
                        SocketChannel client = server.accept();
                        client.configureBlocking(false);
                        // 接受客户端,并将它注册到选择器
                        client.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ, msg.duplicate());
                        System.out.println("Accepted connection from " + client);
                    }
                    if (key.isWritable()) { // 检查套接字是否已经准备好写数据
                        SocketChannel client =
                                (SocketChannel)key.channel();
                        ByteBuffer buffer =
                                (ByteBuffer)key.attachment();
                        while (buffer.hasRemaining()) {
                            if (client.write(buffer) == 0) { // 将数据写到已连接的客户端
                                break;
                            }
                        }
                        client.close();
                    }
                } catch (IOException ex) {
                    key.cancel();
                    try {
                        key.channel().close();
                    } catch (IOException cex) {
                      // ignore on close
                    }
                }
            }
        }
    }

• 使用Netty 的阻塞网络处理

 public void server(int port) throws Exception {
        final ByteBuf buf = Unpooled.unreleasableBuffer(
                Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
        EventLoopGroup group = new OioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // 创建Server-Bootstrap
            b.group(group)
                    .channel(OioServerSocketChannel.class) // 使用OioEventLoopGroup以允许阻塞模式（旧的I/O）
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 指定Channel-Initializer，对于每个已接受的连接都调用它
                        @Override
                        public void initChannel(SocketChannel ch)
                                throws Exception {
                            ch.pipeline().addLast(
                                    new ChannelInboundHandlerAdapter() {
                                        @Override
                                        public void channelActive(
                                                ChannelHandlerContext ctx)
                                                throws Exception {
                                            ctx.writeAndFlush(buf.duplicate()) // 将消息写到客户端，并添加ChannelFutureListener,以便消息一被写完就关闭连接
                                                    .addListener(
                                                            ChannelFutureListener.CLOSE);
                                        }
                                    });
                        }
                    });
            ChannelFuture f = b.bind().sync(); // 绑定服务器以接受连接
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();// 释放所有的资源
        }
    }

• 使用Netty 的异步网络处理

 public void server(int port) throws Exception {
        final ByteBuf buf = Unpooled.unreleasableBuffer(
                Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
       EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap(); // 创建Server-Bootstrap
            b.group(group).channel(NioServerSocketChannel.class) // 使用OioEventLoopGroup以允许阻塞模式（旧的I/O）
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 指定Channel-Initializer，对于每个已接受的连接都调用它
                        @Override
                        public void initChannel(SocketChannel ch)
                                throws Exception {
                            ch.pipeline().addLast(
                                    new ChannelInboundHandlerAdapter() {
                                        @Override
                                        public void channelActive(
                                                ChannelHandlerContext ctx)
                                                throws Exception {
                                            ctx.writeAndFlush(buf.duplicate()) // 将消息写到客户端，并添加ChannelFutureListener,以便消息一被写完就关闭连接
                                                    .addListener(
                                                            ChannelFutureListener.CLOSE);
                                        }
                                    });
                        }
                    });
            ChannelFuture f = b.bind().sync(); // 绑定服务器以接受连接
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();// 释放所有的资源
        }
    }

通过代码比较,netty基本不受影响(通用性强).

传输API

Channel 接口的层次结构

如图所示，每个Channel 都将会被分配一个ChannelPipeline 和ChannelConfig。ChannelConfig 包含了该Channel 的所有配置设置，并且支持热更新。由于特定的传输可能具有独特的设置，所以它可能会实现一个ChannelConfig 的子类型。Comparable保证每一个channel都是独一无二的.
ChannelPipeline 持有所有将应用于入站和出站数据以及事件的ChannelHandler 实例.

• 将数据从一种格式转换为另一种格式；
• 提供异常的通知；
• 提供Channel 变为活动的或者非活动的通知；
• 提供当Channel 注册到EventLoop 或者从EventLoop 注销时的通知；
• 提供有关用户自定义事件的通知。

channel是线程安全的,多个线程可以共用一个channel.

内置的传输

• NIO | io.netty.channel.socket.nio | 使用java.nio.channels 包作为基础——基于选择器的方式
• Epoll | io.netty.channel.epoll | 由JNI 驱动的epoll()和非阻塞IO。这个传输支持只有在Linux 上可用的多种特性，如SO_REUSEPORT，比NIO 传输更快，而且是完全非阻塞的
• OIO | io.netty.channel.socket.oio | 使用java.net 包作为基础——使用阻塞流
• Local | io.netty.channel.local | 可以在VM 内部通过管道进行通信的本地传输
• Embedded | io.netty.channel.embedded | Embedded 传输，允许使用ChannelHandler 而又不需要一个真正的基于网络的传输。这在测试你的ChannelHandler 实现时非常有用

NIO——非阻塞I/O

选择并处理状态的变化

Epoll—用于Linux 的本地非阻塞传输

流程和NIO一样,失去了NIO的通用性,但运行在linux上更快,只需要将NioEventLoopGroup替换成EpollEventLoopGroup,并且将NioServerSocketChannel.class 替换为EpollServerSocketChannel.class 即可.

OIO—旧的阻塞I/O

建立在java.net 包的阻塞实现之上,不是异步的.Netty利用了SO_TIMEOUT这个Socket标志，它指定了等待一个I/O操作完成的最大毫秒数。如果操作在指定的时间间隔内没有完成，则将会抛出一个SocketTimeout Exception。Netty将捕获这个异常并继续处理循环。在EventLoop下一次运行时，它将再次尝试。这实际上也是类似于Netty这样的异步框架能够支持OIO的唯一方式.

OIO 的处理逻辑

用于JVM 内部通信的Local 传输

用于在同一个JVM 中运行的客户端和服务器程序之间的异步通信.

Embedded 传输

可以将一组ChannelHandler 作为帮助器类嵌入到其他的ChannelHandler 内部。通过这种方式，你将可以扩展一个ChannelHandler 的功能，而又不需要修改其内部代码。