參考资料： 阿里巴巴开源项目 CobarClient 源代码实现。
分享作者：闫建忠
分享时间：2014年5月7日
---------------------------------------------------------------------------------------
并行调度封装类设计： BXexample.java
package org.hdht.business.ordermanager.quartzjob; import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.dao.ConcurrencyFailureException; public class BXexample { private static ExecutorService createCustomExecutorService(int poolSize, final String method) {
int coreSize = Runtime.getRuntime().availableProcessors();//返回系统CUP数量
if (poolSize < coreSize) {
coreSize = poolSize;
}
ThreadFactory tf = new ThreadFactory() {
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "thread created at BXexample method [" + method + "]");
t.setDaemon(true);
return t;
}
};
BlockingQueue<Runnable> queueToUse = new LinkedBlockingQueue<Runnable>();
final ThreadPoolExecutor executor = new ThreadPoolExecutor(coreSize, poolSize, 60,
TimeUnit.SECONDS, queueToUse, tf, new ThreadPoolExecutor.CallerRunsPolicy()); return executor;
} public static <T> List<T> getSubListPage(List<T> list, int skip,int pageSize) {
if (list == null || list.isEmpty()) {
return null;
}
int startIndex = skip;
int endIndex = skip + pageSize;
if (startIndex > endIndex || startIndex > list.size()) {
return null;
}
if (endIndex > list.size()) {
endIndex = list.size();
}
return list.subList(startIndex, endIndex);
} public static void BXfunction(Collection<?> paramCollection,final ExectueCallBack ecb){
//构建运行器
ExecutorService executor = createCustomExecutorService(Runtime.getRuntime().availableProcessors(), "batchExecuteProjection");
try {
//监视器
final CountDownLatch latch = new CountDownLatch(paramCollection.size());
final StringBuffer exceptionStaktrace = new StringBuffer();
Iterator<?> iter = paramCollection.iterator();
while (iter.hasNext()) {
final Object entity = iter.next();
Runnable task = new Runnable() {
public void run() {
try {
ecb.doExectue(entity);
} catch (Throwable t) {
exceptionStaktrace.append(ExceptionUtils.getFullStackTrace(t));
} finally {
latch.countDown();
}
}
};
executor.execute(task);//并行调度
} try {
latch.await();//监视器等待全部线程运行完成
} catch (InterruptedException e) {
//调度异常
throw new ConcurrencyFailureException(
"unexpected interruption when re-arranging parameter collection into sub-collections ",e);
}
if (exceptionStaktrace.length() > 0) {
//业务异常
throw new ConcurrencyFailureException(
"unpected exception when re-arranging parameter collection, check previous log for details.\n"+ exceptionStaktrace);
} } finally {
executor.shutdown();//运行器关闭
}
} }
回调接口类设计：ExectueCallBack.java
package org.hdht.business.ordermanager.quartzjob;
public interface ExectueCallBack {
void doExectue(Object executor) throws Exception;
}
演示样例（hello 演示样例）
public static void main(String[] args) {
List<String> paramCollection = new ArrayList<String>();
paramCollection.add("9");
paramCollection.add("2");
paramCollection.add("18");
paramCollection.add("7");
paramCollection.add("6");
paramCollection.add("1");
paramCollection.add("3");
paramCollection.add("4");
paramCollection.add("14");
paramCollection.add("13");
int freesize = 3;//当前处理能力
for(int i=0;i<paramCollection.size();i=i+freesize){
List<String> tl = BXexample.getSubListPage(paramCollection, i, freesize);
BXexample.BXfunction(tl,new ExectueCallBack() {
public void doExectue(Object executor) throws Exception {
int k = Integer.parseInt((String)executor);
for(int i=0;i<k*10000000;i++){
//运行循环
}
System.out.println(k+"：hello world");
}
});
}
}演示样例（实际业务应用演示样例）
/**
* 并行调度相关处理
*
* 按卫星*日期 ，将待处理的任务分解为 卫星+日期 粒度的子任务 加入到paramMapList列表中
*/
List<Map<String, Object>> paramMapList = new ArrayList<Map<String, Object>>();
for (Iterator<OrderParamSatellite> iterator = paramSatellites.iterator(); iterator.hasNext();) {
OrderParamSatellite paramSatellite = iterator.next(); paramMapList.addAll(this.getParamMapList(paramSatellite));
} //依据集群最大处理能力,分页处理任务列表，作为list截取的步长 int fsize = HostServerQueue.getInstance().freeSize();
for(int i=0;i<paramMapList.size();i=i+fsize){
List<Map<String, Object>> tl = BXexample.getSubListPage(paramMapList, i, fsize);
//并行调度
BXexample.BXfunction(tl,new ExectueCallBack(){
public void doExectue(Object executor) throws Exception {
ExecuteOrderBTask((Map<String, Object>)executor);
}
}); //动态查找空暇节点数量，即集群最大处理能力
fsize = HostServerQueue.getInstance().freeSize();
}