java - 如何使用多个线程而不是顺序调用方法

Question

我的两个类中有一个处理方法，它接受字符串映射。

在下面的代码中，我使用了一个 for 循环，它将一一(顺序)调用我的两个类中的 process 方法，这很好。

for (ModuleRegistration.ModulesHolderEntry entry : ModuleRegistration.getInstance()) {
    final Map<String, String> response = entry.getPlugin().process(outputs);

    System.out.println(response);
}

但是有什么办法可以为此启动两个线程吗？一个线程将调用我的类之一的进程方法，第二个线程将调用我的第二个类中的进程方法？然后在得到每个线程的响应后，我想写入数据库。这意味着每个线程都会写入数据库。

每个线程也应该有超时功能。我们将为每个线程等待指定的时间，这意味着如果其中一个进程方法在一定时间内没有返回，那么它将超时。

这可以在我的用例中实现吗？如果是，任何人都可以给我提供一个如何执行此操作的示例吗？谢谢。

对此的任何帮助将不胜感激。

Answer 1

您可以创建 ExecutorService根据需要分配多少个线程来运行，例如

ExecutorService executor = Executors.newFixedThreadPool(2)

现在在 for 循环中你会做类似的事情

for (ModuleRegistration.ModulesHolderEntry entry : ModuleRegistration.getInstance()) {
    executor.submit(new Runnable () {
        public void run() {
            final Map<String, String> response = entry.getPlugin().process(outputs);
            // write to database
            System.out.println(response);
        }
    }
}

您可能还希望有一个单独的线程来处理所有数据库写入 - 您的可运行对象将通过 BlockingQueue 或类似的方式将结果发送给它

// Three threads: one thread for the database writer, two threads for the plugin processors
final ExecutorService executor = Executors.newFixedThreadPool(3);

final BlockingQueue<Map<String, String>> queue = new LikedBlockingQueue<>();

Future future = executor.submit(new Runnable () {
    public void run() {
        Map<String, String> map;
        try {
            while(true) {
                // blocks until a map is available in the queue, or until interrupted
                map = queue.take();
                // write map to database
            }
        } catch (InterruptedException ex) {
            // IF we're catching InterruptedException then this means that future.cancel(true)
            // was called, which means that the plugin processors are finished;
            // process the rest of the queue and then exit
            while((map = queue.poll()) != null) {
                // write map to database
            }
        }
    }
}

for (ModuleRegistration.ModulesHolderEntry entry : ModuleRegistration.getInstance()) {
    executor.submit(new Runnable () {
        public void run() {
            final Map<String, String> response = entry.getPlugin().process(outputs);
            // put the response map in the queue for the database to read
            queue.offer(response);
        }
    }
}

// this interrupts the database thread, which sends it into its catch block
// where it processes the rest of the queue and exits
future.cancel(true); // interrupt database thread

// wait for the threads to finish
executor.awaitTermination(5, TimeUnit.MINUTES);