From 4e5c4a09de11ddd00f784e676d1feebb234ffc21 Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Sun, 4 Sep 2022 20:48:49 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E6=B1=A0=E7=9B=B8=E5=85=B3?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E4=BC=98=E5=8C=96.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/aop/support/AsyncInterceptor.java | 36 +++++++++++++++---- .../core/bean/AbstractBeanFactory.java | 35 +++++++++++++++--- .../core/quartz/DefaultJobSource.java | 3 +- .../neutrino/core/quartz/IJobExecutor.java | 11 ++++-- .../neutrino/core/quartz/JobExecutor.java | 23 ++++++++---- .../core/scheduler/test2/JobCallback.java | 4 +-- .../server/base/rest/config/JobConfig.java | 30 +++++++++++++++- 7 files changed, 118 insertions(+), 24 deletions(-) diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/aop/support/AsyncInterceptor.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/aop/support/AsyncInterceptor.java index 02072de8..d3cb678a 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/aop/support/AsyncInterceptor.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/aop/support/AsyncInterceptor.java @@ -24,10 +24,8 @@ package fun.asgc.neutrino.core.aop.support; import fun.asgc.neutrino.core.aop.Invocation; import fun.asgc.neutrino.core.aop.interceptor.Interceptor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; /** * 异步拦截器 @@ -36,8 +34,7 @@ import java.util.concurrent.TimeUnit; */ public class AsyncInterceptor implements Interceptor { private static final ExecutorService executorService = new ThreadPoolExecutor(10, 50, - 10L, TimeUnit.SECONDS, - new LinkedBlockingQueue()); + 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new AsyncThreadFactory()); @Override public void intercept(Invocation inv) throws Exception { @@ -49,4 +46,31 @@ public class AsyncInterceptor implements Interceptor { } }); } + + static class AsyncThreadFactory implements ThreadFactory { + private static final AtomicInteger poolNumber = new AtomicInteger(1); + private final ThreadGroup group; + private final AtomicInteger threadNumber = new AtomicInteger(1); + private final String namePrefix; + + AsyncThreadFactory() { + SecurityManager s = System.getSecurityManager(); + group = (s != null) ? s.getThreadGroup() : + Thread.currentThread().getThreadGroup(); + namePrefix = "AsyncPool-" + poolNumber.getAndIncrement() + "-thread-"; + } + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), + 0); + if (t.isDaemon()) { + t.setDaemon(false); + } + if (t.getPriority() != Thread.NORM_PRIORITY) { + t.setPriority(Thread.NORM_PRIORITY); + } + return t; + } + } } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/AbstractBeanFactory.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/AbstractBeanFactory.java index dd38b27b..960cba42 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/AbstractBeanFactory.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/AbstractBeanFactory.java @@ -33,10 +33,8 @@ import java.util.Comparator; import java.util.LinkedList; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; /** @@ -73,7 +71,7 @@ public abstract class AbstractBeanFactory implements BeanFactory, BeanRegistry, /** * 调度器 */ - private static final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(); + private static final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(new BeanFactoryThreadFactory()); public AbstractBeanFactory(String name) { this(null, name); @@ -568,4 +566,31 @@ public abstract class AbstractBeanFactory implements BeanFactory, BeanRegistry, () -> environment ); } + + static class BeanFactoryThreadFactory implements ThreadFactory { + private static final AtomicInteger poolNumber = new AtomicInteger(1); + private final ThreadGroup group; + private final AtomicInteger threadNumber = new AtomicInteger(1); + private final String namePrefix; + + BeanFactoryThreadFactory() { + SecurityManager s = System.getSecurityManager(); + group = (s != null) ? s.getThreadGroup() : + Thread.currentThread().getThreadGroup(); + namePrefix = "BeanFactoryPool-" + poolNumber.getAndIncrement() + "-thread-"; + } + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), + 0); + if (t.isDaemon()) { + t.setDaemon(false); + } + if (t.getPriority() != Thread.NORM_PRIORITY) { + t.setPriority(Thread.NORM_PRIORITY); + } + return t; + } + } } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/DefaultJobSource.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/DefaultJobSource.java index 9fae3b45..cb2906a8 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/DefaultJobSource.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/DefaultJobSource.java @@ -25,6 +25,7 @@ import com.google.common.collect.Lists; import fun.asgc.neutrino.core.quartz.annotation.JobHandler; import fun.asgc.neutrino.core.util.BeanManager; import fun.asgc.neutrino.core.util.CollectionUtil; +import fun.asgc.neutrino.core.util.StringUtil; import java.util.List; @@ -44,7 +45,7 @@ public class DefaultJobSource implements IJobSource { List jobInfoList = Lists.newArrayList(); for (IJobHandler jobHandler : jobHandlerList) { JobHandler handler = jobHandler.getClass().getAnnotation(JobHandler.class); - if (null == handler) { + if (null == handler || StringUtil.isEmpty(handler.name()) || StringUtil.isEmpty(handler.cron())) { continue; } jobInfoList.add(new JobInfo() diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/IJobExecutor.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/IJobExecutor.java index 9b7cc1e6..cc4aff31 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/IJobExecutor.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/IJobExecutor.java @@ -42,7 +42,14 @@ public interface IJobExecutor { /** * 删除job - * @param jobId + * @param jobName */ - void remove(String jobId); + void remove(String jobName); + + /** + * 触发 + * @param jobName + * @param param + */ + void trigger(String jobName, String param); } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/JobExecutor.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/JobExecutor.java index 64f9a3b1..26802a01 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/JobExecutor.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/JobExecutor.java @@ -21,6 +21,7 @@ */ package fun.asgc.neutrino.core.quartz; +import com.google.common.collect.Sets; import fun.asgc.neutrino.core.annotation.Autowired; import fun.asgc.neutrino.core.context.ApplicationRunner; import fun.asgc.neutrino.core.context.Environment; @@ -34,6 +35,7 @@ import org.quartz.impl.StdSchedulerFactory; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadPoolExecutor; @@ -52,6 +54,7 @@ public class JobExecutor implements ApplicationRunner, IJobExecutor { private SchedulerFactory schedulerFactory; private Scheduler scheduler; private Map jobHandlerMap = new ConcurrentHashMap<>(); + private Set runJobSet = Sets.newHashSet(); private IJobCallback jobCallback; @Override @@ -68,6 +71,7 @@ public class JobExecutor implements ApplicationRunner, IJobExecutor { continue; } jobHandlerMap.put(jobHandler.name(), item); + runJobSet.add(jobHandler.name()); } } @@ -106,13 +110,13 @@ public class JobExecutor implements ApplicationRunner, IJobExecutor { @Override public synchronized void add(JobInfo jobInfo) throws JobException { - if (null == jobInfo || StringUtil.isEmpty(jobInfo.getId()) || StringUtil.isEmpty(jobInfo.getCron()) || jobInfoMap.containsKey(jobInfo.getId())) { + if (null == jobInfo || StringUtil.isEmpty(jobInfo.getName()) || StringUtil.isEmpty(jobInfo.getCron()) || jobInfoMap.containsKey(jobInfo.getName())) { return; } - jobInfoMap.put(jobInfo.getId(), jobInfo); + jobInfoMap.put(jobInfo.getName(), jobInfo); - TriggerKey triggerKey = TriggerKey.triggerKey(jobInfo.getId()); - JobKey jobKey = new JobKey(jobInfo.getId()); + TriggerKey triggerKey = TriggerKey.triggerKey(jobInfo.getName()); + JobKey jobKey = new JobKey(jobInfo.getName()); CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(jobInfo.getCron()).withMisfireHandlingInstructionDoNothing(); CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity(triggerKey).withSchedule(cronScheduleBuilder).build(); @@ -122,13 +126,18 @@ public class JobExecutor implements ApplicationRunner, IJobExecutor { scheduler.scheduleJob(jobDetail, cronTrigger); scheduler.start(); } catch (Exception e) { - throw new RuntimeException(String.format("新增job[id=%s]异常", jobInfo.getId())); + throw new RuntimeException(String.format("新增job[name=%s]异常", jobInfo.getName())); } } @Override - public void remove(String jobId) { - jobInfoMap.remove(jobId); + public void remove(String jobName) { + runJobSet.remove(jobName); + } + + @Override + public void trigger(String jobId, String param) { + } public void execute(JobExecutionContext context) throws JobExecutionException { diff --git a/neutrino-core/src/test/java/fun/asgc/neutrino/core/scheduler/test2/JobCallback.java b/neutrino-core/src/test/java/fun/asgc/neutrino/core/scheduler/test2/JobCallback.java index 520ce91c..5b585a3a 100644 --- a/neutrino-core/src/test/java/fun/asgc/neutrino/core/scheduler/test2/JobCallback.java +++ b/neutrino-core/src/test/java/fun/asgc/neutrino/core/scheduler/test2/JobCallback.java @@ -36,9 +36,9 @@ public class JobCallback implements IJobCallback { @Override public void executeLog(JobInfo jobInfo, Throwable throwable) { if (null == throwable) { - log.info("job[id={},name={}]执行完毕", jobInfo.getId(), jobInfo.getName()); + log.info("job[name={}]执行完毕", jobInfo.getId(), jobInfo.getName()); } else { - log.error("job[id={},name={}]执行异常", jobInfo.getId(), jobInfo.getName(), throwable); + log.error("job[name={}]执行异常", jobInfo.getId(), jobInfo.getName(), throwable); } } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/base/rest/config/JobConfig.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/base/rest/config/JobConfig.java index 010edd88..68287ae4 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/base/rest/config/JobConfig.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/base/rest/config/JobConfig.java @@ -29,8 +29,10 @@ import fun.asgc.neutrino.core.quartz.JobExecutor; import fun.asgc.neutrino.proxy.server.service.JobLogService; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; /** * @@ -46,9 +48,35 @@ public class JobConfig { public JobExecutor jobExecutor() { JobExecutor executor = new JobExecutor(); executor.setJobSource(new DefaultJobSource()); - executor.setThreadPoolExecutor(new ThreadPoolExecutor(5, 20, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>())); + executor.setThreadPoolExecutor(new ThreadPoolExecutor(5, 20, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new JobThreadFactory())); executor.setJobCallback(jobLogService); return executor; } + static class JobThreadFactory implements ThreadFactory { + private static final AtomicInteger poolNumber = new AtomicInteger(1); + private final ThreadGroup group; + private final AtomicInteger threadNumber = new AtomicInteger(1); + private final String namePrefix; + + JobThreadFactory() { + SecurityManager s = System.getSecurityManager(); + group = (s != null) ? s.getThreadGroup() : + Thread.currentThread().getThreadGroup(); + namePrefix = "JobPool-" + poolNumber.getAndIncrement() + "-thread-"; + } + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), + 0); + if (t.isDaemon()) { + t.setDaemon(false); + } + if (t.getPriority() != Thread.NORM_PRIORITY) { + t.setPriority(Thread.NORM_PRIORITY); + } + return t; + } + } }