java - 如何为现在在 for 循环中完成的一系列操作实现 forkjoin

Question

我有一个发件人列表，我必须单独并行发送邮件。目前我正在迭代列表构造正文（因为它对不同的人不同），然后发送它们。我如何使用 forkjoin 来实现这一点。我尝试使用 recusiveAction 但我猜它仅用于递归任务。

互联网上所有可用的示例都是使用 RecursiveAction 实现的。是否有任何其他类可以实现这一点。

Answer 1

ServiceExecutors 很好地解决了这个问题。它们与 Java 一起提供。

import java.util.*;
import java.util.concurrent.*;

public class SendMailExample
{
  public static void main(String[] args) throws Exception
  {
    ExecutorService executor = Executors.newFixedThreadPool(3);

    Collection<Future> futures = new ArrayList<Future>();
    futures.add(executor.submit(new Mailer("thread1")));
    futures.add(executor.submit(new Mailer("thread2")));
    futures.add(executor.submit(new Mailer("thread3")));

    for (Future future : futures)
    {
      future.get();
    }
    executor.shutdown();
  }

  static class Mailer implements Runnable
  {
    private Object message;

    public Mailer(Object message)
    {
      this.message = message;
    }

    public void run()
    {
      System.out.println("Sending message " + String.valueOf(message));
    }
  }
}

Answer 2

我浏览了一个更好的答案：

package Test1;

import java.util.*;
import java.util.concurrent.*;
import static java.util.Arrays.asList;

public class Sums
{
    static class Sum implements Callable<Long>
    {
        private final long from;
        private final long to;
        Sum(long from, long to)
        {
            this.from = from;
            this.to = to;
        }

        @Override
        public Long call()
        {
            long acc = 0;
            if(from == 0)
            {
                try
                {
                    Thread.sleep(5000);
                }
                catch (InterruptedException e)
                {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            System.out.println(from);
            for (long i = from; i <= to; i++)
            {
                acc = acc + i;
            }
            return acc;
        }                
    }

    public static void main(String[] args) throws Exception
    {
        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        List <Future<Long>> results = executor.invokeAll(asList(
        new Sum(0, 10), new Sum(100, 1000), new Sum(10000, 1000000)
        ));
        executor.shutdown();

        for (Future<Long> result : results)
        {
            System.out.println(result.get());
        }
    }    
}

使用此代码，您将能够获得响应以及引发的任何异常。