网络编程Demo：Java的阻塞与非阻塞模式以及Netty

小灰灰 2024-03-16 16:22 65阅读 0赞

## 前言 ##

IO既神秘，双简单

## IO是什么 ##

从表面理解，IO是输入（input）、输出(output)的英文首字母的缩写形式，可以简单理解为计算机的输入与输出，描述计算机的数据流动，如使用键盘输入了一个“hello world”的字符，通过显示器可以直观看到这个字符,这就是一次完整的IO。

## 怎么理解IO ##

### 从计算机架构层面理解IO ###

从计算机架构上来讲，一台有意义且可运行的计算机，通常会包含：CPU、内存、主板、电源、硬盘、外设等，这里的外设是指可以完成人机交互的辅助工具，如键盘、鼠标、显示器、音箱等；外设有很多种，实际上可分两类：输入设备和输出设备；输入设备的作用就是将人类的各种行为转换为电信号传递给计算机；输出设备的作用就是将计算机内存储的电信号经过处理返回，并以人类可理解的方式呈现。通过键盘输入一段字符并保存到硬盘上，或者从磁盘上读取一段字符，显示到显示器上，都会涉及到计算机的核心部件CPU、内存、硬盘等一系列硬件的协作，这种涉及到计算机核心部件（CPU和内存）与其他设备间的数据转移的过程就是IO。

### 从应用程序层面去理解IO ###

对于非计算机专业的人士，能从计算机的架构层面理解计算机的本质就已经可以了；对于程序员来说，IO的理解更底层一些，需要从应用程序层面来理解IO。通过键盘输入一段字符并保存到硬盘上，除了计算机多个核心硬件的协作外，还离不开应用程序的协作，这里的应用程序是指操作系统，以及运行在操作系统内的用户应用程序；应用程序间的基本管理单位是进程； 用户应用程序进程完成输入字符串的保存的过程实际上是两步：IO调用和IO执行；IO的调用是由用户应用程序进程发起的，IO执行是操作系统实际执行的，串在一起就是用户应用程序向操作系统发起IO调用，操作系统调用与底层物理硬件的交互API执行IO操作，最终字符串被保存在磁盘上。

![251e42a28eae4eebb5f04a285f5255cf.png][]

## IO基础理论 ##

要彻底梳理计算机的IO，还要从与IO相关的一些基础概念开始，如操作系统、用户应用程序、进程、线程等。

### 操作系统 ###

现代计算机是由硬件和操作系统组成，硬件包括CPU、内存、主板、电源、硬盘、外设等，操作系统本身就是应用程序，可以与硬件直接交互的系统程序，用户程序是运行在操作系统内部的；操作系统内可以运行很多应用程序，基本管理的单位是进程。其中用户应用程序操作硬件（如往磁盘上写数据），就需要先与操作系统的内核交互，然后再由操作系统内核与硬件交互。

而操作系统具体可以划分为：内核与应用两部分；内核提供进程管理、内存管理、网络等底层功能，封装了与硬件交互的接口，通过系统调用提供给上层应用使用；应用部分运行的是用户应用程序；

![0fa852969414427184d75df594311316.png][]

#### 内核空间和用户空间 ####

为了保证用户应用程序进程不能直接操作系统内核，保证内核的安全，操作系统将虚拟内存空间分为两部分，一部分为内核空间，一部分是用户空间；内核空间是操作系统内核可以访问的区域，独立于普通用户应用程序，是受保护的内存空间；用户空间是普通用户应用程序可以访问的内存区域；

#### 内核态和用户态 ####

操作系统内的任务进程为系统进程，用户应用程序的进程为用户进程，当实际工作的进程运行在内核空间时，它就处于内核态；当进程运行在用户空间时，它就处于用户态。

那什么时候运行在内核空间，什么时候运行在用户空间呢？

