third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: he bug of calculate thread pool size in ProcessorTracker

Browse Source
This commit is contained in:
tjq 2021-01-16 22:56:40 +08:00
parent dd66c8fd31
commit 3cb9ddf31f
1 changed files with 8 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
powerjob-worker/src/main/java/com/github/kfcfans/powerjob/worker/core/tracker/processor/ProcessorTracker.java
View File

@ -198,8 +198,8 @@ public class ProcessorTracker {
int poolSize = calThreadPoolSize(); int poolSize = calThreadPoolSize();
// 待执行队列，为了防止对内存造成较大压力，内存队列不能太大 // 待执行队列，为了防止对内存造成较大压力，内存队列不能太大
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(THREAD_POOL_QUEUE_MAX_SIZE); BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(THREAD_POOL_QUEUE_MAX_SIZE);
// 自定义线程池中线程名称 // 自定义线程池中线程名称 (PowerJob Processor Pool -> PPP)
ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("oms-processor-pool-%d").build(); ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("PPP-%d").build();
// 拒绝策略：直接抛出异常 // 拒绝策略：直接抛出异常
RejectedExecutionHandler rejectionHandler = new ThreadPoolExecutor.AbortPolicy(); RejectedExecutionHandler rejectionHandler = new ThreadPoolExecutor.AbortPolicy();
@ -214,8 +214,8 @@ public class ProcessorTracker {
*/ */
private void initTimingJob() { private void initTimingJob() {
// 全称 oms-ProcessTracker-TimingPool // PowerJob Processor TimingPool
ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("oms-ProcessorTrackerTimingPool-%d").build(); ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("PPT-%d").build();
timingPool = Executors.newSingleThreadScheduledExecutor(threadFactory); timingPool = Executors.newSingleThreadScheduledExecutor(threadFactory);
timingPool.scheduleAtFixedRate(new CheckerAndReporter(), 0, 10, TimeUnit.SECONDS); timingPool.scheduleAtFixedRate(new CheckerAndReporter(), 0, 10, TimeUnit.SECONDS);
@ -340,13 +340,14 @@ public class ProcessorTracker {
ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType()); ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType());
ProcessorType processorType = ProcessorType.valueOf(instanceInfo.getProcessorType()); ProcessorType processorType = ProcessorType.valueOf(instanceInfo.getProcessorType());
if (executeType == ExecuteType.MAP_REDUCE || executeType == ExecuteType.MAP) {
return instanceInfo.getThreadConcurrency();
}
// 脚本类自带线程池，不过为了少一点逻辑判断，还是象征性分配一个线程 // 脚本类自带线程池，不过为了少一点逻辑判断，还是象征性分配一个线程
if (processorType == ProcessorType.PYTHON || processorType == ProcessorType.SHELL) { if (processorType == ProcessorType.PYTHON || processorType == ProcessorType.SHELL) {
return 1; return 1;
} }
if (executeType == ExecuteType.MAP_REDUCE || executeType == ExecuteType.MAP) {
return instanceInfo.getThreadConcurrency();
}
if (TimeExpressionType.frequentTypes.contains(instanceInfo.getTimeExpressionType())) { if (TimeExpressionType.frequentTypes.contains(instanceInfo.getTimeExpressionType())) {
return instanceInfo.getThreadConcurrency(); return instanceInfo.getThreadConcurrency();
} }