问题描述

当启动 服务端，客户端 连接成功，并向 读写selector 注册 客户端 channel 的读写事件时，读写线程无限循环检测 读写selector select() 阻塞操作 是否有就绪的事件时，select() 一直返回 0 的问题。

正常流程：

正常流程当客户端连接成功，并在 2s 后向 writeSelector 注册写事件时，IO 类中的 startWrite() 方法肯定会执行这句 System.out.println("有就绪的写事件~~~~~~~~~~~~~"); 打印，但是在下面的代码中是不执行的。

代码

服务端

Server.java

/**
 *
 */
public class Server {
    public static void main(String[] args) {
        try {
            Server server = new Server();
            server.start(8899);
        } catch (IOException e) {
            System.err.println("ERROR:" + e);
        }
    }

    private final Selector selector;
    private final ServerSocketChannel server;
    private ClientListener listener;

    public Server() throws IOException {
        selector = Selector.open();
        server = ServerSocketChannel.open();
        server.configureBlocking(false);
    }

    public void start(int port) throws IOException {
        server.bind(new InetSocketAddress(port));
        server.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服务端信息：" + server.getLocalAddress().toString());

        listener = new ClientListener(selector);
        listener.start();
    }
}

ClientListener.java

public class ClientListener extends Thread {
    private boolean done = false;
    private final Selector selector;

    public ClientListener(Selector selector) {
        this.selector = selector;
    }

    @Override
    public void run() {
        System.out.println("服务端准备就绪~~");
        do {
            try {
                if(selector.select() == 0) {
                    if(done) {
                        break;
                    }
                    continue;
                }

                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if(key.isAcceptable()) {
                        ServerSocketChannel server = (ServerSocketChannel) key.channel();
                        SocketChannel channel = server.accept();
                        if(channel != null) {
                            System.out.println("有新客户端连接....");
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } while (!done);
    }
}

客户端

Client.java

public class Client {
    public static void main(String[] args) {
        Client client = null;
        try {
            client = new Client();
            client.connect("127.0.0.1", 8899);
        } catch (IOException e) {
            System.err.println("ERROR:" + e);
        }

        if(client != null) {
            try {
                Thread.sleep(2000);
                client.registerWrite();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private final SocketChannel channel;
    private IO io;

    public Client() throws IOException {
        channel = SocketChannel.open();
        channel.configureBlocking(false);
    }

    public void connect(String host, int port) throws IOException {
        channel.connect(new InetSocketAddress(host, port));
        System.out.println("已发起服务端连接.");

        io = new IO();
    }

    /**
     * 测试注册写事件
     */
    public void registerWrite() {
        try {
            System.out.println("注册写事件开始.");
            SelectionKey key = io.registerWrite(channel);
            System.out.println("注册写事件完成. key:" + key);
        } catch (ClosedChannelException e) {
            System.err.println("注册写事件失败!! " + e);
        }
    }
}

IO 操作

IO.java

public final class IO {
    private final AtomicBoolean isClosed = new AtomicBoolean();

    private final Selector readSelector;
    private final Selector writeSelector;

    public IO() throws IOException {
        readSelector = Selector.open();
        writeSelector = Selector.open();
        
        startRead();
        startWrite();
    }

    private void startRead() {
        Thread thread = new Thread("") {

            @Override
            public void run() {
                while (!isClosed.get()) {
                    try {
                        if(readSelector.select() == 0) {
                            continue;
                        }

                        Iterator<SelectionKey> iterator = readSelector.selectedKeys().iterator();
                        while(iterator.hasNext()) {
                            SelectionKey key = iterator.next();
                            iterator.remove();

                            if(key.isValid()) {
                                System.out.println("处理读事件~~~~~~~~~~~~~ key:" + key);
                            }
                        }
                    } catch (IOException e) {
                        System.out.println("处理读事件失败!! " + e);
                    }
                }
            }
        };
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.start();
    }

    private void startWrite() {
        Thread thread = new Thread("") {

            @Override
            public void run() {
                while (!isClosed.get()) {
                    try {
                        if(writeSelector.select() == 0) {
                            continue;
                        }
                        System.out.println("有就绪的写事件~~~~~~~~~~~~~");

                        Iterator<SelectionKey> iterator = writeSelector.selectedKeys().iterator();
                        while(iterator.hasNext()) {
                            SelectionKey key = iterator.next();
                            iterator.remove();

                            if(key.isValid()) {
                                System.out.println("处理写事件~~~~~~~~~~~~~ key:" + key);
                            }
                        }
                    } catch (IOException e) {
                        System.out.println("处理写事件失败!! " + e);
                    }
                }
            }
        };
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.start();
    }

    public SelectionKey registerWrite(SocketChannel channel) throws ClosedChannelException {
        System.out.println("注册写事件进行中......");
        writeSelector.wakeup();
        return channel.register(writeSelector, SelectionKey.OP_WRITE);
    }
}