Java NIO的理解和应用

Java NIO的理解和应用

Java NIO是一种基于通道和缓冲区的I/O方式,已经被广泛的应用,成为解决高并发与大量连接和I/O处理问题的有效方式。

Java NIO相关组件

Java NIO主要有三个核心部分组成,分别是:Channel(通道),Buffer(缓冲区), Selector(选择器)

  • Channel

Channel是所有访问IO设备的统称。类型与IO中的Stream,而通道是双向的,既可以读又可以写,但是Stream是单项的。常用的通道有:SocketChannelServerSocketChannel(对应TCP的客户端和服务器端)、FileChannel(对应文件IO)、DatagramChannel(对应UDP)等

  • Buffer

所有数据的读写都要经过BufferBuffer直接和Channel打交道,是一个存储数据的容器。通过调用Channel.write方法将数据写入BufferChannel.read方法将数据从Buffer中读取出来。常用的Buffer有:ByteBufferLongBufferIntBufferStringCharBuffer

  • Selector

Selector用来监听多个Channel的事件(比如:Read、Write、Connect和Accept等),通过单个线程轮询的方式实现了对多个Channel的监听。

Java IO与NIO的区别

NIO是一种叫非阻塞IO(Non-blocking I/O),基于I/O多路复用来实现的(可参考:I/O模型、select、poll和epoll之间的区别)。NIO与之前传统的I/O模型有很大的不同,具体表现在以下几个方面:

  • 面向流与面向缓冲

Java IO和NIO之间一个最大的区别是,IO是面向流的,NIO是面向缓冲区的。Java IO每次从数据流中读一个或多个字节,直至读取所有字节,数据流是一次性的,读取完以后,不能前后移动流中的数据。Java NIO是将数据读取到缓冲区,可以通过position来回移动访问缓冲区中的数据。

  • 阻塞与非阻塞IO

Java IO中调用readwrite方法的线程会被阻塞的,直到数据全部读入或者全部写入完为止。而在Java NIO中,如果需要读写数据只用和缓冲区打交道,将数据从缓冲区读取或者写入缓冲区以后,线程可以继续做其他事情,不会被block住。

  • 选择器(Selector)

Selector是基于I/O多路复用的机制实现的,将多个Channel注册到一个Selector上,Selector通过轮询监听所有注册的通道上是否有SelectionKey发生,如果发生了,然后将SelectionKey分派给其他线程处理。

Java NIO的应用

通过Java NIO技术简单实现了一个服务端与客户端通信的case,具体功能如下:

  • 服务端可以向客户端广播消息
  • 服务端将一个客户端的消息转发给其他客户端
  • 客户端向服务端发送消息
  • 客户端接收服务端的消息

