1

我有一个服务器应用程序。Java NIO

我有 Runnable 类 - EventHandler - 处理传入消息。如果 message == "Bye" -> EventHandler 关闭相关的 SocketServer 和 SelectorKey

我有一个 Runnable 对象——Acceptor——在 OP_ACCEPT 事件上被激活。它创建新的 SocketChannel 和新的 EventHandler 来处理来自这个通道的消息

我有个问题。第一个客户端连接。发送信息。断开。一切都好

第一个客户端断开连接后第二个客户端连接。问题从这里开始 - 没有调用 Acceptor 对象，因此没有为新客户端创建 SocketChannel 和 EventHandler。

我的代码有什么问题？SocketChannel 关闭不正确？

我更改了代码以修复评论中提到的错误。现在它工作正常

反应堆。带主循环的类

public class Reactor implements Runnable {

    final Selector selector;
    final ServerSocketChannel serverSocketChannel;

    Reactor(int port) throws IOException {
        //configure server socket channel
        this.selector = Selector.open();
        this.serverSocketChannel = ServerSocketChannel.open();
        this.serverSocketChannel.socket().bind(new InetSocketAddress(port));
        this.serverSocketChannel.configureBlocking(false);

        //start acceptor
        this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT, new Acceptor(this.serverSocketChannel, this.selector));
    }

    public void run() {
        System.out.println("Server is listening to port: " + serverSocketChannel.socket().getLocalPort());
        try {
            while (!Thread.currentThread().isInterrupted()) {
                if (this.selector.select() > 0) {
                    Set<SelectionKey> selected = this.selector.selectedKeys();
                    for (SelectionKey selectionKey : selected) {
                        dispatch(selectionKey);
                    }
                    selected.clear(); //clear set (thanks to EJP for comment)
                }
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    void dispatch(SelectionKey k) {
        Runnable r = (Runnable) (k.attachment());
        if (r != null) {
            r.run();
        }
    }
}

受体

public class Acceptor implements Runnable {

    final ServerSocketChannel serverSocketChannel;
    final Selector selector;

    public Acceptor(ServerSocketChannel serverSocketChannel, Selector selector) {
        this.serverSocketChannel = serverSocketChannel;
        this.selector = selector;
    }

    public void run() {
        try {
            SocketChannel socketChannel = this.serverSocketChannel.accept();
            if (socketChannel != null) {
                new EventHandler(this.selector, socketChannel);
                System.out.println("Connection Accepted");
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

事件处理程序

public class EventHandler implements Runnable {

    EventHandler(Selector selector, SocketChannel socketChannel) throws IOException {
        this.socketChannel = socketChannel;
        socketChannel.configureBlocking(false);
        this.selectionKey = this.socketChannel.register(selector, SelectionKey.OP_READ, this);
        //selector.wakeup();  //we don't need to wake up selector (thanks to EJP for comment)
    }

    @Override
    public void run() {
        try {
            if (this.state == EventHandlerStatus.READING) {
                read();
            } else if (this.state == EventHandlerStatus.SENDING) {
                send();
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * Reading client message
     *
     * @throws IOException
     */
    void read() throws IOException {
        int readCount = this.socketChannel.read(this.input);

        //check whether the result is equal to -1, and close the connection if it is (thanks to EJP for comment)
        if(readCount == -1){
            this.socketChannel.close();
            System.out.println("Stream is closed. Close connection.");
            return;
        }

        if (readCount > 0) {
            processMessage(readCount);
        }

        if(this.clientMessage.equalsIgnoreCase("Bye")){
            this.socketChannel.close();
            //this.selectionKey.cancel(); //we don't need to cancel selectionKey if socketChannel is just closed (thanks to EJP for comment)
            System.out.println("Client said Bye. Close connection.");
            return;
        }

        this.state = EventHandler.Status.SENDING;
        this.selectionKey.interestOps(SelectionKey.OP_WRITE); //mark that we interested in writing
    }

    /**
     * Processing of the read message.
     *
     * @param readCount Number of bytes to read
     */
    synchronized void processMessage(int readCount) {
        this.input.flip();
        StringBuilder sb = new StringBuilder();
        sb.append(new String(Arrays.copyOfRange(input.array(), 0, readCount))); // Assuming ASCII (bad assumption but simplifies the example)
        this.clientMessage = sb.toString().trim();
        this.input.clear();
        System.out.println("Client said: " + this.clientMessage);
    }

    /**
     * Sending response to client
     *
     * @throws IOException
     */
    void send() throws IOException {
        System.out.println("Answer to client: " + this.clientMessage);
        this.socketChannel.write(ByteBuffer.wrap((this.clientMessage + "\n").getBytes()));
        this.state = EventHandler.Status.READING;
        this.selectionKey.interestOps(SelectionKey.OP_READ); //mark that we interested in reading
    }

//----------------------------------------------------------------------------------------------------------------------
//  Fields
//----------------------------------------------------------------------------------------------------------------------

    final SocketChannel socketChannel;
    final SelectionKey selectionKey;

    ByteBuffer input = ByteBuffer.allocate(1024);
    EventHandlerStatus state = EventHandler.Status.READING;
    String clientMessage = "";

//----------------------------------------------------------------------------------------------------------------------
//  Enum to mark current status of EventHandler
//----------------------------------------------------------------------------------------------------------------------

    enum Status {
        READING, SENDING
    }

}
4

0 回答 0