NIO源码解析-SocketChannel

2022-06-19 04:29:01

前言：

SocketChannel作为网络套接字的通道，与之前我们学习到的FileChannel有很多不同之处（就是两个大类别的通道）。

没有SocketChannel之前，我们创建网络连接一般都是通过Socket和ServerSocket，这些都是BIO类别，性能的扩展会受到影响。

借助NIO相关实现SocketChannel和ServerSocketChannel，我们可以管理大量连接并且实现更小的性能损失。

本文就来介绍下SocketChannel的相关使用。

我们来给定一个需求：就是创建一个简易的对话框，使客户端和服务端可以接收到彼此的对话，并予以响应。（本篇专注于client端，也就是Socket和SocketChannel，下一篇会继续将server端的补上）。

1.基于Socket的客户端

public class BIOClientSocket {

    private String address;
    private int port;

    public BIOClientSocket(String address, int port) {
        this.address = address;
        this.port = port;
    }

    public void connectToServer() {
        Socket socket = new Socket();
        try {
            socket.connect(new InetSocketAddress(address, port));

            // 写数据
            new ClientWriteThread(socket).start();
            // 读数据
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String msg = "";
            while ((msg = bufferedReader.readLine()) != null) {
                System.out.println("receive msg: " + msg);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        String address = "localhost";
        int port = 9999;

        BIOClientSocket bioClientSocket = new BIOClientSocket(address, port);
        bioClientSocket.connectToServer();
    }
}

/**
 * 客户端发送请求线程
 */
class ClientWriteThread extends Thread {
    private Socket socket;
    private PrintWriter writer;
    private Scanner scanner;

    public ClientWriteThread(Socket socket) throws IOException {
        this.socket = socket;
        this.scanner = new Scanner(System.in);
        this.writer = new PrintWriter(new OutputStreamWriter(socket.getOutputStream()), true);
    }

    @Override
    public void run() {
        String msg = "";

        try {
            // 通过获取对话框里的消息，不断发送到server端
            while ((msg = scanner.nextLine()) != null) {
                if (msg.equals("bye")) {
                    break;
                }
                writer.println(msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

以上就是标准的Socket客户端与服务端交互的代码，也比较简单，笔者不再详述

2.基于SocketChannel的客户端

public class NIOClientSocket {

    private String address;
    private int port;
    private Selector selector;

    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
    private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

    private Scanner scanner = new Scanner(System.in);


    public NIOClientSocket(String address, int port) throws IOException {
        this.address = address;
        this.port = port;
        this.selector = Selector.open();
    }

    public void connectToServer() {
        try {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);

            socketChannel.register(selector, SelectionKey.OP_CONNECT);
            socketChannel.connect(new InetSocketAddress(address, port));

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

    private void connect() {
        while (true) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                for (SelectionKey key : selectionKeys) {
                    if (key.isConnectable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        if (clientChannel.isConnectionPending()) {
                            clientChannel.finishConnect();
                            System.out.println("client connect success...");
                        }

                        clientChannel.register(selector, SelectionKey.OP_WRITE);
                    } else if (key.isReadable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        StringBuffer sb = new StringBuffer("receive msg: ");

                        readBuffer.clear();
                        while (clientChannel.read(readBuffer) > 0) {
                            readBuffer.flip();
                            sb.append(new String(readBuffer.array(), 0, readBuffer.limit()));
                        }

                        System.out.println(sb.toString());
                        clientChannel.register(selector, SelectionKey.OP_WRITE);
                    } else if (key.isWritable()) {
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        String msg = scanner.nextLine();
                        writeBuffer.clear();

                        writeBuffer.put(msg.getBytes());
                        writeBuffer.flip();
                        clientChannel.write(writeBuffer);
                        clientChannel.register(selector, SelectionKey.OP_READ);
                    }
                }

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

    public static void main(String[] args) {
        String address = "localhost";
        int port = 9999;

        try {
            new NIOClientSocket(address, port).connectToServer();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

借助Selector，我们将想要监听的事件注册到Selector上。

client端默认先进行写，故在连接建立完成之后，直接注册了写事件；

写的事件会阻塞到Scanner上，等待用户输入，输入后传输给Server端，然后注册读事件；

通过这样的读写事件来回注册，就可以实现类似对话框的效果。（当然，必须是一问一答）。

3.SocketChannel API

我们先来看下其类结构图

可以看到，其可读可写（实现了ByteChannel）；可通过Selector进行事件注册（继承了SelectableChannel）；可进行端口绑定，Socket属性设置（实现了NetworkChannel）；

3.1 非阻塞模式

SocketChannel提供configureBlocking方法（本质上是AbstractSelectableChannel提供的），来描述通道的阻塞状态。我们可以将SocketChannel设置为非阻塞状态。同时其还提供了isBlocking方法来查询其阻塞状态。 传统的Socket其阻塞性是影响系统可伸缩性的重要约束。而这种非阻塞的SocketChannel则是许多高性能程序构建的基础。 延伸：阻塞socket与非阻塞socket两者之间有哪些具体区别呢？ 1）输入操作 进程A调用阻塞socket.read方法时，若该socket的接收缓冲区没有数据可读，则该进程A被阻塞，操作系统将进程A睡眠，直到有数据到达；进程A调用非阻塞socket.read方法时，若该socket的接收缓冲区没有数据可读，则进程A收到一个EWOULDBLOCK错误提示，表示无可读数据，read方法立即返回，进程A可针对错误提示进行后续操作。 2）输出操作 进程A调用阻塞socket.write方法时，若该socket的发送缓冲区没有多余空间，则进程A被阻塞，操作系统将进程A睡眠，直到有空间为止；进程A调用非阻塞socket.write方法时，若该socket的发送缓冲区没有多余空间，则进程A收到一个EWOULDBLOCK错误提示，表示无多余空间，write方法立即返回，进程A可针对错误提供进行后续操作 3）连接操作 对于阻塞型的socket而言，调用socket.connect方法创建连接时，会有一个三次握手的过程，每次需要等到三次握手完成之后（ESTABLISHED 状态），connect方法才会返回，这意味着其调用进程需要至少阻塞一个RTT时间。对于非阻塞的SocketChannel而言，调用connect方法创建连接时，当三次握手可以立即建立时（一般发生在客户端和服务端在一个主机上时），connect方法会立即返回；而对于握手需要阻塞RTT时间的，非阻塞的SocketChannel.connect方法也能照常发起连接，同时会立即返回一个EINPROGRESS（在处理中的错误）。正如上述2中的代码：

// SocketChannel直接建立连接，当前进程并没有阻塞
socketChannel.connect(new InetSocketAddress(address, port));

// 后续通过注册的Selector来获取连接状态
// 当selector检测到SocketChannel已经完成连接或连接报错，则会添加OP_CONNECT到key的就绪列表中
if (key.isConnectable()) {
    SocketChannel clientChannel = (SocketChannel) key.channel();
    // 此时需要判断连接是否成功
    if (clientChannel.isConnectionPending()) {
        clientChannel.finishConnect();
        System.out.println("client connect success...");
    }

3.2 NetworkChannel（网络连接相关方法）

SocketChannel实现了NetworkChannel接口的相关方法，来完成ip：port的绑定，socket属性的设置。

// 使当前channel绑定到具体地址
NetworkChannel bind(SocketAddress local) throws IOException;

// 设置socket属性
<T> NetworkChannel setOption(SocketOption<T> name, T value) throws IOException;

3.3 AbstractSelectableChannel（绑定Selector相关方法）

SocketChannel继承了AbstractSelectableChannel抽象类，来完成Selector的注册，多路复用功能。

// 将当前通道注册到Selector上
public abstract SelectionKey register(Selector sel, int ops, Object att)
        throws ClosedChannelException;

// 获取当前selector上可执行的操作（OP_READ OP_WRITE...）
public final SelectionKey keyFor(Selector sel)

3.4 ByteChannel（数据的读写）

SocketChannel实现ByteChannel接口，这个接口我们之前了解过，ByteChannel接口继承了ReadableByteChannel和WritableByteChannel，实现了对数据的读写。

上文中的示例里，clientChannel.read()和clientChannel.write()方法就是对其的使用。

4.Socket与SocketChannel

通过以上的介绍，我们会使用了SocketChannel，也会使用Socket来创建对服务端的连接。那么这两者之间有什么关系吗？

// A socket is an endpoint for communication between two machines
public class Socket implements java.io.Closeable {
    /** A socket will have a channel if, and only if, the channel itself was
     * created via the{@link java.nio.channels.SocketChannel#open
     * SocketChannel.open} or {@link
     * java.nio.channels.ServerSocketChannel#accept ServerSocketChannel.accept} */
    public SocketChannel getChannel() {
        return null;
    }
}

根据其类上面的注释，我们可以看到，Socket是一个端点，用于连接两个机器。

而直接使用socket.getChannel方法来获取其对应的通道时，则返回了null，同时给出提示：我们只能通过SocketChannel.open或者ServerSocketChannel.accept方法来获取通道。

// A selectable channel for stream-oriented connecting sockets
public abstract class SocketChannel
    extends AbstractSelectableChannel
    implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel
{
    // Retrieves a socket associated with this channel.
	public abstract Socket socket();
}

同样看注释，SocketChannel被描述为一个可选择（注册到Selector上）的通道，用来连接socket（client-server）。

而SocketChannel.socket方法，则返回通道对应的Socket。

总结：虽然每个SocketChannel通道都有一个关联的Socket对象，但并非所有的socket都有一个关联的SocketChannel。

如果我们使用传统的方式来new Socket，那么其不会有关联的SocketChannel

参考：

非阻塞式socket_一个菜鸟的博客-CSDN博客_非阻塞socket

SocketChannel---各种注意点_billluffy的博客-CSDN博客_socketchannel NIO相关的坑，大家可以借鉴下

码农公寓

前言：