Netty–高性能通讯框架

Netty是什么?

  • Netty是一个异步的基于事件驱动的网络应用框架,用以快速开发高性能、高可靠性的网络IO程序
  • Netty主要针对在TCP协议下,面向Clients端的高并发应用,或者Peer-to-Peer场景下的大量数据持续传输的应用
  • Netty本质是一个NIO框架,适用于服务器通讯相关的多种应用场景

Netty的应用场景

  • 分布式服务的远程服务调用RPC框架,比如Dubbo就采用Netty框架做RPC
  • Netty作为高性能的基础通信组件,提供了TCP/UDP、HTTP等协议栈,并且能够定制和开发私有协议栈

在学习Netty之前,我们先来看一下为什么Netty能够被广泛使用。

一、IO模型

什么是I/O模型?

简单理解就是用什么样的通道进行数据的发送和接收,并且很大程序上决定了程序通信的性能。

Java中支持的3种网络编程模型/IO模式

  • BIO同步且阻塞

服务器实现模式为一个连接一个线程,即客户端有连接请求时服务端就需要启动一个线程进行处理。适用于连接数较小且固定的机构,对服务器资源要求比较高,如果这个连接不做任何事情就会造成不必要的线程开销。

  • NIO同步非阻塞

服务器实现模式为一个线程处理多个请求(连接),即客户端发送的连接请求都会注册到多路复用器上,多路复用器轮询到连接有I/O请求就进行处理。选择器Selector来维护连接通道channel。Netty框架基于NIO实现。

  • AIO异步非阻塞

AIO引入异步通道的概念,采用了Proactor模式,简化了编程,有效的请求才启动线程。由操作系统完成后才通知服务端程序启动线程去处理,一般应用于连接数较多且连接时间较长的应用。

二、BIO模型

每次读写请求都会创建一个线程去处理。

高性能通讯框架——Netty

BIO模型

2.1 BIO编程流程

  1. 服务端启动一个ServerSocket
  2. 客户端发送请求后,先咨询服务器是否有线程响应,如果没有则会等待,或者被拒绝
  3. 如果有响应,客户端线程会等待请求结束后,再继续执行

服务端

