为什么我的NIO客户端到服务器的连接丢失？

Question

我目前有一个在非阻塞模式下使用 NIO java 库编写的桌面服务器。您可以在此处找到完整的服务器项目。我还创建了一个用于测试目的的非阻塞 NIO 客户端。你可以在这里找到这个项目。最后，服务器应该用于Android即时通讯应用程序。我使客户端和服务器都基于发送“数据包”进行通信的想法。我的意思是，Packet 类的引用被打包到字节缓冲区中并通过套接字通道发送。然后，它们在另一端反序列化并执行。

我当前的问题：似乎我的测试客户端在连接到服务器时断开连接。我知道问题是客户端的，因为当我使用 telnet 通过命令行连接到我的服务器时，连接不会断开。奇怪的部分是，服务器正在打印出已建立与它的连接，但在客户端断开连接时从未说连接已丢失/终止。

这是处理所有 nio 网络的客户端类...

 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.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.baiocchi.enigma.client.test.Engine;
import org.baiocchi.enigma.client.test.databundle.DataBundle;
import org.baiocchi.enigma.client.test.packet.Packet;
import org.baiocchi.enigma.client.test.ui.LogType;
import org.baiocchi.enigma.client.test.ui.Logger;

public class Client extends Thread {

    private boolean running;
    private final int port;
    private SocketChannel connection;
    private final ByteBuffer buffer;
    private Selector selector;
    private List<DataBundle> pendingDataBundleQue;

