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

Compare commits

merge into: xiaoguangbin:mdc
Branches Tags
xiaoguangbin:mdc
xiaoguangbin:SelfStation-pgsql
xiaoguangbin:SelfStation
xiaoguangbin:SelfStation-testdll
xiaoguangbin:mdc-merge
xiaoguangbin:sleepDownload
xiaoguangbin:noFtp-seflStation
xiaoguangbin:noFtp
xiaoguangbin:noFtp-F
xiaoguangbin:onlyDownload
xiaoguangbin:master
...
pull from: xiaoguangbin:sleepDownload
Branches Tags
xiaoguangbin:mdc
xiaoguangbin:SelfStation-pgsql
xiaoguangbin:SelfStation
xiaoguangbin:SelfStation-testdll
xiaoguangbin:mdc-merge
xiaoguangbin:sleepDownload
xiaoguangbin:noFtp-seflStation
xiaoguangbin:noFtp
xiaoguangbin:noFtp-F
xiaoguangbin:onlyDownload
xiaoguangbin:master

4 Commits
mdc ... sleepDownl

Author SHA1 Message Date
xiaoguangbin
1423b2f09d fix：增加对异常的拦截，修改map为ConcurrentHashMap 2024-06-27 09:58:22 +08:00
xiaoguangbin
f6b963145f fix:增加catch中exception写入log日志 2024-06-25 11:03:09 +08:00
nieziyan
64a98bd4ea fix：邮件接收超时时间 2024-06-12 15:33:23 +08:00
nieziyan
f1410028fa fix：不删除saveFile下谱文件 2024-06-06 16:51:24 +08:00
4 changed files with 94 additions and 70 deletions
Show Stats Expand all files Collapse all files

14
jeecg-boot-base-core/src/main/java/org/jeecg/common/email/EmailServiceManager.java
View File

