注：（本例用submit实现会更简单，可以参见ThreadPoolExecutor execute 和 submit）
昨日上线了我的并发编程改造后的代码，速率提升十分明显，原本需要运行将近30小时的任务缩短到十分之一（我开启了最大十个并发），不过，虽然达到了提升速率的目的，但是结合日志我还是发现了两个潜在Bug。上一篇传送门：ThreadPoolExecutor + CountDownLatch 实际应用
首先，我在线程池中捕获了异常，并且记录日志。
try {
// 模拟耗时
Thread.sleep(Long.valueOf(String.valueOf(new Double(Math.random() * 1000).intValue())));
} catch (InterruptedException e) {
// 实际应用在此记录日志
log.error("ThreadPoolExecutor error", e);
}
接着我在日志中发现的确有报错，这种情况下，我应该判断任务运行结果为失败，然而数据库中最终任务运行状态是成功。再一看代码就明白了，因为我只判断了线程池中子线程的运行状态是否完成，但是没有判断逻辑结果是不是对的，再加上线程内异常被吞了，最后任务的状态肯定是成功的，哪怕所有线程都报错，结果也能输出，只不过是错的。解决方法如下：
1、创建内部类，作为状态标识变量（还可以用其他引用类型变量代替）
static class Flag {
private boolean f;
public Flag() {
f = true;// 初始设为true
}
public boolean isF() {
return f;
}
public void setF(boolean f) {
this.f = f;
}
}
2、初始化Flag
final Flag flag = new Flag();
3、在异常捕获中设置为false
...
catch (Exception e) {
// 如果子线程报错，状态标识为false
flag.setF(false);
}
4、最后判断状态是否为true
if (flag.isF()) {
// 打印计数
System.out.println("结束：" + totalRows.get());
} else {
System.out.println("有子线程报错，结果不准确");
}
另外，我还发现了一个潜在Bug。我的任务核心部分是在while 循环内请求第三方api下载数据，如果第三方返回结果为空，手动break退出循环，但是未在catch中退出。这样就有问题了，一旦循环体内抛出运行时异常，代码未执行到判断是否break时就被异常了，但是catch未执行退出，将可能导致死循环。好比示例代码中：
while (true) {
// 此处发送请求拉取数据，一旦报错，将跳过下面的break 判断，而catch中未退出，将导致死循环
String result = download();
// 判断是否break
if (result == null) {
break;
}
} catch (Exception e) {
e.printStackTrace();
// 未执行退出
}
}
修改后
boolean loop = true;
while (loop) {
String result = download();
// 判断是否break
if (result == null) {
loop = false;
}
} catch (Exception e) {
// 退出循环
loop = false;
}
}
完整示例（while循环内用次数模拟和真实业务逻辑还是有区别的）
package com.yzy.test;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Main1 {
static class Flag {
private boolean f;
public Flag() {
f = true;
}
public boolean isF() {
return f;
}
public void setF(boolean f) {
this.f = f;
}
}
public static void main(String[] args) throws InterruptedException {
// 线程安全的计数器
AtomicInteger totalRows = new AtomicInteger(0);
// 创建线程池，其中核心线程10，也是我期望的最大并发数，最大线程数和队列大小都为30，即我的总任务数
ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 30, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(30));
// 初始化CountDownLatch，大小为30
CountDownLatch countDownLatch = new CountDownLatch(30);
// 记录状态
final Flag flag = new Flag();
// 模拟遍历参数集合
for (int i = 0; i < 30; i++) {
// 往线程池提交任务
executor.execute(new Runnable() {
@Override
public void run() {
int times = 0;
boolean loop = true;
// 模拟数据拉取过程可能需要分页
while (loop) {
// 模拟每个任务需要分页5次
if (times >= 5) {
break;
}
times++;
// 模拟计数
totalRows.incrementAndGet();
try {
// 模拟耗时
Thread.sleep(Long.valueOf(String.valueOf(new Double(Math.random() * 1000).intValue())));
} catch (Exception e) {
// 如果子线程报错，退出循环
loop = false;
// 如果子线程报错，状态标识为false
flag.setF(false);
}
}
// 子线程完成，countDownLatch执行countDown
countDownLatch.countDown();
}
});
// 打印线程池运行状态
System.out.println("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" +
executor.getQueue().size() + "，已执行结束的任务数目：" + executor.getCompletedTaskCount());
}
// 标记多线程关闭，但不会立马关闭
executor.shutdown();
// 阻塞当前线程，知道所有子线程都执行countDown方法才会继续执行
countDownLatch.await();
// 打印线程池运行状态
System.out.println("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" +
executor.getQueue().size() + "，已执行结束的任务数目：" + executor.getCompletedTaskCount());
if (flag.isF()) {
// 打印计数
System.out.println("结束：" + totalRows.get());
} else {
System.out.println("有子线程报错，结果不准确");
}
}
}
