关闭套接字唤醒选择器

Question

我们编写了一个以这种方式工作的传入反应器：

1. 打开一个选择器
2. 打开服务器套接字通道
3. 启动一个选择循环，其中： ServerSocketChannel 接受新的 SocketChannel 进入循环，每个 SocketChannel 读取数据并将其传输给工作人员。

reactor 的关闭过程是遍历selector.keys() 并为它们中的每一个关闭相应的通道并取消密钥。

我们为关机程序编写了以下单元测试：

1. 打开一个运行选择循环的反应器线程。
2. 打开几个发件人线程。每个都打开一个到反应器的套接字并读取。
3. 读取会阻塞，直到它变为 -1（意味着反应器关闭了套接字）。
4. 读取返回 -1 后，发送方关闭套接字并完成。

测试导致 ConcurrentModificationException 指向循环遍历套接字并关闭它们（在主线程上下文中）。

我们的假设是，当 Sender 读取方法得到 -1 时，它会关闭套接字并以某种方式唤醒选择器选择方法，然后选择器访问其键集，该键集由关闭循环迭代，因此出现异常。

我们通过使用选择器的所有键创建一个新列表来解决这个问题。通过迭代这个列表来取消这些键可以防止两个对象修改同一个键的集合。

我们的问题是：

1. 我们的假设正确吗？当客户端套接字调用 close 方法时，它真的会唤醒选择器吗？
2. 创建新列表是合适的解决方案还是只是一种变通方法？

编辑：添加了一些代码 sn-ps 以进行说明 （我们尽量缩小代码范围）

传入反应器：

public boolean startAcceptingIncomingData() {
    Selector selector = Selector.open();
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open());
    serverSocketChannel.bind(new InetSocketAddress(incomingConnectionsPort));
    serverSocketChannel.configureBlocking(false);
    SelectionKey acceptorSelectionKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    acceptorSelectionKey.attach((Worker) this::acceptIncomingSocket);
    startSelectionLoop(selector);
    return true;
  }

private boolean acceptIncomingSocket() {
    try {
        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        SelectionKey selectionKey = socketChannel.register(selector, SelectionKey.OP_READ);
        selectionKey.attach(new WorkerImpl() /*Responsible for reading data and tranferring it into a parsing thread*/);
        return true;
    } catch (IOException e) {
      return false;
    }
  }

private void startSelectionLoop(Selector selector) {
    shouldLoop = true;
    while (shouldLoop) {
      try {
        selector.select();
        Set<SelectionKey> selectedKeys = selector.selectedKeys();
        if (!shouldLoop) {
          break;
        }
        selectedKeys.forEach((key) -> {
          boolean workSuccess = ((Worker) key.attachment()).work();
          if (!workSuccess) {
              key.channel().close();
              key.cancel();
          }
        });
        selectedKeys.clear();
      } catch (ClosedSelectorException ignore) {
      }
    }
  }

public void shutDown() {
    shouldLoop = false;
    selector.keys().forEach(key -> { /***EXCEPTION - This is where the exception points to (this is line 129) ***/
        key.channel().close();
        key.cancel();
    });
    try {
      selector.close();
    } catch (IOException e) {
    }
  }

单元测试：

   @Test
  public void testMaximumConnectionsWithMultipleThreads() {
    final int PORT = 24785;
    final int MAXINUM_CONNECTIONS = 10;

    IncomingReactor incomingReactor = new IncomingReactor(PORT);
    Callable<Boolean> acceptorThread = () -> {
      incomingReactor.startAcceptingIncomingData();
      return true;
    };

    ExecutorService threadPool = Executors.newFixedThreadPool(MAXIMUM_CONNECTIONS + 1);
    Future<Boolean> acceptorFuture = threadPool.submit(acceptorThread);

    List<Future<Boolean>> futureList = new ArrayList<>(MAXIMUM_CONNECTIONS);
    for (int currentSenderThread = 0; currentSenderThread < MAXIMUM_CONNECTIONS; currentSenderThread++) {
      Future<Boolean> senderFuture = threadPool.submit(() -> {
        Socket socket = new Socket(LOCALHOST, PORT);
        int bytesRead = socket.getInputStream().read();
        if (bytesRead == -1) { //The server has closed us
          socket.close();
          return true;
        } else {
          throw new RuntimeException("Got real bytes from socket.");
        }
      });
      futureList.add((senderFuture));
    }

    Thread.sleep(1000); //We should wait to ensure that the evil socket is indeed the last one that connects and the one that will be closed
    Socket shouldCloseSocket = new Socket(LOCALHOST, PORT);
    Assert.assertEquals(shouldCloseSocket.getInputStream().read(), -1);
    shouldCloseSocket.close();
    incomingReactor.shutDown();
    for (Future<Boolean> senderFuture : futureList) {
      senderFuture.get();
    }
    acceptorFuture.get();
    threadPool.shutdown();
  }