@ -1,8 +1,6 @@
package org.jeecg.common.email; package org.jeecg.common.email;
import cn.hutool.core.util.ArrayUtil; import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil; import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import com.sun.mail.imap.IMAPStore; import com.sun.mail.imap.IMAPStore;
@ -32,7 +30,6 @@ import java.io.*;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.Socket; import java.net.Socket;
import java.util.*; import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -127,7 +124,7 @@ public class EmailServiceManager {
/** /**
* 接收邮件 * 接收邮件
*/ */
public Message[] receiveMail() throws MessagingException { public Message[] receiveMail() throws Exception {
String status = EmailLogManager.STATUS_SUCCESS; String status = EmailLogManager.STATUS_SUCCESS;
try{ try{
//配置邮件服务属性 //配置邮件服务属性
@ -135,6 +132,8 @@ public class EmailServiceManager {
properties.put("mail.store.protocol", "imap"); properties.put("mail.store.protocol", "imap");
properties.put("mail.imap.host", email.getEmailServerAddress()); properties.put("mail.imap.host", email.getEmailServerAddress());
properties.put("mail.imap.port",email.getPort()); properties.put("mail.imap.port",email.getPort());
properties.put("mail.imap.connectiontimeout", "3000"); // 设置连接超时时间为3秒
properties.put("mail.imap.timeout", "3000"); // 设置读取超时时间为3秒
if (email.getIsQiye() == 1) { if (email.getIsQiye() == 1) {
properties.put("mail.imap.ssl.enable", "true"); properties.put("mail.imap.ssl.enable", "true");
} else { } else {
@ -177,6 +176,7 @@ public class EmailServiceManager {
return o1.getReceivedDate().compareTo(o2.getReceivedDate()); return o1.getReceivedDate().compareTo(o2.getReceivedDate());
} catch (MessagingException e) { } catch (MessagingException e) {
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
} }
return 0; return 0;
}); });
@ -186,7 +186,7 @@ public class EmailServiceManager {
return Arrays.copyOfRange(messages,0,this.receiveNum-1); return Arrays.copyOfRange(messages,0,this.receiveNum-1);
} }
} }
} catch (MessagingException e){ } catch (Exception e){
status = EmailLogManager.STATUS_ERROR; status = EmailLogManager.STATUS_ERROR;
log.error("Email connection is abnormal, account is {}, service is {},the reason is {}.",email.getName(),email.getEmailServerAddress(),e.getMessage()); log.error("Email connection is abnormal, account is {}, service is {},the reason is {}.",email.getName(),email.getEmailServerAddress(),e.getMessage());
throw e; throw e;
@ -727,6 +727,7 @@ public class EmailServiceManager {
status = EmailLogManager.STATUS_ERROR; status = EmailLogManager.STATUS_ERROR;
log.error("Email deletion failed, the subject of the email is :{}, the reason is :{}.",subject,e.getMessage()); log.error("Email deletion failed, the subject of the email is :{}, the reason is :{}.",subject,e.getMessage());
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
}finally { }finally {
EmailLogEvent removeEvent = new EmailLogEvent(batchesCounter,Thread.currentThread().getId(),EmailLogManager.GS_TYPE_GET,status,EmailLogManager.DELETEID,subject,DateUtils.formatDate(receivedDate,"yyyy-MM-dd HH:mm:ss:SSS")); EmailLogEvent removeEvent = new EmailLogEvent(batchesCounter,Thread.currentThread().getId(),EmailLogManager.GS_TYPE_GET,status,EmailLogManager.DELETEID,subject,DateUtils.formatDate(receivedDate,"yyyy-MM-dd HH:mm:ss:SSS"));
EmailLogManager.getInstance().offer(Thread.currentThread().getId(),removeEvent); EmailLogManager.getInstance().offer(Thread.currentThread().getId(),removeEvent);
@ -748,7 +749,7 @@ public class EmailServiceManager {
if(null != store){ if(null != store){
store.close(); store.close();
} }
log.info("EmailServiceManage资源关闭完成."); log.info(Thread.currentThread().getName() + ",EmailServiceManage资源关闭完成.");
// for(String messageId : messageIds){ // for(String messageId : messageIds){
// String key = RedisConstant.EMAIL_MSG_ID+StringConstant.COLON+messageId; // String key = RedisConstant.EMAIL_MSG_ID+StringConstant.COLON+messageId;
// redisUtil.del(key); // redisUtil.del(key);
@ -756,6 +757,7 @@ public class EmailServiceManager {
} catch (MessagingException e) { } catch (MessagingException e) {
log.error("Email closure failed, email address is: {}, reason is: {}",email.getUsername(),e.getMessage()); log.error("Email closure failed, email address is: {}, reason is: {}",email.getUsername(),e.getMessage());
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
} }
} }

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

