使用線程池批量順序執行任務
AtomicInteger、CountDownLatch、ThreadPoolExecutor組合使用
package tool.thread;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Created by zengrenyuan on 17/9/19.
*/
public class ThreadPollUtils {
private static final Logger LOG = LoggerFactory.getLogger(ThreadPollUtils.class);
/**
* @param paramList 任務參數列表
* @param process 要執行的功能
* @param threadNum 線程數
* @param <P> 參數類型
* @param <V> 返回值類型
* @return 返回結果集
* @throws InterruptedException
*/
public static <P, V> Map<Integer, V> process(final List<P> paramList, final Process<P, V> process, int threadNum) throws InterruptedException {
//如果任務數少於線程數取任務為線程個數
threadNum = Math.min(paramList.size(), threadNum);
//保證任務是順序執行的
final AtomicInteger atomicInteger = new AtomicInteger(0);
//用於統計已經執行完成的線程
final CountDownLatch countDownLatch = new CountDownLatch(threadNum);
//保存任務執行結果
final ConcurrentHashMap<Integer, V> resultMap = new ConcurrentHashMap<>();
ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(threadNum);
threadPool.setMaximumPoolSize(threadNum);
for (int i = 0; i < threadNum; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
while (true) {
int index = atomicInteger.getAndAdd(1);
if (index >= paramList.size()) {
break;
}
P param = paramList.get(index);
try {
LOG.info("index:{}", index);
V result = process.doTask(param);
if (result != null) {
LOG.info("index:{} result:{}", index, JSON.toJSONString(result));
resultMap.put(index, result);
}
} catch (Exception e) {
LOG.error("運行失敗 param:{}", JSON.toJSONString(param), e);
}
}
} catch (Exception e) {
LOG.error(null, e);
} finally {
countDownLatch.countDown();
}
}
});
}
try {
countDownLatch.await();
} catch (Exception e) {
LOG.error(null, e);
}
threadPool.shutdown();
return resultMap;
}
/**
* @param <P> 參數類型
* @param <V> 返回值類型
*/
public interface Process<P, V> {
V doTask(P p);
}
}
————————————————
版權聲明：本文為CSDN博主「元小帥」的原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/zengrenyuan/article/details/78117559