在使用像 Erlang 和其他具有轻量级并发进程的语言之后,我发现很难理解它是如何转化为 Java 的。鉴于我使用单核机器,有没有办法执行多个并发 IO 绑定(bind)操作(http)?
我发现如下ExecutorService
和 CompletableFuture
.我遇到的问题是它们基于线程池。默认线程池使用 core# - 1,在我使用的单核机器上,它没有并发。解决方案是否只是提供自定义 Executor
线程数更多?或者在 Java 的单核机器上是否有更惯用的方式来实现 IO 绑定(bind)并发?
我在具有单核的 AWS Lambda 上运行此代码。
最佳答案
" 默认线程池使用 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 密集型程序,线程数应该等于内核数。
关于java - 单核 Java 11 上的并发 IO,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/64323108/