java - 单核 Java 11 上的并发 IO

Question

在使用像 Erlang 和其他具有轻量级并发进程的语言之后，我发现很难理解它是如何转化为 Java 的。鉴于我使用单核机器，有没有办法执行多个并发 IO 绑定(bind)操作(http)？
我发现如下ExecutorService和 CompletableFuture .我遇到的问题是它们基于线程池。默认线程池使用 core# - 1，在我使用的单核机器上，它没有并发。解决方案是否只是提供自定义 Executor线程数更多？或者在 Java 的单核机器上是否有更惯用的方式来实现 IO 绑定(bind)并发？
我在具有单核的 AWS Lambda 上运行此代码。

Answer 1

" 默认线程池使用 core# - 1，在我使用的单核机器上，它没有并发。 "- 为什么？并发程序可以很好地在单核机器上运行。它与并行性无关。
当一个 Java 线程在等待 I/O 时，内核的调度器会将它移到等待队列中，而其他一些需要 CPU 时间的线程将会运行。因此，您可以创建一个包含任意数量线程的线程池，调度程序将负责并发处理。这即使在单核机器上也能正常工作。
这里唯一的限制是您将创建的线程数。线程的默认堆栈大小因/w而异512K至1M .所以这不能很好地扩展，并且在某些时候，你会用完线程。在我的系统上，我可以创建大约 5k 个。像 Go 这样的语言通过在有限数量的内核线程上多路复用多个 goroutine 来管理这一点。这需要 Go 运行时进行调度。
如果您想缓解这种情况，您应该查看 NIO .我编写了一个快速程序，您可以使用它来找出以这种方式实际支持的并发连接数。这应该在导入后按原样运行:

public class ConcurrentBlockingServer {

  private ExecutorService pool = Executors.newCachedThreadPool();

  public static void main(String[] args) {
    ConcurrentBlockingServer bs = new ConcurrentBlockingServer();
    try {
      bs.listen();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  private void listen() throws IOException {
    int connectionId = 0;
    ServerSocket ss = new ServerSocket(8080);
    while (true) {
      Socket s = ss.accept(); // blocking call, never null
      System.out.println("Connection: " + (++connectionId));
      process(s);
    }
  }

  private void process(Socket s) {
    Runnable task =
        () -> {
          try (InputStream is = s.getInputStream();
              OutputStream os = s.getOutputStream()) {
            int data;
            // -1 is EOF, .read() is blocking
            while (-1 != (data = is.read())) {
              os.write(flipCase(data));
              os.flush();
            }
          } catch (IOException e) {
            e.printStackTrace();
          }
        };
    pool.submit(task);
  }

  private int flipCase(int input) {
    if (input >= 65 && input <= 90) {
      return input + 32;
    } else if (input >= 97 && input <= 122) {
      return input - 32;
    } else {
      return input;
    }
  }
}

运行这个程序，看看你能建立多少个连接。

public class RogueClient {

  private static long noClients = 9000;

  public static void main(String[] args) {
    for (int i = 0; i < noClients; i++) {
      try {
        new Socket("localhost", 8080);
        System.out.println("Connection No: " + i);
      } catch (IOException e) {
        System.err.println("Exception: " + e.getMessage() + ", for connection: " + i);
      }
    }
    try {
      Thread.sleep(Integer.MAX_VALUE);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

编辑:池大小应取决于程序的性质。如果它是一个 I/O 绑定(bind)任务，您可以继续创建许多线程。但是对于 CPU 密集型程序，线程数应该等于内核数。