ForkJoinPool的使用

我在工作中遇到过这样一个问题，是给网站用户发送营销短信。我记得我们网站会员大约有7万个，要给这7万个用户发送营销短信。而短信运营商那边要求每次发送的手机号码最好要少于3000。

那时候的我，没有什么经验，不会使用多线程，将整个程序顺序写下来，大体思路如下：



这样做的后果，整个完成7万条短信的发送，需要10多分钟。最要命的是，当时我们使用nignx，或许是没配置好的原因，当一个程序执行超过6分钟没有完成时，则会自动重新执行。所以很悲剧的是我们给每个会员都发送了两条一模一样的短信。有的会员还好心的告诉我们客服，说是短信发重复了。感觉那段时间压力很大，没法跟公司老总交代了。老总还比较好，念在我是新员工的份上，没有追究责任。

然后怎么办？改呗，通过这件事，我第一次自己使用了多线程，以前只是听说，并没有真正用到，甚至都没有考虑过什么时候用它。

思路很简单，也是将list，3000条分一次，只不过不用主程序去发送短信，而是new Thread()去发送短信。在后来的优化中，又将new Thread()优化成了线程池。

每次公司发送营销短信的时候，我都提心吊胆的。生怕某个线程执行的时候出现什么问题。

后来即使换了公司，我还时常想起这件事情，总是想我的代码还可以在优化。偶然的机会，我接触了ForkJoinPool。若是我还在以前的公司，我一定要用ForkJoinPool去试一下，可不可以。

ForkJoinPool

ForkJoinPool线程池思想：将大任务分解成若干个小任务，当小任务均执行结束后，将任务做一个整合，在网上盗了一张图



使用接口：

RecursiveAction：没有返回值，只是执行任务

RecursiveTask：有返回值，小任务结束后，返回结果。大任务可将小任务返回结果进行整合

用ForkJoinPool完成上述发送短信功能

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeUnit;

public class SendMessageTask extends RecursiveAction{

// 每个“小任务”只最多只给10名用户发送短信
private static final int THRESHOLD = 10;
private int start;
private int end;
List<String> list = null;
// 从start到end的任务
public SendMessageTask(int start, int end,List<String> list)
{
this.start = start;
this.end = end;
this.list = list;
}

@Override
protected void compute() {
// TODO Auto-generated method stub
if(end - start < THRESHOLD){
String mobileno="";
for (int i = start ; i < end ; i++ )
{
//此处做手机号码累加，用于发送给短信运营商
mobileno+=list.get(i)+","
}
System.out.println("给手机号码=="+mobileno+"的用户发送手机短信");
}else{
// 如果当end与start之间的差大于THRESHOLD时，即要发送的数超过10个
// 将大任务分解成两个小任务。
int middle = (start + end) /2;
SendMessageTask left = new SendMessageTask(start, middle,list);
SendMessageTask right = new SendMessageTask(middle, end,list);
// 并行执行两个“小任务”
left.fork();
right.fork();
}

}

public static void main(String[] args) throws InterruptedException {
List<String> list = new ArrayList<String>();
for(int i=1;i<=380;i++){
list.add("i------"+i);//假设此处为手机号码--项目中从数据库中获取
}
ForkJoinPool pool = new ForkJoinPool();
// 提交可分解的PrintTask任务
pool.submit(new SendMessageTask(0 , list.size(),list));
//线程阻塞，等待所有任务完成
pool.awaitTermination(10, TimeUnit.SECONDS);
// 关闭线程池
pool.shutdown();
}

}

我又在网上找了一个有累加结果的示例，如下所示:

//继承RecursiveTask来实现”可分解”的任务

class CalTask extends RecursiveTask<Integer>
{
// 每个“小任务”只最多只累加20个数
private static final int THRESHOLD = 20;
private int arr[];
private int start;
private int end;
// 累加从start到end的数组元素
public CalTask(int[] arr , int start, int end)
{
this.arr = arr;
this.start = start;
this.end = end;
}
@Override
protected Integer compute()
{
int sum = 0;
// 当end与start之间的差小于THRESHOLD时，开始进行实际累加
if(end - start < THRESHOLD)
{
for (int i = start ; i < end ; i++ )
{
sum += arr[i];
}
return sum;
}
else
{
// 如果当end与start之间的差大于THRESHOLD时，即要打印的数超过20个
// 将大任务分解成两个小任务。
int middle = (start + end) /2;
CalTask left = new CalTask(arr , start, middle);
CalTask right = new CalTask(arr , middle, end);
// 并行执行两个“小任务”
left.fork();
right.fork();
// 把两个“小任务”累加的结果合并起来
return left.join() + right.join();
}
}
}
public class Sum
{
public static void main(String[] args)
throws Exception
{
int[] arr = new int[100];
Random rand = new Random();
int total = 0;
// 初始化100个数字元素
for (int i = 0 , len = arr.length; i < len ; i++ )
{
int tmp = rand.nextInt(20);
// 对数组元素赋值，并将数组元素的值添加到total总和中。
total += (arr[i] = tmp);
}
System.out.println(total);

ForkJoinPool pool = new ForkJoinPool();
// 提交可分解的CalTask任务
Future<Integer> future = pool.submit(new CalTask(arr , 0 , arr.length));
System.out.println(future.get());
// 关闭线程池
pool.shutdown();
}
}

我现在并不知道ForkJoinPool具体的实现思路及它是怎么执行我拆分的小任务，但它内部讲究的是线程之间的整体协作。当内部线程A空闲时，它会拿线程B的任务去执行。线程B也会执行自己的任务，只不过两个线程是从任务队列的两端执行，尽量减少冲突。

只有不断的学习，才会有提高，才会变得更加强大。希望过段时间我会发现更好的实现方式。