举个例子，当我们需要进行IO操作，从硬盘上读取文件或从网卡上读取数据时，用户应用程序先发起IO调用，操作系统进程开始实际执行IO请求，数据会先从外部写入到内核内存，这时实际工作进程是系统进程，处于内核态；在内核内存中数据准备完毕后，数据会再复制到用户内存空间，这时的实际工作进程是用户进程，处于用户态；

如果是向硬盘写入文件 、向网卡写入数据时，内核态和用户态的切换则刚好与读取相反，所谓的上下文切换就是指从内核态到用户态，或从用户态到内核态的切换。

那为什么要有这样的切换呢？

为了保证内核安全，用户应用程序不能直接操作内核空间的数据，需要把内核态的数据拷贝到用户空间才能操作，否则无法进行这样的操作。

![1a4be953ab554e6b9c5eba7a3db08615.png][]

### IO分类 ###

IO从读取数据的来源分为内存IO、 网络IO和磁盘IO三种，通常我们说的IO指的是后两者(因为内存IO的读写速度比网络IO和磁盘IO快的多)。

I/O按照设备来分的话，分为两种：一种是网络I/O，也就是通过网络进行数据的拉取和输出。一种是磁盘I/O，主要是对磁盘进行读写工作。

网络IO：等待网络数据到达网卡→把网卡中的数据读取到内核缓冲区，然后从内核缓冲区复制数据到进程空间。

磁盘IO：把数据从磁盘中读取到内核缓冲区，然后从内核缓冲区复制数据到进程空间。

由于CPU和内存的速度远远高于外部设备（网卡，磁盘等）的速度，所以在IO编程中，存在速度严重不匹配的问题，才会有各种的IO模型来解决这个问题。关于IO模型，有两组容易混淆的概念：同步与异步、阻塞与非阻塞。

### IO的两个重要阶段 ###

在Linux中，对于一次I/O读取或写入的操作，数据并不会直接读取到用户内存空间或者直接写出到外部。通常包括两个不同阶段，以IO读取为例：

1、等待数据准备好，外部数据到达内核空间；

2、从内核向用户内存空间复制数据；

### 同步与异步IO ###

同步与异步IO的描述主体对象是当前用户应用程序进程或线程，发起IO调用后，当前进程或线程是否挂起等待操作系统完成IO执行。

同步IO的意思是指，发起IO调用后，当前进程或线程需要等待操作系统完成IO执行并告知数据已经用户内存空间准备完成，当前进程或线程才能继续向下执行其他指令。

异步IO的意思是指，发起IO调用后，当前进程或线程不需要等待操作系统完成IO执行，可以直接向下执行其他指令，当操作系统完成IO执行后，当前用户应用程序的进程或线程会收到操作系统的数据已经在用户内存空间准备完成的通知。

以一个读取IO操作而言，在操作系统将外部数据写入到用户内存空间期间，用户进程或线程挂起等待操作系统完成IO执行，这种IO执行策略就是同步IO；如果用户进程或线程并不挂起而是继续后续工作，这种IO执行策略就是异步IO。

### 阻塞与非阻塞IO ###

阻塞与非阻塞IO的描述主体对象是当前用户应用程序进程或线程，同步与异步IO强调的是用户当前进程或线程发起IO调用后执行方式，则阻塞与非阻塞强调的是操作系统IO执行用户内存空间数据是否处于就绪状态的处理方式。

## ServerSocket与Socket ##

### 服务端 ###

