NIO 实现非阻塞 Socket 通讯

NIO 实现多人聊天室的案例

服务端

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;

/**
 * 聊天室服务端
 */
public class NServer {

    private Selector selector = null;

    static final int PORT = 30000;

    private Charset charset = Charset.forName("UTF-8");

    ServerSocketChannel server = null;

    public void init() throws IOException {
        selector = Selector.open();
        server = ServerSocketChannel.open();
        InetSocketAddress isa = new InetSocketAddress("127.0.0.1", PORT);
        // 将该 ServerSocketChannel 绑定到指定 IP 地址
        server.bind(isa);

        //设置为以非阻塞方式工作
        server.configureBlocking(false);
        // 将 server 注册到指定的 Selector
        server.register(selector, SelectionKey.OP_ACCEPT);

        while (selector.select() > 0) {
            for (SelectionKey sk : selector.selectedKeys()) {
                // 从 selector 上的已选择 Key 集中删除正在处理的 SelectionKey
                selector.selectedKeys().remove(sk);
                // 如果 sk 对应 channel 包含客户端的连接请求
                if (sk.isAcceptable()) {
                    // 接受请求
                    SocketChannel sc = server.accept();
                    // 采用非阻塞模式
                    sc.configureBlocking(false);
                    // 将该 SocketChannel 也注册到 selector
                    sc.register(selector, SelectionKey.OP_READ);
                    // 将 sk 对应的 channel 设置成准备接受其他请求
                    sk.interestOps(SelectionKey.OP_ACCEPT);
                }
                // 如果 sk 对应的channel 有数据需要读取
                if (sk.isReadable()) {
                    SocketChannel sc = (SocketChannel) sk.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    String content = "";

                    try {
                        // 读取数据操作
                        while (sc.read(buffer) > 0) {
                            buffer.flip();
                            content += charset.decode(buffer);
                        }
                        System.out.println("读取的数据： " + content);
                        sk.interestOps(SelectionKey.OP_READ);
                    } catch (IOException e) {
                        sk.cancel();
                        if (sk.channel() != null) {
                            sk.channel().close();
                        }
                    }
                    // 如果 content 的长度大于 0，即聊天信息不为空
                    if (content.length() > 0) {
                        // 遍历该 selector 里注册的所有 SelectionKey
                        for (SelectionKey key : selector.keys()) {
                            // 获取 channel
                            Channel targetChannel = key.channel();
                            // 如果该 channel 是 SocketChannel
                            if (targetChannel instanceof SocketChannel) {
                                // 将读到的内容写到该 channel 中
                                SocketChannel dest = (SocketChannel) targetChannel;
                                dest.write(charset.encode(content));
                            }
                        }
                    }
                }
            }
        }

    }

    public static void main(String[] args) throws IOException {
        new NServer().init();
    }

}

启动时简历一个可监听连接请求的 ServerSocketChannel，并注册到 Selector，接着直接采用循环不断监听 Selector 对象的 select() 方法返回值，大于0时，处理该 Selector 上所有被选择的 SelectionKey。

服务端仅需监听两种操作：连接和读取数据。

处理连接操作时，只需将连接完成后产生的 SocketChannel 注册到指定的 Selector 对象；

处理读取数据时，先从该 Socket 中读取数据，再将数据写入 Selector 上注册的所有 Channel 中。

客户端

import java.io.IOException;
import java.net.InetSocketAddress;
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;

/**
 * 聊天室客户端
 */
public class NClient {
    private Selector selector = null;
    static final int PORT = 30000;
    private Charset charset = Charset.forName("UTF-8");

    private SocketChannel sc = null;

    public void init() throws IOException {
        selector = Selector.open();
        InetSocketAddress isa = new InetSocketAddress("127.0.0.1", PORT);
        // 打开套接字通道并将其连接到远程地址
        sc = SocketChannel.open(isa);

        // 设置为非阻塞模式
        sc.configureBlocking(false);
        // 注册到 selector
        sc.register(selector, SelectionKey.OP_READ);

        new ClientThread().start();
        // 创建键盘输入流
        Scanner scan = new Scanner(System.in);
        while (scan.hasNextLine()) {
            String line = scan.nextLine();
            // 将键盘大忽如的内容输出到 SocketChannel 中
            sc.write(charset.encode(line));
        }

    }

    private class ClientThread extends  Thread {
        @Override
        public void run() {
            try {
                while (selector.select() > 0) {
                    for (SelectionKey sk : selector.selectedKeys()) {
                        // 从 set集合删除正在处理的 SelectionKey
                        selector.selectedKeys().remove(sk);
                        // 如果 sk 对应的 channel 中有可读数据
                        if (sk.isReadable()) {
                            // 使用 NIO 读取 channel 中的数据
                            SocketChannel sc = (SocketChannel) sk.channel();
                            ByteBuffer buff = ByteBuffer.allocate(1024);
                            String content = "";
                            while (sc.read(buff) > 0) {
                                sc.read(buff);
                                buff.flip();
                                content += charset.decode(buff);
                            }
                            System.out.println("聊天信息： " + content);
                            // 为下一次读取做准备
                            sk.interestOps(SelectionKey.OP_READ);
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        new NClient().init();
    }
}

相比于服务端程序，客户端要简单一些，只有一个 SocketChannel ，将其注册到指定的 Selector 后，程序启动另一个线程来监听该 Selector 即可。

分别启动两个程序后，可以在客户点输入内容，在服务端就可以读取到输入的内容：

服务端内容：