java - executorService多线程池的性能

Question

我正在使用 Java 的并发库 ExecutorService 来运行我的任务。写入数据库的阈值是200 QPS，但是这个程序在15个线程下只能达到20 QPS。我尝试了5、10、20、30个线程，它们甚至比15个线程还慢。这是代码:

ExecutorService executor = Executors.newFixedThreadPool(15);
List<Callable<Object>> todos = new ArrayList<>();

for (final int id : ids) {
    todos.add(Executors.callable(() -> {
        try {
            TestObject test = testServiceClient.callRemoteService();
    SaveToDatabase();
        } catch (Exception ex) {}
    }));
}
try {
    executor.invokeAll(todos);
} catch (InterruptedException ex) {} 
executor.shutdown();

1)我检查了运行这个程序的linux服务器的CPU使用率，使用率是90%和60%(它有4个CPU)。内存使用率仅为20%。所以CPU和内存还是没问题的。数据库服务器的CPU 使用率较低(大约20%)。什么会阻止速度达到 200 QPS？也许这个服务调用:testServiceClient.callRemoteService()？我检查了该调用的服务器配置，它允许每秒进行大量调用。

2) 如果ids中id的数量超过50000，使用invokeAll是个好主意吗？我们是否应该将其分成较小的批处理，例如每批处理 5000 个？

Answer 1

此代码中没有任何内容可以阻止此查询速率，除了重复创建和销毁线程池非常昂贵之外。我建议使用 Streams API，它不仅更简单，而且重用内置线程池

int[] ids = ....
IntStream.of(ids).parallel()
                 .forEach(id -> testServiceClient.callRemoteService(id));

这是使用简单服务的基准。主要开销是创建连接的延迟。

public static void main(String[] args) throws IOException {
    ServerSocket ss = new ServerSocket(0);
    Thread service = new Thread(() -> {
        try {
            for (; ; ) {
                try (Socket s = ss.accept()) {
                    s.getOutputStream().write(s.getInputStream().read());
                }
            }
        } catch (Throwable t) {
            t.printStackTrace();
        }
    });
    service.setDaemon(true);
    service.start();

    for (int t = 0; t < 5; t++) {
        long start = System.nanoTime();
        int[] ids = new int[5000];
        IntStream.of(ids).parallel().forEach(id -> {
            try {
                Socket s = new Socket("localhost", ss.getLocalPort());
                s.getOutputStream().write(id);
                s.getInputStream().read();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        long time = System.nanoTime() - start;
        System.out.println("Throughput " + (int) (ids.length * 1e9 / time) + " connects/sec");
    }
}

打印

Throughput 12491 connects/sec
Throughput 13138 connects/sec
Throughput 15148 connects/sec
Throughput 14602 connects/sec
Throughput 15807 connects/sec

正如 @grzegorz-piwowarek 提到的那样，使用 ExecutorService 会更好。

    ExecutorService es = Executors.newFixedThreadPool(8);
    for (int t = 0; t < 5; t++) {
        long start = System.nanoTime();
        int[] ids = new int[5000];
        List<Future> futures = new ArrayList<>(ids.length);
        for (int id : ids) {
            futures.add(es.submit(() -> {
                try {
                    Socket s = new Socket("localhost", ss.getLocalPort());
                    s.getOutputStream().write(id);
                    s.getInputStream().read();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }));
        }
        for (Future future : futures) {
            future.get();
        }
        long time = System.nanoTime() - start;
        System.out.println("Throughput " + (int) (ids.length * 1e9 / time) + " connects/sec");
    }
    es.shutdown();

在这种情况下会产生大致相同的结果。