@Slf4j
    public class MySocketServer {
        private ServerSocket serverSocket;
    
        public MySocketServer(Integer port)  {
            try {
                //声明一个服务端网络套接字对象
                this.serverSocket = new ServerSocket(port);
            } catch (IOException e) {
                log.info("serverSocket构建异常");
                e.printStackTrace();
            }
        }
        public void start() {
            //定义死循环，持续接受来自客户端的网络请求
            while (true) {
                if (this.serverSocket != null) {
                    InputStream inputStream = null;
                    BufferedReader bufferedReader = null;
                    OutputStream outputStream = null;
                    PrintWriter printWriter = null;
                    try {
                        //接受客户端的网络请求后，得到了一个关于这次请求的一个套接字封装对象
                        Socket serverSocket = this.serverSocket.accept();
                        //从socket对象中读取请求报文数据
                        inputStream = serverSocket.getInputStream();
                        bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
                        String body = null;
                        while ((body = bufferedReader.readLine()) != null) {
                            log.info("收到客户端消息：{}", body);
                        }
                        //报文数据读取完毕后，关闭输入流；
                        serverSocket.shutdownInput();
                        //读取到客户端的报文数据后，向客户端侧写入一个网络请求处理成功的标记信息；
                        outputStream = serverSocket.getOutputStream();
                        printWriter = new PrintWriter(outputStream);
                        printWriter.write("success");
                        printWriter.flush();
                        serverSocket.shutdownOutput();
                    } catch (Exception e) {
                        log.info("服务端异常");
                        e.printStackTrace();
                    } finally {
                        if (inputStream != null) {
                            try {
                                inputStream.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (bufferedReader != null) {
                            try {
                                bufferedReader.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (outputStream != null) {
                            try {
                                outputStream.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (printWriter != null) {
                            printWriter.close();
                        }
                    }
                }
            }
        }
        public static void main(String[] args) {
            MySocketServer myServerSocket = new MySocketServer(8080);
            myServerSocket.start();
        }
    }

### 客户端 ###

@Slf4j
    public class MySocketClient {
        private Socket socketClient = null;
    
        public MySocketClient(String host, Integer port) throws IOException {
            //声明一个客户端的网络套接字对象
            socketClient = new Socket("127.0.0.1", 8080);
        }
    
        public void start() throws IOException {
            //向服务端发送一段报文消息
            OutputStream outputStream = this.socketClient.getOutputStream();
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write("hello，I am client ！\r\n");
            printWriter.flush();
            //发送完毕后，关闭输出流
            this.socketClient.shutdownOutput();
            //接收服务端的响应消息
            InputStream inputStream = this.socketClient.getInputStream();
            InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
            BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
            String body = null;
            while ((body = bufferedReader.readLine()) != null) {
                log.info("收到服务端响应消息：{}", body);
            }
            this.socketClient.shutdownInput();
            inputStream.close();
            inputStreamReader.close();
            bufferedReader.close();
            this.socketClient.close();
        }
    
        public static void main(String[] args) throws IOException, InterruptedException {
            MySocketClient mySocketClient = null;
            for (int i = 0; i < 10; i++) {
                mySocketClient = new MySocketClient("127.0.0.1", 8080);
                mySocketClient.start();
                Thread.sleep(3000);
            }
        }
    }

## ServerSocketChannel与SocketChannel ##

### 基本工作原理 ###

### Selector常用方法 ###

Selector.open()：得到一个选择器对象；

selector.select()：阻塞 监控所有注册的通道,当有对应的事件操作时, 会将SelectionKey放入集合内部并返回事件数量；

selector.select(1000)：阻塞 1000 毫秒，监控所有注册的通道,当有对应的事件操作时, 会将SelectionKey放入集合内部并返回；

selector.selectedKeys()：返回存有SelectionKey的集合

### SelectionKey常用方法 ###

SelectionKey.isAcceptable()：是否是连接继续事件

SelectionKey.isConnectable()：是否是连接就绪事件

SelectionKey.isReadable()：是否是读就绪事件

SelectionKey.isWritable()：是否是写就绪事件

### SelectionKey中定义的4种事件 ###

SelectionKey.OP\_ACCEPT： 接收连接继续事件，表示服务器监听到了客户连接，服务器可以接收这个连接了

SelectionKey.OP\_CONNECT：连接就绪事件，表示客户端与服务器的连接已经建立成功

SelectionKey.OP\_READ： 读就绪事件，表示通道中已经有了可读的数据，可以执行读操作了（通道目前有数据，可以进行读操作了）

SelectionKey.OP\_WRITE： 写就绪事件，表示已经可以向通道写数据了（通道目前可以用于写操作）

### 服务端 ###

@Slf4j
    public class MyServerSocketChannel {
    
        public static void main(String[] args) throws IOException {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(8080));
            serverSocketChannel.configureBlocking(false);
            while (true) {
                SocketChannel socketChannel = serverSocketChannel.accept();
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                if (socketChannel != null) {
                    int read = socketChannel.read(byteBuffer);
                    String body = new String(byteBuffer.array(), 0, read, StandardCharsets.UTF_8);
                    log.info("收到客户端消息：{}", body);
                    //给客户端响应值
                    socketChannel.write(ByteBuffer.wrap("success".getBytes(StandardCharsets.UTF_8)));
                    socketChannel.close();
                }
            }
        }
    }

@Slf4j
    public class MyServerSocketChannel2 {
    
        public static void main(String[] args) throws IOException {
            //在服务端服务端声明一个服务端的网络通道
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            //网络通道上绑定端口号
            serverSocketChannel.bind(new InetSocketAddress(8080));
            //网络通道设置为非阻塞
            serverSocketChannel.configureBlocking(false);
            //服务端声明一个选择器
            Selector selector = Selector.open();
            //把通服务端的网络通道道注册到选择器上,并指定监听accept事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            //定义为死循环，持续接受来自客户端的网络请求
            while (true) {
                //阻塞监控所有注册到选择器上的网络通道，
                // 如果有来自客户端的网络请求，那么就会将网络请求的对应事件操作（连接事件、读取事件、写入事件等）存入SelectionKey集合内，并反返回可操作的事件数；
                int select = selector.select();
                //如果等于0，则表示无网络请求事件，直接循环跳过；
                if (select == 0) {
                    continue;
                }
                //如果监测到有网络请求，则拉取出网络请求的事件；
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                //遍历这些事件操作
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    //如果是请求连接事件就绪，则接受连接，设置通道的工作模式为非阻塞，并注册一个可读事件
                    if (selectionKey.isAcceptable()) {
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        socketChannel.configureBlocking(false);
                        socketChannel.register(selector, SelectionKey.OP_READ);
                    }
                    //如果是读取事件就绪，则获取客户端发来的数据，读取完毕后，注册一个写入事件
                    if (selectionKey.isReadable()) {
                        SocketChannel socketChannel = ((SocketChannel) selectionKey.channel());
                        ByteBuffer byteBuffer = ByteBuffer.allocate(5);
                        int read = 0;
                        StringBuilder sb = new StringBuilder();
                        while ((read = socketChannel.read(byteBuffer)) != 0) {
                            String body = new String(byteBuffer.array(), 0, read, StandardCharsets.UTF_8);
                            sb.append(body);
                            byteBuffer.clear(); log.info(body);
                        }
                        log.info("收到客户端的消息：{}", sb.toString());
                       socketChannel.register(selector,SelectionKey.OP_WRITE);
                    }
                    //如果是写入事件就绪，则向客户端写入内容，并最终关闭通道
                    if (selectionKey.isWritable()) {
                        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                        socketChannel.write(ByteBuffer.wrap(new String("success").getBytes(StandardCharsets.UTF_8)));
                        socketChannel.close();
                    }
                    //网络事件处理完后，从selectionKey集合中移除该事件；
                    iterator.remove();
                }
            }
        }
    }

### 客户端 ###

@Slf4j
    public class MyClientSocketChannel {
        public static void main(String[] args) throws IOException {
            SocketChannel socketChannel = SelectorProvider.provider().openSocketChannel();
            socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            socketChannel.write(ByteBuffer.wrap("hello ,I am client !".getBytes(StandardCharsets.UTF_8)));
            //读取服务端的响应值
            int read = socketChannel.read(byteBuffer);
            byte[] bytes = byteBuffer.array();
            String body = new String(bytes, 0, read, StandardCharsets.UTF_8);
            log.info("收到服务端的响应值：{}", body);
            socketChannel.close();
        }
    }

## Netty ##

### 服务端 ###

@Slf4j
    public class NettyServer {
        public void start(InetSocketAddress socketAddress) {
            //声明主线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            //声明作线程组
            EventLoopGroup workGroup = new NioEventLoopGroup(200);
            //声明服务端网络对象，并设置相应线程组、
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    //自定义的IO事件处理类
                    .childHandler(new ServerChannelInitializer())
                    .localAddress(socketAddress)
                    //设置队列大小
                    .option(ChannelOption.SO_BACKLOG, 1024);
            //绑定端口,开始接收进来的连接
            try {
                ChannelFuture future = bootstrap.bind(socketAddress).sync();
                log.info("服务器启动开始监听端口: {}", socketAddress.getPort());
                //阻塞等待服务端监听端口关闭
                future.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //优雅退出，释放资源
                //关闭主线程组
                bossGroup.shutdownGracefully();
                //关闭工作线程组
                workGroup.shutdownGracefully();
            }
        }
        public static void main(String[] args) {
            //启动服务端
            NettyServer nettyServer = new NettyServer();
            nettyServer.start(new InetSocketAddress("127.0.0.1", 8090));
        }
    }

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel socketChannel) {
            //添加编解码
    //        socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
            //自定义解码器，解决粘包的情况，DelimiterBasedFrameDecoder是以自定义分隔符作为报文结束的标志来解码报文数据
            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024 * 1024, Unpooled.copiedBuffer("|".getBytes())));
            //解码、编码的方式使用UTF-8字符集
            socketChannel.pipeline().addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
            socketChannel.pipeline().addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
            //声明事件处理器类
            socketChannel.pipeline().addLast(new NettyServerHandler());
        }
    
    }

