一个调度线程
线程池 多个线程执行推送信息service
用ScheduledExecutorService 每2秒监控调度线程以及线程池状态
使用CountDownLatch 闭锁等待调度线程以及线程池执行完推送任务 后执行主线程业务逻辑
private void start(SendTask sendTask)throws Exception {
// CountDownLatch latch = new CountDownLatch(1);
CountDownLatch threadPoolLatch =new CountDownLatch(1);
CountDownLatch dispatchLatch =new CountDownLatch(1);
String msgContent = sendTask.getMsgContent();
String sendConfId =sendTask.getProviderId();
if(StringUtils.isNotEmpty(sendConfId)){
ProviderSendConf providerSendConf =providerSendConfRepository.findOne(new Integer(sendConfId));
if(providerSendConf !=null){
String sendService = providerSendConf.getSendService();
if(StringUtils.isNotEmpty(sendService)){
sendMsgService = (MsgSendService) SpringContextUtil.getBean(sendService);
}
}
}
ScheduledExecutorService scheduledService = Executors.newScheduledThreadPool(1);
ExecutorService taskProcessPool =new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),5000,60L, TimeUnit.SECONDS,
new LinkedBlockingQueue(10000));
new Thread(new Runnable() {
@Override
public void run() {
try {
dispatchLatch.await();
threadPoolLatch.await();
}catch (InterruptedException e) {
e.printStackTrace();
}
logger.info("task end");
System.out.println("task end");
scheduledService.shutdownNow();
}
}).start();
new Thread(new Runnable() {
public boolean isAllSend =false;
@Override
public void run() {
while (!isAllSend) {
List sendRecordList =sendRecordRepository.findBySendStatusAndFileId(SendStatus.UN_SEND.getStatus(),new Integer(sendTask.getFileId()));
if(sendRecordList !=null && sendRecordList.size()>0){
for (SendRecord sendRecord:sendRecordList) {
sendRecord.setSendStatus(SendStatus.SEND.getStatus());
sendRecord.setSendTime(new Date());
sendRecordRepository.save(sendRecord);
Msg msg =new Msg();
msg.setContent(msgContent);
msg.setMobile(sendRecord.getMobile());
MsgSendTask task =new MsgSendTask();
task.setSendService(sendMsgService);
task.setMsg(msg);
taskProcessPool.submit(task);
try {
Thread.sleep(1);
}catch (InterruptedException e) {
e.printStackTrace();
}
}
}else {
isAllSend =true;
}
}
dispatchLatch.countDown();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
//等任务派发完成才开始检查线程池状态
dispatchLatch.await();
}catch (InterruptedException e) {
e.printStackTrace();
}
ThreadPoolExecutor executor = (ThreadPoolExecutor) taskProcessPool;
if (executor.getQueue().size() ==0 && executor.getActiveCount() ==0) {
executor.shutdownNow();
threadPoolLatch.countDown();
ImportFile importFile =importFileRepository.findOne(new Integer(sendTask.getFileId()));
importFile.setStatus(ImportFileStatus.TASK_FINISH.getStatus());
importFileRepository.save(importFile);
sendTask.setTaskStatus(TaskStatus.FINISH.getStatus());
sendTaskRepository.save(sendTask);
logger.info("ThreadPool monitor exit....");
System.out.println("ThreadPool monitor exit....");
return;
}
logger.info("ThreadPool monitor....");
System.out.println("ThreadPool monitor....");
scheduledService.schedule(this,2, TimeUnit.SECONDS);
}
}).start();
// latch.await();
}
网友评论