java google 并发包 封装

package com.tuchaoshi.base.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

public abstract class ListeningExecutorServiceJob<K, V> {

private final static Logger logger = LoggerFactory
.getLogger(ListeningExecutorServiceJob.class);

public static final int AVAILABLE_PROCESSORS_SIZE = Runtime.getRuntime()
.availableProcessors();

private static ListeningExecutorService executorService = MoreExecutors
.listeningDecorator(Executors
.newFixedThreadPool(AVAILABLE_PROCESSORS_SIZE * 2));

public ListeningExecutorServiceJob() {

}

/**
* 多线程执行商品生成信息
*
* @description
* @return List<TbArtsGoods>
* @Exception
*/
public List<V> executeTask(List<K> keyList) {
if (keyList == null) {
return null;
}

List<V> vList = Lists.newArrayList();

long gstartTime = System.currentTimeMillis();

List<ListenableFuture<List<V>>> futures = Lists.newArrayList();
for (K k : keyList) {// 3级类别分组
futures.add(executeTask(k));
}
ListenableFuture<List<List<V>>> successfulQueries = Futures
.successfulAsList(futures);
try {
// 获取所有线程的执行结果
List<List<V>> lists = successfulQueries.get();
if (!CollectionUtils.isEmpty(lists)) {
for (List<V> list : lists) {
if (!CollectionUtils.isEmpty(list)) {
vList.addAll(list);
}
}

}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
logger.info(" executeTask ! cost time:"
+ (System.currentTimeMillis() - gstartTime));

return vList;
}

/**
*
* @description
* @return ListenableFuture<List<TbArtsGoods>>
* @Exception
*/
private ListenableFuture<List<V>> executeTask(final K k) {
ListenableFuture<List<V>> listenableFuture = executorService
.submit(new Callable<List<V>>() {
@Override
public List<V> call() throws Exception {
return createTask(k);
}
});
return listenableFuture;
}

public abstract List<V> createTask(K k);

/**
* 拆分任务
*
* @param tasks
* @param 拆分数量
* @return
*/
public static <T> List<List<T>> splitTask(List<T> tasks, Integer taskSize) {
List<List<T>> list = Lists.newArrayList();

int baseNum = tasks.size() / taskSize; // 每个list的最小size
int remNum = tasks.size() % taskSize; // 得到余数

int index = 0;
for (int i = 0; i < taskSize; i++) {
int arrNum = baseNum; // 每个list对应的size
if (i < remNum) {
arrNum += 1;
}
List<T> ls = Lists.newArrayList();
for (int j = index; j < arrNum + index; j++) {
ls.add(tasks.get(j));
}
list.add(ls);
index += arrNum;
}
return list;
}

public static void main(String[] args) {
ListeningExecutorServiceJob<String, String> executorServiceTask = new ListeningExecutorServiceJob<String, String>() {
@Override
public List<String> createTask(String k) {
List<String> list = Lists.newArrayList();
list.add(k + "====");
return list;
}

};

List<String> keyList = Lists.newArrayList();
keyList.add("1");
keyList.add("2");
keyList.add("3");
keyList.add("4");
keyList.add("5");
keyList.add("6");
keyList.add("7");
keyList.add("8");
keyList.add("9");
List<String> valueList = executorServiceTask.executeTask(keyList);
System.out.println(valueList);

System.out.println(splitTask(keyList, 6));
System.out.println(splitTask(keyList, 5));
System.out.println(splitTask(keyList, 4));
System.out.println(splitTask(keyList, 3));
System.out.println(splitTask(keyList, 2));
System.out.println(splitTask(keyList, 1));

ListeningExecutorServiceJob<List<String>, String> executorServiceTask2 = new ListeningExecutorServiceJob<List<String>, String>() {
@Override
public List<String> createTask(List<String> k) {
List<String> list = Lists.newArrayList();
for (String s : k) {
list.add(s + "=====");
}
return list;
}

};
List<List<String>> strings = splitTask(keyList, 3);

System.out.println(executorServiceTask2.executeTask(strings));

}
}