如何每秒执行数以万计的异步操作，每个操作都有三秒的等待时间？

Question

我需要在一个应用程序中每秒执行数万个 IO 绑定操作，每个操作都有 3 秒的延迟。在 node.js 中，这可以通过以下方式轻松完成：

var events = require('events');
var eventEmitter = new events.EventEmitter();

var count = 0;

var connectHandler = function connected() {
    setTimeout(function() {
        count++;
        console.log('task complete ' + count);
    }, 3000);
}

for (var i = 0; i < 30000; i++) {
    eventEmitter.on('mockedIOTask', connectHandler);
}

eventEmitter.emit('mockedIOTask');

如何在 Java 应用程序中实现这一点？我尝试了以下方法，但它一次只执行 3 个任务（原因很明显）：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;

public class Main {
    public static void main(String[] args) {
        //ExecutorService threadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 30000; i++) {
            final int count = i;
            ForkJoinPool.commonPool().submit(() -> {
                try {
                    Thread.sleep(3000);
                    System.out.println("task complete" + count);
                } catch (InterruptedException e) {
                }
            });
        }
        // Keep the program alive
        try {
            Thread.sleep(50000);
        } catch (InterruptedException e) {
        }
    }
} 

Answer 1

您可以使用 ScheduledExecutorService 在 3 秒后执行 30K 任务

ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor();
List<Future<?>> tasks = new ArrayList<>();
for (int i = 0; i < 30000; i++) {
    final int count = i;
    tasks.add(ses.submit(() -> System.out.println("task complete" + count)));
}
for (Future f : tasks) f.get();
ses.shutdown();

如果你想添加 IO 绑定操作，我会先从阻塞操作开始，然后再迁移到使用 netty 之类的库来处理非阻塞 IO。