标签：inf timeout 响应 str any 背景 map 主线程 link
公司很多项目都会用到YED包来画关联方的关联图，当某个公司的4度关联方个数过多以及关联关系太过复杂时，使用YED画图会造成CPU占用一直100%并且长时间占用不释放CPU资源。所以在使用YED画图的代码流程设计需要一些技巧来保证YED画图不会阻塞主线程，并且确保当图片长时间无法绘制成功时，需要有一种机制强制终止绘制，避免CPU资源长时间无意义的浪费。
首先尝试了找寻问题的根本原因，提出3个假设：由于数据有问题？YED代码bug？数据量过大造成内存不足？ 然而遗憾的是时间有限从这3方面入手都并没有进展，如果各位同事有兴趣可以尝试找出根本原因，有进展可与我联系。
所以开始尝试超时终止绘制方案，具体思路如下：
阅读YED源代码发现，项目中使用的YED版本并不响应线程中断，所以只有使用stop()方法强制终止线程，这样我们就需要在线程池中取到需要终止的线程，考虑自定义一个连接池，连接池中维护一个以任务为key，以执行这个任务线程为value的map，让我们在需要时可以使用任务取到对应的线程并终止它。
同时应产品要求如果4度图片绘制失败需要尝试3度图片绘制，3度失败再尝试2度，所以提交任务时需要传入一个供用户回调的接口。
用以上思路代码实现大致如下：
自定义线程池如下：
public class TraceThreadPoolExecutor extends ThreadPoolExecutor {
// 1 minutes
private static final int MONITOR_SHUTDOWN_TIMEOUT = 60;
private final Map<Runnable, Thread> tracer = new ConcurrentHashMap<>();
private final ExecutorService monitor;
/**
* Submit a job and monitor elapse time, if job elapse time greater than
* timeout threshold, force stop it.
*
* @param job
* @param timeOut
* @param timeUnit
* @param handler When job timeout, call it.
* @return
*/
public Future<?> submitJob(Runnable job, int timeOut, TimeUnit timeUnit, TimeOutHandler handler) {
Future<?> future = submit(job);
monitor.submit(new MonitorJob(job, future, timeOut, timeUnit, handler));
return future;
}
/**
* Stop executor & monitor, if timeout, force stop.
*/
public void stop() {
/** Stop monitor executor */
shutdownExecutorQuietly(monitor);
/** Stop worker executor */
shutdownExecutorQuietly(this);
}
private void shutdownExecutorQuietly(ExecutorService executor) {
if (null != executor) {
try {
executor.shutdown();
if (!executor.isShutdown()) {
try {
executor.awaitTermination(MONITOR_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
} catch (InterruptedException ignore) {
}
/** 等待超时或被中断，直接强制停止 */
if (!executor.isShutdown()) {
executor.shutdownNow();
}
}
} catch (Exception ignore) {
// do nothing
}
}
}
@Override
protected void beforeExecute(Thread worker, Runnable job) {
traceThread(job, worker);
}
@Override
protected void afterExecute(Runnable job, Throwable error) {
removeTraceThread(job);
}
@SuppressWarnings("rawtypes")
private class MonitorJob implements Runnable {
private final Future<?> future;
private final Runnable originJob;
private final int timeOut;
private final TimeUnit timeUnit;
private final TimeOutHandler handler;
public MonitorJob(Runnable originJob, Future<?> future, int timeOut, TimeUnit timeUnit, TimeOutHandler handler) {
this.originJob = originJob;
this.future = future;
this.timeOut = timeOut;
this.timeUnit = timeUnit;
this.handler = handler;
}
@Override
public void run() {
if (null != future) {
try {
future.get(timeOut, timeUnit);
} catch (InterruptedException ignore) {
} catch (ExecutionException ignore) {
} catch (TimeoutException timeOut) {
/** Stop worker thread force */
Thread worker = getTraceThread((FutureTask) future);
if (null != worker && worker.isAlive()) {
worker.stop();
}
/** Do callback */
if (null != handler) {
handler.doHandle(originJob);
}
}
}
}
}
private void traceThread(Runnable job, Thread worker) {
tracer.put(job, worker);
}
private Thread getTraceThread(Runnable job) {
return tracer.get(job);
}
private void removeTraceThread(Runnable job) {
tracer.remove(job);
}
private TraceThreadPoolExecutor(int workerThreads, int monitorThreads) {
super(workerThreads, workerThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
monitor = Executors.newFixedThreadPool(monitorThreads);
}
public static TraceThreadPoolExecutor createFixedThreadPool(int workerThreads, int monitorThreads) {
return new TraceThreadPoolExecutor(workerThreads, monitorThreads);
}
}
并发任务处理代码如下：
@Component
@Lazy(false)
public class TasksHandler {
@Value("${system.processor.num:2}")
private int processors;
@Value("${system.monitor.num:2}")
private int monitorProcessors;
/**
* 日志
*/
private final Logger logger = LoggerFactory.getLogger(TasksHandler.class);
private TraceThreadPoolExecutor executor;
@Autowired
private JobScheduleService jobScheduleService;
@Autowired
private FileFactory fileFactory;
@PostConstruct
public void initialize() {
executor = TraceThreadPoolExecutor.createFixedThreadPool(processors, monitorProcessors);
List<JobSchedule> jobSchedules = jobScheduleService.selectByStatuses(Lists.newArrayList(NEW, CREATING));
jobSchedules.forEach(jobSchedule -> submitJob(jobSchedule));
}
@PreDestroy
public void stop() {
executor.stop();
}
public void submitJob(JobSchedule jobSchedule) {
Map<Object, Object> params = new HashMap<>();
switch (jobSchedule.getJobType()) {
case "DNA":
params.put("dnaDeep", 4);
break;
case "HTML":
break;
case "PDF":
params.put("bothTemplate", true);
break;
}
executor.submitJob(new TaskThread(fileFactory, jobScheduleService, jobSchedule, params), 5, TimeUnit.MINUTES, job -> {
switch (jobSchedule.getJobType()) {
case "DNA":
recursionHandler(jobSchedule, params);
break;
case "HTML":
logger.info("HTML第一次生成超时，稍后进行重试");
retryHandler(jobSchedule, params);
break;
case "PDF":
logger.info("PDF第一次生成超时，稍后进行重试");
retryHandler(jobSchedule, params);
break;
}
});
}
private void recursionHandler(JobSchedule jobSchedule, Map params) {
if ((int) params.get("dnaDeep") == 1)
return;
logger.info("{}度DNA生成超时，开始尝试生成{}度DNA", params.get("dnaDeep"), (int) params.get("dnaDeep") - 1);
params.put("dnaDeep", (int) params.get("dnaDeep") - 1);
executor.submitJob(new TaskThread(fileFactory, jobScheduleService, jobSchedule, params), 2, TimeUnit.MINUTES, dnaJob -> {
recursionHandler(jobSchedule, params);
});
}
private void retryHandler(JobSchedule jobSchedule, Map params) {
executor.submitJob(new TaskThread(fileFactory, jobScheduleService, jobSchedule, params), 5, TimeUnit.MINUTES, job -> {
switch (jobSchedule.getJobType()) {
case "HTML":
logger.info("HTML生成超时 id:{}, companyName:{}", jobSchedule.getId(), jobSchedule.getCompanyName());
break;
case "PDF":
logger.info("PDF生成超时 id:{}, companyName:{}", jobSchedule.getId(), jobSchedule.getCompanyName());
break;
}
});
}
}标签：inf timeout 响应 str any 背景 map 主线程 link
原文地址：http://www.cnblogs.com/liushuncheng/p/7428511.html