例外：

java.util.ConcurrentModificationException
    at java.util.HashMap$HashIterator.nextNode(HashMap.java:1437)
    at java.util.HashMap$KeyIterator.next(HashMap.java:1461)
    at java.lang.Iterable.forEach(Iterable.java:74)
    at java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1080)
    at mypackage.IncomingReactor.shutDown(IncomingReactor.java:129)
    at mypackage.tests.TestIncomingReactor.testMaximumConnectionsWithMultipleThreads(TestIncomingReactor.java:177)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:85)
    at org.testng.internal.Invoker.invokeMethod(Invoker.java:659)
    at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:845)
    at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1153)
    at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:125)
    at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:108)
    at org.testng.TestRunner.privateRun(TestRunner.java:771)
    at org.testng.TestRunner.run(TestRunner.java:621)
    at org.testng.SuiteRunner.runTest(SuiteRunner.java:357)
    at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:352)
    at org.testng.SuiteRunner.privateRun(SuiteRunner.java:310)
    at org.testng.SuiteRunner.run(SuiteRunner.java:259)
    at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
    at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:86)
    at org.testng.TestNG.runSuitesSequentially(TestNG.java:1199)
    at org.testng.TestNG.runSuitesLocally(TestNG.java:1124)
    at org.testng.TestNG.run(TestNG.java:1032)
    at org.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:74)
    at org.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:124)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)

Answer 1

reactor 的关闭过程是遍历 selector.keys() 并为它们中的每一个关闭相应的通道并取消键。

它应该从停止选择器循环开始。 NB 关闭通道会取消密钥。您不必自己取消。

我们为关机程序编写了以下单元测试：

打开一个运行选择循环的反应器线程。 打开几个发件人线程。每个都打开一个到反应器的套接字并读取。 读取会阻塞，直到它得到 -1（意味着反应器关闭了套接字）。

反应器关闭了其接受的套接字。您的客户端套接字保持打开状态。

读取返回-1后，发送方关闭套接字并结束。

我希望这意味着发送方关闭了它的客户端套接字。

测试导致ConcurrentModificationException 指向循环遍历套接字并关闭它们（在主线程上下文中）。

真的吗？我在您的问题中没有看到任何堆栈跟踪。

我们的假设是，当 Sender 读取方法得到 -1 时，它会关闭套接字并以某种方式唤醒选择器选择方法

除非反应堆没有关闭通道，否则不可能，在这种情况下，您不会从读取等中得到 -1。

选择器随后访问了它的键集，该键集由关闭循环迭代，因此出现异常。

该异常是由于在迭代过程中修改了键集引起的。服务器代码中的错误。

我们通过使用选择器的所有键创建一个新列表来解决这个问题。通过迭代这个列表来取消这些键可以防止两个对象修改同一个键的集合。

您需要解决实际问题，为此您需要发布实际代码。

我们的问题是：

我们的假设正确吗？当客户端套接字调用 close 方法时——它真的唤醒了选择器吗？

除非选择器端通道仍然打开，否则不会。

创建新列表是合适的解决方案还是只是一种变通方法？

对于您尚未发现的问题，这只是一个令人讨厌的解决方法。

Answer 2

您不能从 for 循环内部修改 selector.keys() Set<SelectionKey>，因为 Set 不能同时修改。 （调用channel.close() 将修改Set 从循环内部读取Set）

https://docs.oracle.com/javase/7/docs/api/java/util/HashSet.html

这个类的迭代器方法返回的迭代器是快速失败的： 如果集合在迭代器创建后的任何时间被修改，在 任何方式，除了通过迭代器自己的删除方法，迭代器 引发 ConcurrentModificationException。于是，面对 并发修改，迭代器快速而干净地失败， 而不是冒着随意的、非确定性的行为的风险 未来时间未定。

SelectionKey[] keys = selector.keys().toArray(new SelectionKey[0]);

for( SelectionKey k : keys )
{
    try
    {
        k.channel().close();
    }
    catch(Throwable x )
    {
        // print
    }
}

try
{
    selector.close();
}
catch(IoException e )
{
    // print
}