yeyinlaoying/neutrino-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

线程池相关使用优化.

Browse Source
This commit is contained in:
aoshiguchen
2022-09-04 20:48:49 +08:00
parent c8cd56417e
commit 4e5c4a09de
7 changed files with 118 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
neutrino-core/src/main/java/fun/asgc/neutrino/core/aop/support/AsyncInterceptor.java
+30 -6
View File
@@ -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<Runnable>());
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;
}
}
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/bean/AbstractBeanFactory.java
+30 -5
View File
@@ -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;
}
}
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/DefaultJobSource.java
+2 -1
View File
@@ -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<JobInfo> 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()
neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/IJobExecutor.java
+9 -2
View File
@@ -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);
}
neutrino-core/src/main/java/fun/asgc/neutrino/core/quartz/JobExecutor.java
+16 -7
View File
@@ -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<String, IJobHandler> jobHandlerMap = new ConcurrentHashMap<>();
private Set<String> 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 {
neutrino-core/src/test/java/fun/asgc/neutrino/core/scheduler/test2/JobCallback.java
+2 -2
View File
@@ -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);
}
}
}
neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/base/rest/config/JobConfig.java
+29 -1
View File
@@ -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;
}
}
}