@ -1,6 +1,5 @@
package org.jeecg.modules; package org.jeecg.modules;
import cn.hutool.core.util.RandomUtil;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.constant.RedisConstant; import org.jeecg.common.constant.RedisConstant;
@ -17,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;
/** /**
@ -40,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<>();
/** /**
* 启动自动处理 * 启动自动处理
@ -82,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);
@ -120,6 +126,7 @@ public class AutoProcessManager{
TimeUnit.MILLISECONDS.sleep(sleepTime); TimeUnit.MILLISECONDS.sleep(sleepTime);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
} }
} }
} }
@ -169,6 +176,7 @@ public class AutoProcessManager{
//捕获异常不处理保障线程异常不退出 //捕获异常不处理保障线程异常不退出
}catch (Exception e){ }catch (Exception e){
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
} }
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
long sleepTime = taskProperties.getMonitoringMailCommStatusCycle() - (end-start); long sleepTime = taskProperties.getMonitoringMailCommStatusCycle() - (end-start);
@ -176,8 +184,10 @@ 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);
} }
} }
} }
@ -251,6 +261,7 @@ public class AutoProcessManager{
//捕获异常不处理保障线程异常不退出 //捕获异常不处理保障线程异常不退出
}catch (Exception e){ }catch (Exception e){
e.printStackTrace(); 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);
@ -260,6 +271,7 @@ public class AutoProcessManager{
TimeUnit.MILLISECONDS.sleep(sleepTime); TimeUnit.MILLISECONDS.sleep(sleepTime);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
} }
} }
} }
@ -277,45 +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(); }
}finally { } catch (Exception e) {
if(emailExecThread.getState() != State.TERMINATED && emailExecThread.isStop()){ e.printStackTrace();
final long nowTime = System.currentTimeMillis(); log.error(e.getMessage(), e);
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);
@ -325,6 +343,7 @@ public class AutoProcessManager{
TimeUnit.MILLISECONDS.sleep(sleepTime); TimeUnit.MILLISECONDS.sleep(sleepTime);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
} }
} }
} }

5
jeecg-module-auto-process/src/main/java/org/jeecg/modules/EmailParsingActuator.java
View File

@ -86,8 +86,8 @@ public class EmailParsingActuator extends Thread{
List<String> messageIds = new ArrayList<>(); List<String> messageIds = new ArrayList<>();
try { try {
Message[] messages = emailServiceManager.receiveMail(); Message[] messages = emailServiceManager.receiveMail();
log.info("EmailParsingActuator本次{}获取邮件数量为：{}", Thread.currentThread().getName(), ArrayUtils.isEmpty(messages) ? 0 : messages.length);
if(ArrayUtils.isNotEmpty(messages)){ if(ArrayUtils.isNotEmpty(messages)){
log.info("EmailParsingActuator本次{}获取邮件数量为：{}", Thread.currentThread().getName(), messages.length);
//检验获取的邮件是否在之前删除失败列表中，若在直接调用邮件API删除，并且此次数组里元素也删除 //检验获取的邮件是否在之前删除失败列表中，若在直接调用邮件API删除，并且此次数组里元素也删除
for(int i=messages.length-1;i>=0;i--){ for(int i=messages.length-1;i>=0;i--){
if (null == messages[i].getHeader("Message-ID")) { if (null == messages[i].getHeader("Message-ID")) {
@ -122,6 +122,7 @@ public class EmailParsingActuator extends Thread{
log.error("EmailParsingActuator has exception: {}", e.getMessage()); log.error("EmailParsingActuator has exception: {}", e.getMessage());
log.info("Mail-Parsing线程池资源关闭..."); log.info("Mail-Parsing线程池资源关闭...");
closeResource(); closeResource();
log.error(e.getMessage(), e);
throw new RuntimeException(e); throw new RuntimeException(e);
} finally { } finally {
//清除本批次邮件日志缓存 //清除本批次邮件日志缓存
@ -142,6 +143,8 @@ public class EmailParsingActuator extends Thread{
TimeUnit.MILLISECONDS.sleep(sleepTime); TimeUnit.MILLISECONDS.sleep(sleepTime);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
log.error(e.getMessage(), e);
throw new RuntimeException();
} }
} }
} }

2
jeecg-module-auto-process/src/main/java/org/jeecg/modules/spectrum/AbstractSpectrumHandler.java
View File

@ -316,7 +316,7 @@ public abstract class AbstractSpectrumHandler extends AbstractChain {
ex.printStackTrace(); ex.printStackTrace();
} }
} else if(SpectrumSource.FROM_FILE_SOURCE.getSourceType().equals(spectrumSource) && (e instanceof FileRepeatException)){ } else if(SpectrumSource.FROM_FILE_SOURCE.getSourceType().equals(spectrumSource) && (e instanceof FileRepeatException)){
this.spectrumFile.delete(); // TODO 删除原始谱文件 //this.spectrumFile.delete(); // TODO 删除原始谱文件
} else if (SpectrumSource.FORM_FILE_UNDEL.getSourceType().equals(spectrumSource) && !(e instanceof FileRepeatException)) { } else if (SpectrumSource.FORM_FILE_UNDEL.getSourceType().equals(spectrumSource) && !(e instanceof FileRepeatException)) {
try { try {
if (isDateFormatErr) { if (isDateFormatErr) {