@Slf4j
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 读取事件触发会执行
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            log.info("服务端收到服务端消息: {}", msg.toString());
    
        }
    
        /**
         * 读取事件完成后触发执行
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
           ctx.writeAndFlush("success|");
        }
    
        /**
         * 发生异常触发
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }

### 客户端 ###

@Slf4j
    public class NettyClient {
        public void start() {
            EventLoopGroup group = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    //该参数的作用就是禁止使用Nagle算法，使用于小数据即时传输
                    .option(ChannelOption.TCP_NODELAY, true)
                    .channel(NioSocketChannel.class)
                    .handler(new NettyClientInitializer());
    
            try {
                ChannelFuture future = bootstrap.connect("127.0.0.1", 8090).sync();
                //等待客户端链路关闭
                future.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //优雅退出，释放资源
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) {
            //启动netty客户端
            NettyClient nettyClient = new NettyClient();
            nettyClient.start();
        }
    }

public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
    //            socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024 * 1024, Unpooled.copiedBuffer("|".getBytes())));
            socketChannel.pipeline().addLast("decoder", new StringDecoder());
            socketChannel.pipeline().addLast("encoder", new StringEncoder());
            socketChannel.pipeline().addLast(new NettyClientHandler());
        }
    }

@Slf4j
    public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 连接事件触发执行
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush("hello ,I am client ! |");
        }
    
        /**
         * 读取事件触发执行
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            log.info("客户端收到消息: {}", msg.toString());
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.close();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }

源代码地址：[凡夫贩夫 / fanfu-web · GitCode][_ fanfu-web _ GitCode]（netty-demo分支）

[251e42a28eae4eebb5f04a285f5255cf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/48b90df549d44bfe961d66c0c44e8687.png
[0fa852969414427184d75df594311316.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/dd419f17d5424213b3ff8843efea138b.png
[1a4be953ab554e6b9c5eba7a3db08615.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/16/2f9a9651863d4d46903fd62837ae44fb.png
[_ fanfu-web _ GitCode]: https://gitcode.net/fox9916/fanfu-web