xiaoguangbin/AnalysisSystemForRadionuclide
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：增加对异常的拦截，修改map为ConcurrentHashMap

Browse Source
This commit is contained in:
xiaoguangbin 2024-06-27 09:58:22 +08:00
parent f6b963145f
commit 1423b2f09d
1 changed files with 75 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

137
jeecg-module-auto-process/src/main/java/org/jeecg/modules/AutoProcessManager.java
View File

@ -16,6 +16,7 @@ import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
/** /**
@ -39,12 +40,12 @@ public class AutoProcessManager{
/** /**
* 以邮件Id为key，邮件信息为value * 以邮件Id为key，邮件信息为value
*/ */
private Map<String,EmailProperties> emailMap = new HashMap<>(); private Map<String,EmailProperties> emailMap = new ConcurrentHashMap<>();
/** /**
* 以邮件id为key，以邮件执行线程为value * 以邮件id为key，以邮件执行线程为value
*/ */
private Map<String,EmailParsingActuator> emailExecThreadMap = new HashMap<>(); private Map<String,EmailParsingActuator> emailExecThreadMap = new ConcurrentHashMap<>();
/** /**
* 启动自动处理 * 启动自动处理
@ -81,34 +82,40 @@ public class AutoProcessManager{
public void run() { public void run() {
for(;;){ for(;;){
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
if(!CollectionUtils.isEmpty(emailMap)){ try {
synchronized (lock){ if(!CollectionUtils.isEmpty(emailMap)){
Iterator<EmailProperties> iterator = emailMap.values().iterator(); synchronized (lock){
while(iterator.hasNext()){ Iterator<EmailProperties> iterator = emailMap.values().iterator();
EmailProperties next = iterator.next(); while(iterator.hasNext()){
if (next.isResetFlag()) { EmailProperties next = iterator.next();
EmailParsingActuator actuator = emailExecThreadMap.get(next.getId()); // ConcurrentModificationException
actuator.updateEmail(next); if (next.isResetFlag()) {
next.setResetFlag(false); EmailParsingActuator actuator = emailExecThreadMap.get(next.getId());
} actuator.updateEmail(next);
if(next.isNewEmailFlag()){ next.setResetFlag(false);
// 网络正常之后才允许创建新的实例 }
final EmailServiceManager emailServiceManager = EmailServiceManager.getInstance(); if(next.isNewEmailFlag()){
emailServiceManager.init(next); // 网络正常之后才允许创建新的实例
boolean testFlag = emailServiceManager.testConnectEmailServer(); final EmailServiceManager emailServiceManager = EmailServiceManager.getInstance();
if(testFlag){ emailServiceManager.init(next);
EmailParsingActuator emailParsingActuator = new EmailParsingActuator(); boolean testFlag = emailServiceManager.testConnectEmailServer();
emailParsingActuator.init(next,spectrumServiceQuotes,emailCounter,systemStartupTime); if(testFlag){
emailParsingActuator.setName(next.getUsername()+"-email-monitor"); EmailParsingActuator emailParsingActuator = new EmailParsingActuator();
emailParsingActuator.start(); emailParsingActuator.init(next,spectrumServiceQuotes,emailCounter,systemStartupTime);
//把邮件监测执行线程加入管理队列 emailParsingActuator.setName(next.getUsername()+"-email-monitor");
emailExecThreadMap.put(next.getId(),emailParsingActuator); emailParsingActuator.start();
//新邮件监测监测线程已启动则修改新邮件标记为false //把邮件监测执行线程加入管理队列
next.setNewEmailFlag(false); emailExecThreadMap.put(next.getId(),emailParsingActuator);
//新邮件监测监测线程已启动则修改新邮件标记为false
next.setNewEmailFlag(false);
}
} }
} }
} }
} }
} catch (Exception e) {
e.printStackTrace();
log.error(e.getMessage(), e);
} }
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
long sleepTime = taskProperties.getMonitoringMailDataCycle() - (end-start); long sleepTime = taskProperties.getMonitoringMailDataCycle() - (end-start);
@ -177,6 +184,7 @@ public class AutoProcessManager{
if(sleepTime > 0){ if(sleepTime > 0){
try { try {
TimeUnit.MILLISECONDS.sleep(sleepTime); TimeUnit.MILLISECONDS.sleep(sleepTime);
// throw new RuntimeException("运行时异常");
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e); log.error(e.getMessage(), e);
@ -281,46 +289,51 @@ public class AutoProcessManager{
public void run() { public void run() {
for(;;){ for(;;){
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
if(!CollectionUtils.isEmpty(emailExecThreadMap)){ try {
//遍历邮箱执行线程，如果状态为已停止则删除 if(!CollectionUtils.isEmpty(emailExecThreadMap)){
final Iterator<Map.Entry<String, EmailParsingActuator>> checkStopThreads = emailExecThreadMap.entrySet().iterator(); //遍历邮箱执行线程，如果状态为已停止则删除
while (checkStopThreads.hasNext()){ final Iterator<Map.Entry<String, EmailParsingActuator>> checkStopThreads = emailExecThreadMap.entrySet().iterator();
final Map.Entry<String, EmailParsingActuator> next = checkStopThreads.next(); while (checkStopThreads.hasNext()){
if(next.getValue().getState() == State.TERMINATED){ final Map.Entry<String, EmailParsingActuator> next = checkStopThreads.next();
log.info("{}邮箱执行线程已停止，emailExecThreadMap删除此邮箱数据",next.getValue().getEmailProperties().getUsername()); if(next.getValue().getState() == State.TERMINATED){
checkStopThreads.remove(); log.info("{}邮箱执行线程已停止，emailExecThreadMap删除此邮箱数据",next.getValue().getEmailProperties().getUsername());
emailMap.remove(next.getKey()); checkStopThreads.remove();
emailMap.remove(next.getKey());
}
} }
}
//遍历邮箱执行线程，如果邮箱执行线程stop属性已被设置为true则关闭资源停止线程 //遍历邮箱执行线程，如果邮箱执行线程stop属性已被设置为true则关闭资源停止线程
final Iterator<Map.Entry<String, EmailParsingActuator>> iterator = emailExecThreadMap.entrySet().iterator(); final Iterator<Map.Entry<String, EmailParsingActuator>> iterator = emailExecThreadMap.entrySet().iterator();
emailExecThreadMap.forEach((emailId,emailExecThread)->{ emailExecThreadMap.forEach((emailId,emailExecThread)->{
try{ try{
if(emailExecThread.getState() != State.TERMINATED && emailExecThread.isStop()){ if(emailExecThread.getState() != State.TERMINATED && emailExecThread.isStop()){
final long nowTime = System.currentTimeMillis(); final long nowTime = System.currentTimeMillis();
final long setStoptime = emailExecThread.getStopTime().getTime(); final long setStoptime = emailExecThread.getStopTime().getTime();
final long val = nowTime - setStoptime; final long val = nowTime - setStoptime;
if(val >= taskProperties.getForceDeletedTime()){ if(val >= taskProperties.getForceDeletedTime()){
log.info("关闭{}邮箱线程内部线程池资源",emailExecThread.getEmailProperties().getUsername()); log.info("关闭{}邮箱线程内部线程池资源",emailExecThread.getEmailProperties().getUsername());
emailExecThread.closeResource(); emailExecThread.closeResource();
}
}
}catch (Exception e){
e.printStackTrace();
log.error(e.getMessage(), e);
}finally {
if(emailExecThread.getState() != State.TERMINATED && emailExecThread.isStop()){
final long nowTime = System.currentTimeMillis();
final long setStoptime = emailExecThread.getStopTime().getTime();
final long val = nowTime - setStoptime;
if(val >= taskProperties.getForceDeletedTime()){
log.info("强制停止{}邮箱线程",emailExecThread.getEmailProperties().getUsername());
emailExecThread.stop();
}
} }
} }
}catch (Exception e){ });
e.printStackTrace(); }
log.error(e.getMessage(), e); } catch (Exception e) {
}finally { e.printStackTrace();
if(emailExecThread.getState() != State.TERMINATED && emailExecThread.isStop()){ log.error(e.getMessage(), e);
final long nowTime = System.currentTimeMillis();
final long setStoptime = emailExecThread.getStopTime().getTime();
final long val = nowTime - setStoptime;
if(val >= taskProperties.getForceDeletedTime()){
log.info("强制停止{}邮箱线程",emailExecThread.getEmailProperties().getUsername());
emailExecThread.stop();
}
}
}
});
} }
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
long sleepTime = taskProperties.getDeletedMailThreadExecCycle() - (end-start); long sleepTime = taskProperties.getDeletedMailThreadExecCycle() - (end-start);