    public Client(int port) {
        this.port = port;
        pendingDataBundleQue = new LinkedList<DataBundle>();
        buffer = ByteBuffer.allocate(8192);
        try {
            selector = initiateSelector();
            connection = initiateConnection();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private Selector initiateSelector() throws IOException {
        return SelectorProvider.provider().openSelector();
    }

    private SocketChannel initiateConnection() throws IOException {
        SocketChannel connection = SocketChannel.open();
        connection.configureBlocking(false);
        connection.connect(new InetSocketAddress("localhost", port));
        connection.register(selector, SelectionKey.OP_CONNECT);
        return connection;
    }

    public SocketChannel getConnection() {
        return connection;
    }

    @Override
    public void start() {
        running = true;
        super.start();
    }

    public void addToPendingDataBundleQue(DataBundle bundle) {
        synchronized (pendingDataBundleQue) {
            pendingDataBundleQue.add(bundle);
        }
    }

    @Override
    public void run() {
        while (running) {
            System.out.println("loop");
            try {
                synchronized (pendingDataBundleQue) {
                    System.out.println("Checking for que changes.");
                    if (!pendingDataBundleQue.isEmpty()) {
                        System.out.println("Found que change.");
                        SelectionKey key = connection.keyFor(selector);
                        key.interestOps(SelectionKey.OP_WRITE);
                    }
                }
                System.out.println("Selecting keys");
                selector.select();
                System.out.println("Creating selected keys list.");
                Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
                while (selectedKeys.hasNext()) {
                    System.out.println("scrolling through list");
                    SelectionKey key = (SelectionKey) selectedKeys.next();
                    selectedKeys.remove();
                    if (!key.isValid()) {
                        System.out.println("invalid");
                        continue;
                    } else if (key.isConnectable()) {
                        System.out.println("connect");
                        establishConnection(key);
                    } else if (key.isReadable()) {
                        System.out.println("read");
                        readData(key);
                    } else if (key.isWritable()) {
                        System.out.println("write");
                        writeData(key);
                    }
                }
            } catch (IOException | ClassNotFoundException e) {
                e.printStackTrace();
            }
        }
        System.out.println("Broke loop");
    }

    private void writeData(SelectionKey key) throws IOException {
        synchronized (pendingDataBundleQue) {
            SocketChannel connection = (SocketChannel) key.channel();
            for (DataBundle bundle : pendingDataBundleQue) {
                System.out.println("sent packet");
                connection.write(bundle.getBuffer());
            }
            pendingDataBundleQue.clear();
            if (pendingDataBundleQue.isEmpty()) {
                Logger.write("All packets sent.", LogType.CLIENT);
                connection.keyFor(selector).interestOps(SelectionKey.OP_READ);
            }
        }
    }

    private void readData(SelectionKey key) throws IOException, ClassNotFoundException {
        buffer.clear();
        int byteCount;
        try {
            byteCount = connection.read(buffer);
        } catch (IOException e) {
            Logger.writeException("Connenction closed.", LogType.CLIENT);
            connection.close();
            key.cancel();
            return;
        }
        if (byteCount == -1) {
            Logger.writeException("Connection error. Attempting to terminate connection.", LogType.CLIENT);
            key.channel().close();
            key.cancel();
        }
        Engine.getInstance().getPacketProcessor().processData(buffer);
    }

    private void establishConnection(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        try {
            if (channel.finishConnect()) {
                Logger.write("Connection established.", LogType.CLIENT);
                key.interestOps(SelectionKey.OP_READ);
            }
        } catch (IOException e) {
            Logger.write("Failed to establish connection.", LogType.CLIENT);
            key.channel().close();
            key.cancel();
            return;
        }
    }

}

处理所有服务器网络的服务器类在这里。

    package org.baiocchi.enigma.server.network;

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.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.Iterator;

import org.baiocchi.enigma.server.Engine;
import org.baiocchi.enigma.server.databundle.DataBundle;
import org.baiocchi.enigma.server.ui.components.logger.LogType;
import org.baiocchi.enigma.server.ui.components.logger.Logger;

public class Server extends Thread {

    private boolean running;
    private final int port;
    private ServerSocketChannel server;
    private final ByteBuffer buffer;
    private Selector selector;
    private ArrayList<DataBundle> pendingDataBundleQue;

    public Server(int port) {
        this.port = port;
        buffer = ByteBuffer.allocate(8192);
        pendingDataBundleQue = new ArrayList<DataBundle>();
        try {
            server = ServerSocketChannel.open().bind(new InetSocketAddress("localhost", port));
            server.configureBlocking(false);
            selector = SelectorProvider.provider().openSelector();
            server.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void start() {
        running = true;
        super.start();
    }

    public void terminateConnection(SocketChannel channel) {
        SelectionKey key = channel.keyFor(selector);
        try {
            key.channel().close();
            key.cancel();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void addToPendingPacketQue(DataBundle bundle) {
        synchronized (pendingDataBundleQue) {
            pendingDataBundleQue.add(bundle);
        }
    }

    @Override
    public void run() {
        while (running) {
            try {
                synchronized (pendingDataBundleQue) {
                    if (!pendingDataBundleQue.isEmpty()) {
                        for (DataBundle bundle : pendingDataBundleQue) {
                            SelectionKey key = bundle.getChannel().keyFor(selector);
                            key.interestOps(SelectionKey.OP_WRITE);
                        }
                    }
                }
                selector.select();
                Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
                while (selectedKeys.hasNext()) {
                    SelectionKey key = (SelectionKey) selectedKeys.next();
                    selectedKeys.remove();
                    if (!key.isValid()) {
                        continue;
                    } else if (key.isAcceptable()) {
                        acceptConnection(key);
                    } else if (key.isReadable()) {
                        readData(key);
                    } else if (key.isWritable()) {
                        writeData(key);
                    }
                }
            } catch (IOException | ClassNotFoundException e) {
                Logger.writeException("Internal server error.", LogType.SERVER);
                Logger.writeException(e.getMessage(), LogType.SERVER);
            }
        }
    }

    private void writeData(SelectionKey key) throws IOException {
        DataBundle bundle = null;
        for (DataBundle b : pendingDataBundleQue) {
            if (b.getChannel().equals((SocketChannel) key.channel())) {
                bundle = b;
                break;
            }
        }
        if (bundle == null) {
            Logger.writeException("Couldn't find out bound packet in list.", LogType.SERVER);
            return;
        }
        SocketChannel connection = bundle.getChannel();
        connection.write(bundle.getBuffer());
        connection.keyFor(selector).interestOps(SelectionKey.OP_READ);
        pendingDataBundleQue.remove(bundle);
    }

    private void readData(SelectionKey key) throws IOException, ClassNotFoundException {
        SocketChannel channel = (SocketChannel) key.channel();
        buffer.clear();
        int byteCount;
        try {
            byteCount = channel.read(buffer);
        } catch (IOException e) {
            Logger.writeException("Connenction terminated.", LogType.SERVER);
            channel.close();
            key.cancel();
            return;
        }
        if (byteCount == -1) {
            Logger.writeException("Connection error. Terminating connection.", LogType.SERVER);
            key.channel().close();
            key.cancel();
            return;
        }
        Engine.getInstance().getPacketProcessor().processData(buffer, channel);
    }

    private void acceptConnection(SelectionKey key) throws IOException {
        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
        SocketChannel connection = channel.accept();
        connection.configureBlocking(false);
        connection.register(selector, SelectionKey.OP_READ);
        Logger.write("Connection established.", LogType.SERVER);
    }

}

提前谢谢！

Answer 1

这里没有证据表明客户端已经断开连接。如果有，这些块中的一个会在服务器上执行:

    try {
        byteCount = channel.read(buffer);
    } catch (IOException e) {
        Logger.writeException("Connenction terminated.", LogType.SERVER);
        channel.close();
        key.cancel();
        return;
    }
    if (byteCount == -1) {
        Logger.writeException("Connection error. Terminating connection.", LogType.SERVER);
        key.channel().close();
        key.cancel();
        return;
    }

我的结论是客户端根本没有断开连接。

我注意到你有一个名为 Logger.writeException（） 的方法，你从来没有用它来记录异常，并且你的日志消息是从前到前的：catch （IOException） 表示连接错误，您应该记录实际的异常，readBytes == -1 表示“连接终止”，这不是错误。

我还注意到客户端中相应代码中缺少返回。

注意关闭频道会取消按键。不需要自己取消。