为什么我的nio客户端到服务器的连接会中断？

xriantvc  于 2021-07-03  发布在  Java
关注(0)|答案(1)|浏览(495)

我目前有一个桌面服务器，它是在非阻塞模式下使用niojava库编写的。您可以在这里找到完整的服务器项目。我还为测试目的创建了一个非阻塞nio客户机。你可以在这里找到那个项目。最后，该服务器应该用于android即时消息应用程序。我将客户端和服务器都建立在发送“数据包”进行通信的思想上。我的意思是，包类的引用被打包到字节缓冲区中，并通过套接字通道发送。然后在另一端反序列化并执行它们。
我当前的问题是：我的测试客户机在连接到服务器时似乎会断开连接。我知道问题是客户端的，因为当我使用telnet通过命令行连接到服务器时，连接不会断开。奇怪的是，服务器正在打印与它的连接已经建立，但是当客户端断开连接时，它从不说连接已经丢失/终止。
这是处理所有nio网络的客户机类。。。

  1. import java.io.IOException;
  2. import java.net.InetSocketAddress;
  3. import java.nio.ByteBuffer;
  4. import java.nio.channels.SelectionKey;
  5. import java.nio.channels.Selector;
  6. import java.nio.channels.SocketChannel;
  7. import java.nio.channels.spi.SelectorProvider;
  8. import java.util.Iterator;
  9. import java.util.LinkedList;
  10. import java.util.List;
  12. import org.baiocchi.enigma.client.test.Engine;
  13. import org.baiocchi.enigma.client.test.databundle.DataBundle;
  14. import org.baiocchi.enigma.client.test.packet.Packet;
  15. import org.baiocchi.enigma.client.test.ui.LogType;
  16. import org.baiocchi.enigma.client.test.ui.Logger;
  18. public class Client extends Thread {
  20.     private boolean running;
  21.     private final int port;
  22.     private SocketChannel connection;
  23.     private final ByteBuffer buffer;
  24.     private Selector selector;
  25.     private List<DataBundle> pendingDataBundleQue;
  27.     public Client(int port) {
  28.         this.port = port;
  29.         pendingDataBundleQue = new LinkedList<DataBundle>();
  30.         buffer = ByteBuffer.allocate(8192);
  31.         try {
  32.             selector = initiateSelector();
  33.             connection = initiateConnection();
  34.         } catch (IOException e) {
  35.             e.printStackTrace();
  36.         }
  37.     }
  39.     private Selector initiateSelector() throws IOException {
  40.         return SelectorProvider.provider().openSelector();
  41.     }
  43.     private SocketChannel initiateConnection() throws IOException {
  44.         SocketChannel connection = SocketChannel.open();
  45.         connection.configureBlocking(false);
  46.         connection.connect(new InetSocketAddress("localhost", port));
  47.         connection.register(selector, SelectionKey.OP_CONNECT);
  48.         return connection;
  49.     }
  51.     public SocketChannel getConnection() {
  52.         return connection;
  53.     }
  55.     @Override
  56.     public void start() {
  57.         running = true;
  58.         super.start();
  59.     }
  61.     public void addToPendingDataBundleQue(DataBundle bundle) {
  62.         synchronized (pendingDataBundleQue) {
  63.             pendingDataBundleQue.add(bundle);
  64.         }
  65.     }
  67.     @Override
  68.     public void run() {
  69.         while (running) {
  70.             System.out.println("loop");
  71.             try {
  72.                 synchronized (pendingDataBundleQue) {
  73.                     System.out.println("Checking for que changes.");
  74.                     if (!pendingDataBundleQue.isEmpty()) {
  75.                         System.out.println("Found que change.");
  76.                         SelectionKey key = connection.keyFor(selector);
  77.                         key.interestOps(SelectionKey.OP_WRITE);
  78.                     }
  79.                 }
  80.                 System.out.println("Selecting keys");
  81.                 selector.select();
  82.                 System.out.println("Creating selected keys list.");
  83.                 Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
  84.                 while (selectedKeys.hasNext()) {
  85.                     System.out.println("scrolling through list");
  86.                     SelectionKey key = (SelectionKey) selectedKeys.next();
  87.                     selectedKeys.remove();
  88.                     if (!key.isValid()) {
  89.                         System.out.println("invalid");
  90.                         continue;
  91.                     } else if (key.isConnectable()) {
  92.                         System.out.println("connect");
  93.                         establishConnection(key);
  94.                     } else if (key.isReadable()) {
  95.                         System.out.println("read");
  96.                         readData(key);
  97.                     } else if (key.isWritable()) {
  98.                         System.out.println("write");
  99.                         writeData(key);
  100.                     }
  101.                 }
  102.             } catch (IOException | ClassNotFoundException e) {
  103.                 e.printStackTrace();
  104.             }
  105.         }
  106.         System.out.println("Broke loop");
  107.     }
  109.     private void writeData(SelectionKey key) throws IOException {
  110.         synchronized (pendingDataBundleQue) {
  111.             SocketChannel connection = (SocketChannel) key.channel();
  112.             for (DataBundle bundle : pendingDataBundleQue) {
  113.                 System.out.println("sent packet");
  114.                 connection.write(bundle.getBuffer());
  115.             }
  116.             pendingDataBundleQue.clear();
  117.             if (pendingDataBundleQue.isEmpty()) {
  118.                 Logger.write("All packets sent.", LogType.CLIENT);
  119.                 connection.keyFor(selector).interestOps(SelectionKey.OP_READ);
  120.             }
  121.         }
  122.     }
  124.     private void readData(SelectionKey key) throws IOException, ClassNotFoundException {
  125.         buffer.clear();
  126.         int byteCount;
  127.         try {
  128.             byteCount = connection.read(buffer);
  129.         } catch (IOException e) {
  130.             Logger.writeException("Connenction closed.", LogType.CLIENT);
  131.             connection.close();
  132.             key.cancel();
  133.             return;
  134.         }
  135.         if (byteCount == -1) {
  136.             Logger.writeException("Connection error. Attempting to terminate connection.", LogType.CLIENT);
  137.             key.channel().close();
  138.             key.cancel();
  139.         }
  140.         Engine.getInstance().getPacketProcessor().processData(buffer);
  141.     }
  143.     private void establishConnection(SelectionKey key) throws IOException {
  144.         SocketChannel channel = (SocketChannel) key.channel();
  145.         try {
  146.             if (channel.finishConnect()) {
  147.                 Logger.write("Connection established.", LogType.CLIENT);
  148.                 key.interestOps(SelectionKey.OP_READ);
  149.             }
  150.         } catch (IOException e) {
  151.             Logger.write("Failed to establish connection.", LogType.CLIENT);
  152.             key.channel().close();
  153.             key.cancel();
  154.             return;
  155.         }
  156.     }
  158. }

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

  1. package org.baiocchi.enigma.server.network;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.ServerSocketChannel;
  9. import java.nio.channels.SocketChannel;
  10. import java.nio.channels.spi.SelectorProvider;
  11. import java.util.ArrayList;
  12. import java.util.Iterator;
  14. import org.baiocchi.enigma.server.Engine;
  15. import org.baiocchi.enigma.server.databundle.DataBundle;
  16. import org.baiocchi.enigma.server.ui.components.logger.LogType;
  17. import org.baiocchi.enigma.server.ui.components.logger.Logger;
  19. public class Server extends Thread {
  21.     private boolean running;
  22.     private final int port;
  23.     private ServerSocketChannel server;
  24.     private final ByteBuffer buffer;
  25.     private Selector selector;
  26.     private ArrayList<DataBundle> pendingDataBundleQue;
  28.     public Server(int port) {
  29.         this.port = port;
  30.         buffer = ByteBuffer.allocate(8192);
  31.         pendingDataBundleQue = new ArrayList<DataBundle>();
  32.         try {
  33.             server = ServerSocketChannel.open().bind(new InetSocketAddress("localhost", port));
  34.             server.configureBlocking(false);
  35.             selector = SelectorProvider.provider().openSelector();
  36.             server.register(selector, SelectionKey.OP_ACCEPT);
  37.         } catch (IOException e) {
  38.             e.printStackTrace();
  39.         }
  40.     }
  42.     @Override
  43.     public void start() {
  44.         running = true;
  45.         super.start();
  46.     }
  48.     public void terminateConnection(SocketChannel channel) {
  49.         SelectionKey key = channel.keyFor(selector);
  50.         try {
  51.             key.channel().close();
  52.             key.cancel();
  53.         } catch (IOException e) {
  54.             e.printStackTrace();
  55.         }
  56.     }
  58.     public void addToPendingPacketQue(DataBundle bundle) {
  59.         synchronized (pendingDataBundleQue) {
  60.             pendingDataBundleQue.add(bundle);
  61.         }
  62.     }
  64.     @Override
  65.     public void run() {
  66.         while (running) {
  67.             try {
  68.                 synchronized (pendingDataBundleQue) {
  69.                     if (!pendingDataBundleQue.isEmpty()) {
  70.                         for (DataBundle bundle : pendingDataBundleQue) {
  71.                             SelectionKey key = bundle.getChannel().keyFor(selector);
  72.                             key.interestOps(SelectionKey.OP_WRITE);
  73.                         }
  74.                     }
  75.                 }
  76.                 selector.select();
  77.                 Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
  78.                 while (selectedKeys.hasNext()) {
  79.                     SelectionKey key = (SelectionKey) selectedKeys.next();
  80.                     selectedKeys.remove();
  81.                     if (!key.isValid()) {
  82.                         continue;
  83.                     } else if (key.isAcceptable()) {
  84.                         acceptConnection(key);
  85.                     } else if (key.isReadable()) {
  86.                         readData(key);
  87.                     } else if (key.isWritable()) {
  88.                         writeData(key);
  89.                     }
  90.                 }
  91.             } catch (IOException | ClassNotFoundException e) {
  92.                 Logger.writeException("Internal server error.", LogType.SERVER);
  93.                 Logger.writeException(e.getMessage(), LogType.SERVER);
  94.             }
  95.         }
  96.     }
  98.     private void writeData(SelectionKey key) throws IOException {
  99.         DataBundle bundle = null;
  100.         for (DataBundle b : pendingDataBundleQue) {
  101.             if (b.getChannel().equals((SocketChannel) key.channel())) {
  102.                 bundle = b;
  103.                 break;
  104.             }
  105.         }
  106.         if (bundle == null) {
  107.             Logger.writeException("Couldn't find out bound packet in list.", LogType.SERVER);
  108.             return;
  109.         }
  110.         SocketChannel connection = bundle.getChannel();
  111.         connection.write(bundle.getBuffer());
  112.         connection.keyFor(selector).interestOps(SelectionKey.OP_READ);
  113.         pendingDataBundleQue.remove(bundle);
  114.     }
  116.     private void readData(SelectionKey key) throws IOException, ClassNotFoundException {
  117.         SocketChannel channel = (SocketChannel) key.channel();
  118.         buffer.clear();
  119.         int byteCount;
  120.         try {
  121.             byteCount = channel.read(buffer);
  122.         } catch (IOException e) {
  123.             Logger.writeException("Connenction terminated.", LogType.SERVER);
  124.             channel.close();
  125.             key.cancel();
  126.             return;
  127.         }
  128.         if (byteCount == -1) {
  129.             Logger.writeException("Connection error. Terminating connection.", LogType.SERVER);
  130.             key.channel().close();
  131.             key.cancel();
  132.             return;
  133.         }
  134.         Engine.getInstance().getPacketProcessor().processData(buffer, channel);
  135.     }
  137.     private void acceptConnection(SelectionKey key) throws IOException {
  138.         ServerSocketChannel channel = (ServerSocketChannel) key.channel();
  139.         SocketChannel connection = channel.accept();
  140.         connection.configureBlocking(false);
  141.         connection.register(selector, SelectionKey.OP_READ);
  142.         Logger.write("Connection established.", LogType.SERVER);
  143.     }
  145. }

提前谢谢！

JavanioserverConnectionClient

1条答案

按热度按时间
cig3rfwq

cig3rfwq1#

这里没有证据表明客户端已断开连接。如果有，其中一个块将在服务器上执行：

  1. try {
  2.         byteCount = channel.read(buffer);
  3.     } catch (IOException e) {
  4.         Logger.writeException("Connenction terminated.", LogType.SERVER);
  5.         channel.close();
  6.         key.cancel();
  7.         return;
  8.     }
  9.     if (byteCount == -1) {
  10.         Logger.writeException("Connection error. Terminating connection.", LogType.SERVER);
  11.         key.channel().close();
  12.         key.cancel();
  13.         return;
  14.     }

我的结论是客户端根本没有断开连接。
我注意到您有一个名为 Logger.writeException() 您从不使用它来记录异常，并且您的日志消息是背对背的： catch (IOException ) 指示一个连接错误，您应该在该错误上记录实际的异常，并且 readBytes == -1 表示“连接已终止”，这不是错误。
我还注意到一个 return 在客户端的相应代码中。
nb关闭频道取消按键。你不需要自己取消。

展开查看全部
赞(0）分享  回复(0）举报  2021-07-03
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com