public class BIOMain {
    public static void main(String[] args) {
        ServerSocket serverSocket = null;
        try {
            serverSocket = new ServerSocket();
            serverSocket.bind(new InetSocketAddress(6666));
            System.out.println("服务器已启动,端口号:6666");
            while (true){
                System.out.println("等待客户端连接...");
                // 等待客户端连接,当没有客户端连接时,会阻塞
                Socket socket = serverSocket.accept();
                System.out.println("客户端:"   socket.getLocalAddress()   "连接成功");
                // 每当有客户端连接进来,就启动一个线程进行处理
                new BioServer(socket).start();
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if(serverSocket !=null) {
                System.out.println("服务器关闭了");
                try {
                    serverSocket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
    }
}

客户端:

public class BioServer extends Thread {

    private Socket socket;

    public BioServer(Socket socket) {
        this.socket = socket;
    }

    @Override
    public void run() {
        try {
            while (true) {
                BufferedInputStream bufferedInputStream =
                        new BufferedInputStream(socket.getInputStream());
                byte[] bytes = new byte[1024];
                System.out.println("等待数据发送...");
                // 当没有数据的时候,这个地方会阻塞
                int read = bufferedInputStream.read(bytes, 0, 1024);
                String result = new String(bytes, 0, read);
                System.out.println(">>> "   result);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

缺点:每来一个连接都会创建一个线程,消耗CPU资源,如果加上线程池也效果不好,因为它在处理连接Accept和Read地方会造成线程阻塞,浪费资源。

三、NIO模型

我们知道BIO模型主要问题就在线程阻塞的地方,因此,NIO引入Selector就解决了线程阻塞的问题。

public class NioServer {

    public static void main(String[] args) {
        try {
            // 1. 创建一个ServerSocketChannel
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

            // 2. 获取绑定端口
            serverSocketChannel.socket().bind(new InetSocketAddress(6666));

            // 3. 设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);

            // 4. 获取Selector
            Selector selector = Selector.open();

            // 5. 将serverSocketChannel注册到selector上, 并且设置selector对客户端Accept事件感兴趣
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            // 6. 循环等待客户端连接
            while (true) {
                // 当没有事件注册到selector时,继续下一次循环
                if (selector.select(1000) == 0) {
                    //System.out.println("当前没有事件发生,继续下一次循环");
                    continue;
                }
                // 获取相关的SelectionKey集合
                Set selectionKeys = selector.selectedKeys();
                Iterator it = selectionKeys.iterator();
                while (it.hasNext()) {
                    SelectionKey selectionKey = it.next();
                    // 基于事件处理的handler
                    handler(selectionKey);
                    it.remove();
                }
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 基于事件处理的,根据key对应的通道发生的事件做相应的处理
     * @param selectionKey
     * @throws IOException
     */
    private static void handler(SelectionKey selectionKey) throws IOException {
        if (selectionKey.isAcceptable()) {  // 如果是OP_ACCEPT事件,则表示有新的客户端连接
            ServerSocketChannel channel = (ServerSocketChannel) selectionKey.channel();
            // 给客户端生成相应的Channel
            SocketChannel socketChannel = channel.accept();
            // 将socketChannel设置为非阻塞
            socketChannel.configureBlocking(false);
            System.out.println("客户端连接成功...生成socketChannel");
            // 将当前的socketChannel注册到selector上, 关注事件:读, 同时给socketChannel关联一个Buffer
            socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(1024));
        } else if (selectionKey.isReadable()) { // 如果是读取事件
            // 通过key反向获取Channel
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
            // 获取该channel关联的buffer
            //ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
            ByteBuffer buffer = ByteBuffer.allocate(512);

            // 把当前channel数据读到buffer里面去
            socketChannel.read(buffer);
            System.out.println("从客户端读取数据:" new String(buffer.array()));

            //
            ByteBuffer buffer1 = ByteBuffer.wrap("hello client".getBytes());
            socketChannel.write(buffer1);
            selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        } else if (selectionKey.isWritable()){ // 如果是写事件
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
            System.out.println("写事件");
            selectionKey.interestOps(SelectionKey.OP_READ);
        }
    }
}

为了方便大家清晰地认识NIO架构,下面以一个总体流程图来展示:

高性能通讯框架——Netty

NIO

说起NIO,就必须要知道其三大核心模块:

NIO三大核心部分

    • Channel通道:客户端与服务端之间的双工连接通道。所以在请求的过程中,客户端与服务端中间的Channel就在不停地执行“连接、询问、断开”的过程。直到数据准备好,再通过Channel传回来。Channel主要有4个类型:FileChannel(从文件读取数据)、DatagramChannel(读写UDP网络协议数据)、SocketChannel(读写TCP网络协议数据)、ServerSocketChannel(可以监听TCP连接)
    • Buffer缓冲区:客户端存放服务端信息的一个缓冲区容器,服务端如果把数据准备好了,就会通过Channel往Buffer缓冲区里面传。Buffer有7个类型:ByteBuffer、CharBuffer、DoubleBuffer、FloatBuffer、IntBuffer、LongBuffer、ShortBuffer。
    • Selector选择器:服务端选择Channel的一个复用器。Selector有两个核心任务:监控数据是否准备好,应答Channel。

NIO工作原理

NIO是面向缓冲区编程的。它是将数据读取到缓冲区中,需要时可在缓冲区前后移动。

NIO工作模式——非阻塞模式

Java NIO的非阻塞模式,使一个线程从某通道发送请求或者读取数据,但是它仅能获得目前可用的数据,如果目前没有数据可用,就什么都不会获取,而不是保持线程阻塞。

NIO特点

一个线程维护一个Selector, Selector维护多个Channel, 当channel有事件时,则该线程进行处理。

BIO和NIO的对比

  • BIO以流的方式处理数据,NIO以块的方式处理数据,块的方式处理数据比流的效率高
  • BIO是阻塞的,而NIO是非阻塞的
  • BIO是基于字节流和字符流进行操作,而NIO是基于channel和buffer进行操作,数据从通道读到缓冲区或者从缓冲区写到通道中,selector用于监听多个通道的事件(比如:连接请求,数据到达等),因此使用单个线程就可以监听多个客户端通道

NIO缺点:

编程复杂,缓冲区Buffer要考虑读写指针切换。而Netty把它封装之后,进行优化并提供了一个易于操作的使用模式和接口,因此Netty就被广泛使用于通信框架。

三、Netty

Netty是一个异步的、基于事件驱动的网络应用框架,它底层封装了NIO。

Netty框架:

高性能通讯框架——Netty

Netty执行流程

NettyNIO服务端和客户端的区别

Netty

NIO

服务端

NioServerSocketChannel

ServerSocketChannel

客户端

NioSocketChannel

SocketChanel

线程模型

基于主从Reactor多线程模型,它维护两个线程池,一个是处理Accept连接,另一个是处理读写事件。

服务端:

@Slf4j
public class TcpServer extends Thread {
    private Integer port;
    public TcpServer(Integer port){
        this.port = port;
    }

    @Override
    public void run() {
        // 根据主机名和端口号创建ip套接字地址(ip地址 端口号)
        InetSocketAddress socketAddress = new InetSocketAddress(port);
        // 主线程组,处理Accept连接事件的线程,这里线程数设置为1即可,netty处理链接事件默认为单线程,过度设置反而浪费cpu资源
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 工作线程,处理hadnler的工作线程,其实也就是处理IO读写,线程数据默认为 CPU 核心数乘以2
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        // 创建ServerBootstrap实例
        ServerBootstrap serverBootstrap = new ServerBootstrap()
                .group(bossGroup, workerGroup) //初始化ServerBootstrap的线程组
                .channel(NioServerSocketChannel.class)  // 设置将要被实例化的ServerChannel类
                .childHandler(new ServerChannelInitializer()) // 初始化ChannelPipeline责任链
                .localAddress(socketAddress)
                .option(ChannelOption.SO_BACKLOG, 1024) //设置队列大小
                .childOption(ChannelOption.SO_KEEPALIVE, true); // 是否启动心跳保活机制

        try {
            // 绑定端口,开始接收进来的连接,异步连接
            ChannelFuture channelFuture = serverBootstrap.bind(socketAddress).sync();
            log.info("TCP服务器开始监听端口:{}", socketAddress.getPort());
            if (channelFuture.isSuccess()) {
                log.info("TCP服务启动成功-------------------");
            }

            // 主线程执行到这里就 wait 子线程结束,子线程才是真正监听和接受请求的,
            // closeFuture()是开启了一个channel的监听器,负责监听channel是否关闭的状态,
            // 如果监听到channel关闭了,子线程才会释放,syncUninterruptibly()让主线程同步等待子线程结果
            channelFuture.channel().closeFuture().sync();
            log.info("TCP服务已关闭");
        } catch (InterruptedException e) {
            log.error("tcp server exception: {}", e.getMessage());
        } finally {
            // 关闭主线程组
            bossGroup.shutdownGracefully();
            // 关闭工作组
            workerGroup.shutdownGracefully();
        }
    }
}

自定义Handler

@Slf4j
public class TCPServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 客户端连接标识
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端已连接:{}", ctx.channel().localAddress().toString());
        // 获取当前客户端的唯一标识
        String uuid = ctx.channel().id().asLongText();
        log.info("当前连接的客户端id:{}", uuid);
        // 将其对应的标识和channel存入到map中
        CLIENT_MAP.put(uuid, ctx.channel());
    }


    /**
     * 读取客户端发送的消息
     * @param ctx
     * @param msg 客户端发送的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 使用netty提供的ByteBuf生成字节Buffer,里面维护一个字节数组,注意不是JDK自带的ByteBuffer
        ByteBuf byteBuf = (ByteBuf) msg;
        // 读取byteBuf
        // 业务处理 
        // 回消息给客户端
        
    }

    /**
     * 客户端断开连接时触发
     * 当客户端主动断开服务端的链接后,这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("断开前,CLIENT_MAP:{}", CLIENT_MAP);
        //当客户端断开连接时,清除map缓存的客户端信息
        CLIENT_MAP.clear();
        log.info(ctx.channel().localAddress().toString()   " 通道不活跃!并且关闭。");
        log.info("断开后,CLIENT_MAP:{}", CLIENT_MAP);
        // 关闭流
        ctx.close();
    }

    /**
     * 发生异常时触发
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("异常情况: {}", cause.toString());
    }
    
    
    /**
     * channelRead方法执行完成后调用,发送消息给客户端
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // writeAndFlush = write   flush:将数据写入缓存,并刷新
        // 需要对发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("收到消息,返回ok!"));
    }
}

客户端:

public class NettyClient {

    public void run(){
        // 一个事件循环组
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            // 客户端启动helper
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new MyClientHandler());
                        }
                    });
            System.out.println("客户端准备就绪,即将连接服务端...");
            // 连接服务端,并返回channelFuture对象,它用来进来异步通知
            // 一般在Socket编程中,等待响应结果都是同步阻塞的,而Netty则不会造成阻塞,因为ChannelFuture是采取类似观察者模式的形式进行获取结果
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1",6666).sync();
            // 对通道关闭进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭netty
            eventLoopGroup.shutdownGracefully();
        }
    }
}

客户端自定义的Handler

public class MyClientHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端已连接..");
        ctx.writeAndFlush(Unpooled.copiedBuffer("msg", CharsetUtil.UTF_8));
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 读取服务器发送的消息
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("接收到服务器:"   ctx.channel().remoteAddress()   "的消息:"   byteBuf.toString(CharsetUtil.UTF_8));
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
    }
}

至此,一个Netty客户端服务器就搭建完成,启动两个服务。

客户端控制台打印结果

客户端准备就绪,即将连接服务端…

客户端已连接..

接收到服务器:/127.0.0.1:6666的消息:服务器收到了你的消息,并给你发送一个ok

服务端控制台打印结果

服务端已经准备就绪

客户端连接地址:/127.0.0.1:6666, 收到的消息:msg

内容出处:,

声明:本网站所收集的部分公开资料来源于互联网,转载的目的在于传递更多信息及用于网络分享,并不代表本站赞同其观点和对其真实性负责,也不构成任何其他建议。如果您发现网站上有侵犯您的知识产权的作品,请与我们取得联系,我们会及时修改或删除。文章链接:http://www.yixao.net/procedure/22051.html

发表评论

登录后才能评论