服务端代码如下:

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Server {

    public static void main(String[] args) throws IOException {
        new Server().start(); // 启动服务端程序
    }

    public Server() throws IOException {
        this.init(); // 初始化服务端数据
    }

    /**
     * 服务端端口
     */
    private int port = 9999;

    /**
     * 服务端的Selector用来监听Channel的事件.
     */
    private Selector selector;

    /**
     * 字符数据编码
     */
    private Charset charset = Charset.forName("UTF-8");

    /**
     * 读缓存,分配1024Byte的空间
     */
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);

    /**
     * 写缓存,分配1024Byte的空间
     */
    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    /**
     * 存储所有客户端的Channel,转发的时候使用
     */
    private Map<String, Channel> clientSocketChannels = new HashMap<>();

    /**
     * 定义了一个线程池,服务端用来发送数据给客户端
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(1, runnable -> {
        Thread thread = new Thread(runnable);
        thread.setDaemon(true);
        thread.setName("server-sender");
        return thread;
    });

    /**
     * 初始化Channel.
     */
    private void init() throws IOException {
        // 声明一个服务端的ServerSocketChannel
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 将服务端的ServerSocketChannel设置成非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 设置服务端的socket
        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(this.port));
        // 声明一个Selector,用来监听服务端的所有Channel
        this.selector = Selector.open();
        // 在ServerSocketChannel上注册Accept事件,用来接收客户端的连接
        serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        System.out.println("Server is started, the port is " + this.port);
    }

    /**
     * 处理服务端监听到的事件
     */
    private void work(SelectionKey selectionKey) throws IOException {
        // 客户端有Socket连接请求
        if (selectionKey.isAcceptable()) {
            // 从SelectionKey中获取服务端的ServerSocketChannel,SelectionKey中包含了服务端与客户端的所有信息
            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
            // 服务端打开一个新的SocketChannel用来与客户端的SocketChannel进行通信,服务端同时会随机分配一个端口
            SocketChannel socketChannel = serverSocketChannel.accept();
            // 将SocketChannel设置成非阻塞模式
            socketChannel.configureBlocking(false);
            // 将SocketChannel中的Read事件注册到Selector上
            socketChannel.register(this.selector, SelectionKey.OP_READ);
            // 存储服务端为客户端创建的SocketChannel,为后面的转发消息服务
            this.clientSocketChannels.put(this.getClientName(socketChannel), socketChannel);
            // 通过System.in IO流来创建Scanner
            Scanner scanner = new Scanner(System.in);
            // 收集服务端控制台输入的数据,通过线程池将数据广播给所有客户端SocketChannel
            this.executorService.submit(() -> {
                while (true) {
                    // 该方法会被block住,一直等到服务端控制台有数据输入完为止
                    String sendText = scanner.nextLine();
                    // 将服务端的数据广播给所有客户端
                    transferToOthers(sendText, null);
                }
            });
        // 服务端监听到有数据可以读取,主要是来源于客户端发送的数据
        } else if (selectionKey.isReadable()) {
            // 获取服务端的SocketChannel,然后与客户端进行通信
            // 需要注意的是:当前获取的SocketChannel与ServerSocketChannel是不同的,
            // 这个SocketChannel是通过调用ServerSocketChannel.accept方法创建的(存储在clientSocketChannels集合中)
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
            // 清空当前的用来存储读数据的buffer
            readBuffer.clear();
            // 将数据从SocketChannel读入buffer
            int bytes = socketChannel.read(readBuffer);
            if (bytes > 0) {
                // 使得buffer中的数据可读
                readBuffer.flip();
                // 读取buffer中的数据
                String text = String.valueOf(this.charset.decode(readBuffer));
                System.out.println(this.getClientName(socketChannel) + ": " + text);
                // 将客户端发送过来的数据转发给其他客户端
                this.transferToOthers(text, socketChannel);
            }
        }
    }

    /**
     * 将数据发送给其他客户端
     */
    private void transferToOthers(String text, final SocketChannel socketChannel) {
        this.clientSocketChannels.forEach((channelName, channel) -> {
            // 获取之前存储的与服务端建立连接的客户端
            SocketChannel otherSocketChannel = (SocketChannel) channel;
            if (!otherSocketChannel.equals(socketChannel)) {
                // 清空写缓存
                this.writeBuffer.clear();
                // 将数据写入缓存
                this.writeBuffer.put(this.charset.encode(this.getClientName(socketChannel) + ": " + text));
                // 使得缓存中的数据变得可用
                this.writeBuffer.flip();
                try {
                    // 将buffer中的数据写入到其它客户端
                    otherSocketChannel.write(this.writeBuffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 通过SocketChannel生成客户端的名字,用来标识
     */
    private String getClientName(SocketChannel socketChannel) {
        if (socketChannel == null)
            return "[server]";
        Socket socket = socketChannel.socket();
        return "[" + socket.getInetAddress().toString().substring(1) + ":" + socket.getPort() + "]";
    }

    /**
     * 启动服务端程序
     */
    public void start() {
        // 无限循环来轮询所有注册的Channel
        while (true) {
            try {
                // 选择已经准备好的Channel,该方法是会block住的,直到有事件到达
                this.selector.select();
                // 获取所有监听到的事件
                Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    // 找到事件SelectionKey,里面包含了事件相关的所有数据
                    SelectionKey selectionKey = iterator.next();
                    // 如果事件是有效的
                    if (selectionKey.isValid()) {
                        // 处理事件
                        this.work(selectionKey);
                    }
                    // 删除已经处理过的事件
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

客户端代码如下:

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class Client {

    public static void main(String[] args) throws IOException {
        new Client().start(); // 客户端程序执行入口
    }

    /**
     * 注册监听的服务的端口,并初始化
     */
    public Client() throws IOException {
        this.serverSocketAddress = new InetSocketAddress("127.0.0.1", 9999);
        this.init();
    }

    /**
     * 服务的Socket地址
     */
    private SocketAddress serverSocketAddress;

    /**
     * 客户端Selector
     */
    private Selector selector;

    /**
     * 字符编码
     */
    private Charset charset = Charset.forName("UTF-8");

    /**
     * 读缓冲区
     */
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);

    /**
     * 写缓冲区
     */
    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    /**
     * 线程池执行客户端发送数据
     */
    private static ExecutorService executorService = Executors.newFixedThreadPool(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable);
            thread.setDaemon(true);
            thread.setName("client-sender");
            return thread;
        }
    });

    /**
     * 初始化客户端信息
     */
    private void init() throws IOException {
        // 声明一个客户端SocketChannel
        SocketChannel socketChannel = SocketChannel.open();
        // 设置成非阻塞模式
        socketChannel.configureBlocking(false);
        // 声明一个Selector
        this.selector = Selector.open();
        // 将客户端的SocketChannel的连接事件注册到selector上
        socketChannel.register(this.selector, SelectionKey.OP_CONNECT);
        // 连接服务端
        socketChannel.connect(this.serverSocketAddress);
    }

    /**
     * 处理客户端数据
     */
    private void work(SelectionKey selectionKey) {
        try {
            //  与服务端建立连接
            if (selectionKey.isConnectable()) {
                // 从SelectionKey中获取客户端的ServerSocketChannel
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                // 判断连接是否完成
                if (socketChannel.isConnectionPending()) {
                    // 完成连接
                    socketChannel.finishConnect();
                    System.out.println("The connection is successful!");
                    // 通过System.in IO流来创建Scanner
                    Scanner scanner = new Scanner(System.in);
                    // 使用线程池来完成对客户端的控制台数据输入的监听
                    executorService.submit((Runnable) () -> {
                        while (true) {
                            try {
                                // 清空写缓冲区
                                writeBuffer.clear();
                                // 该方法会被block住,一直等到客户端控制台有数据输入完为止
                                String sendText = scanner.nextLine();
                                // 将数据写入写缓冲区
                                writeBuffer.put(charset.encode(sendText));
                                // 使得写缓冲区中的数据可读
                                writeBuffer.flip();
                                // 将数据通过SocketChannel发送到服务端
                                socketChannel.write(writeBuffer);
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
                // 注册可读事件,应该当前的SocketChannel与服务端建立连接以后,不需要再监听创建连接的事件
                // 为了复用SocketChannel,将SocketChannel的Read事件注册到Selector
                socketChannel.register(selector, SelectionKey.OP_READ);
            }
            // 可读事件,有从服务器端发送过来的信息,读取输出到控制台上
            else if (selectionKey.isReadable()) {
                // 获取与服务端通信的客户端SocketChannel
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                // 清空读缓冲区
                this.readBuffer.clear();
                // 将数据读取到读缓冲区,并将数据输出到客户端控制台
                int count = socketChannel.read(this.readBuffer);
                if (count > 0) {
                    String text = new String(this.readBuffer.array(), 0, count);
                    System.out.println(text);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动客户端程序
     */
    public void start() throws IOException {
        // 无限循环,轮询所有监听的SocketChannel
        while (true) {
            // 选择已经准备好的Channel,该方法是会block住的,直到有事件到达
            int events = this.selector.select();
            if (events > 0) {
                // 找到事件SelectionKey,里面包含了事件相关的所有数据
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 处理事件
                selectionKeys.forEach(selectionKey -> this.work(selectionKey));
                // 清空已处理的事件
                selectionKeys.clear();
            }
        }
    }
}

总结

  • 服务端的ServerSocketChannel是用来监听客户端的连接请求,只有1个且端口固定,主要监听accept事件
  • 服务端的SocketChannel是用来和客户端建立数据读写操作通信,数量与客户端的连接数量一致,每个都分配一个随机的端口,主要监听read事件
  • 每个客户端有一个SocketChannel,用来和服务端进行通信,主要监听connect事件和read事件,connect事件只会在第一连接时发生,read事件是在每次接收服务端数据时发生
  • 服务端和客户端各有一个Selector,用来监听所有的SocketChannel或者ServerSocketChannel中注册的事件,在没有事件发生的时候,Selector.select()会被block住
  • 在定义缓冲区的时候要注意缓冲区的大小,如果太小会报BufferOverflowException
hmoban主题是根据ripro二开的主题,极致后台体验,无插件,集成会员系统
自学咖网 » Java NIO的理解和应用