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

onlyDownload修改邮箱处理程序频繁增加线程问题

Browse Source 
onlyDownload修改文件下载部分长期堵塞问题
This commit is contained in:
qiaoqinzheng 2024-04-26 10:31:59 +08:00
parent 7da8eeb921
commit f62158a721
4 changed files with 148 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -9,6 +9,8 @@ import com.sun.mail.imap.IMAPStore;
import com.sun.mail.smtp.SMTPAddressFailedException;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.Charsets;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.constant.RedisConstant;
@ -66,8 +68,6 @@ public class EmailServiceManager {
private RedisUtil redisUtil;
private Object receive;
private Object downloadEmlLocal = new Object();
@NotNull
@ -89,7 +89,7 @@ public class EmailServiceManager {
*/
public void init(SysEmail email, Integer receiveNum, String temporaryStoragePath,
Date systemStartupTime, SpectrumPathProperties pathProperties,TaskProperties taskProperties,
RedisUtil redisUtil, Object receive){
RedisUtil redisUtil){
this.email = email;
this.receiveNum = receiveNum;
this.temporaryStoragePath = temporaryStoragePath;
@ -97,7 +97,6 @@ public class EmailServiceManager {
this.spectrumPathProperties = pathProperties;
this.taskProperties = taskProperties;
this.redisUtil = redisUtil;
this.receive = receive;
}
/**
@ -128,79 +127,77 @@ public class EmailServiceManager {
* 接收邮件
*/
public Message[] receiveMail() {
synchronized (receive) {
String status = EmailLogManager.STATUS_SUCCESS;
try{
//配置邮件服务属性
Properties properties = new Properties();
properties.put("mail.store.protocol", "imap");
properties.put("mail.imap.host", email.getEmailServerAddress());
properties.put("mail.imap.port",email.getPort());
if (email.getIsQiye() == 1) {
properties.put("mail.imap.ssl.enable", "true");
} else {
properties.put("mail.imap.ssl.enable", "false");
}
HashMap IAM = new HashMap();
//带上IMAP ID信息，由key和value组成，例如name，version，vendor，support-email等。
IAM.put("name","myname");
IAM.put("version","1.0.0");
IAM.put("vendor","myclient");
IAM.put("support-email","testmail@test.com");
//获取邮件回话
final Session session = Session.getDefaultInstance(properties);
//获取smtp协议的存储对象
store = (IMAPStore) session.getStore();
//连接
store.connect(email.getUsername(),email.getPassword());
// 解决163普通邮箱无法建立连接问题
store.id(IAM);
//获取收件箱
folder = store.getFolder("INBOX");//INBOX
folder.open(Folder.READ_WRITE);
//如果邮箱邮件数量 > 0
final int messageCount = folder.getMessageCount();
if(messageCount > 0){
Message[] messages = null;
if(Objects.isNull(this.systemStartupTime)){
int finalNum = messageCount > this.receiveNum?this.receiveNum:messageCount;
//邮箱邮件下标是从1开始的
return folder.getMessages(1,finalNum);
}
SearchTerm searchTerm = new ReceivedDateTerm(ComparisonTerm.GE,this.systemStartupTime);
messages = folder.search(searchTerm);
Arrays.sort(messages, (o1, o2) -> {
try {
return o1.getReceivedDate().compareTo(o2.getReceivedDate());
} catch (MessagingException e) {
e.printStackTrace();
}
return 0;
});
if(this.receiveNum >= messages.length){
return messages;
}else{
return Arrays.copyOfRange(messages,0,this.receiveNum-1);
}
}
}catch (MessagingException e){
status = EmailLogManager.STATUS_ERROR;
log.error("Email connection is abnormal, account is {}, service is {},the reason is {}.",email.getName(),email.getEmailServerAddress(),e.getMessage());
e.printStackTrace();
return null;
}finally {
EmailLogEvent connectEvent = new EmailLogEvent(EmailLogManager.GS_TYPE_GET,email,status,EmailLogManager.CONNECT);
EmailLogManager.getInstance().setConnectLogEvent(connectEvent);
//GetAllId C++原业务是把远程邮箱邮件同步到C++，本次java编写没有这一步，所以和Connect绑定，若Connect成功则GetAllId成功
EmailLogEvent getAllEvent = new EmailLogEvent(EmailLogManager.GS_TYPE_GET,status,EmailLogManager.GETALLID);
EmailLogManager.getInstance().setGetAllIdLogEvent(getAllEvent);
String status = EmailLogManager.STATUS_SUCCESS;
try{
//配置邮件服务属性
Properties properties = new Properties();
properties.put("mail.store.protocol", "imap");
properties.put("mail.imap.host", email.getEmailServerAddress());
properties.put("mail.imap.port",email.getPort());
if (email.getIsQiye() == 1) {
properties.put("mail.imap.ssl.enable", "true");
} else {
properties.put("mail.imap.ssl.enable", "false");
}
HashMap IAM = new HashMap();
//带上IMAP ID信息，由key和value组成，例如name，version，vendor，support-email等。
IAM.put("name","myname");
IAM.put("version","1.0.0");
IAM.put("vendor","myclient");
IAM.put("support-email","testmail@test.com");
//获取邮件回话
final Session session = Session.getDefaultInstance(properties);
//获取smtp协议的存储对象
store = (IMAPStore) session.getStore();
//连接
store.connect(email.getUsername(),email.getPassword());
// 解决163普通邮箱无法建立连接问题
store.id(IAM);
//获取收件箱
folder = store.getFolder("INBOX");//INBOX
folder.open(Folder.READ_WRITE);
//如果邮箱邮件数量 > 0
final int messageCount = folder.getMessageCount();
if(messageCount > 0){
Message[] messages = null;
if(Objects.isNull(this.systemStartupTime)){
int finalNum = messageCount > this.receiveNum?this.receiveNum:messageCount;
//邮箱邮件下标是从1开始的
return folder.getMessages(1,finalNum);
}
SearchTerm searchTerm = new ReceivedDateTerm(ComparisonTerm.GE,this.systemStartupTime);
messages = folder.search(searchTerm);
Arrays.sort(messages, (o1, o2) -> {
try {
return o1.getReceivedDate().compareTo(o2.getReceivedDate());
} catch (MessagingException e) {
e.printStackTrace();
}
return 0;
});
if(this.receiveNum >= messages.length){
return messages;
}else{
return Arrays.copyOfRange(messages,0,this.receiveNum-1);
}
}
}catch (MessagingException e){
status = EmailLogManager.STATUS_ERROR;
log.error("Email connection is abnormal, account is {}, service is {},the reason is {}.",email.getName(),email.getEmailServerAddress(),e.getMessage());
e.printStackTrace();
return null;
}finally {
EmailLogEvent connectEvent = new EmailLogEvent(EmailLogManager.GS_TYPE_GET,email,status,EmailLogManager.CONNECT);
EmailLogManager.getInstance().setConnectLogEvent(connectEvent);
//GetAllId C++原业务是把远程邮箱邮件同步到C++，本次java编写没有这一步，所以和Connect绑定，若Connect成功则GetAllId成功
EmailLogEvent getAllEvent = new EmailLogEvent(EmailLogManager.GS_TYPE_GET,status,EmailLogManager.GETALLID);
EmailLogManager.getInstance().setGetAllIdLogEvent(getAllEvent);
}
return null;
}
/*
@ -537,6 +534,7 @@ public class EmailServiceManager {
File emlFile = null;
String status = EmailLogManager.STATUS_SUCCESS;
Date receivedDate = null;
InputStream inputStream = null;
BufferedOutputStream outputStream = null;
try {
//获取发件人
@ -571,26 +569,19 @@ public class EmailServiceManager {
final String rootPath = spectrumPathProperties.getRootPath();
final String emlPath = spectrumPathProperties.getEmlPath();
emlFile = new File(rootPath+emlPath+File.separator+fileName);
outputStream = new BufferedOutputStream(new FileOutputStream(emlFile));
message.writeTo(outputStream);
// 刷新缓冲区内容存入内存
outputStream.flush();
// int bufferSize = 1024 * 1024; // 1M
// InputStream inputStream = message.getInputStream();
// BufferedInputStream bufferedInputStream = new BufferedInputStream(inputStream, bufferSize);
// // 或者使用 BufferedOutputStream
// OutputStream outputStream = new FileOutputStream(emlFile);
// BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(outputStream, bufferSize);
// // 从邮件的输入流读取内容，并写入到本地文件
// byte[] buffer = new byte[bufferSize];
// int bytesRead;
// while ((bytesRead = bufferedInputStream.read(buffer)) != -1) {
// bufferedOutputStream.write(buffer, 0, bytesRead);
// }
//
// // 关闭流
// bufferedInputStream.close();
// bufferedOutputStream.close();
// outputStream = new FileOutputStream(emlFile);
// message.writeTo(outputStream);
int bufferSize = 1024 * 1024; // 1M
inputStream = message.getInputStream();
outputStream = new BufferedOutputStream(new FileOutputStream(emlFile), bufferSize);
// 从邮件的输入流读取内容，并写入到本地文件
byte[] buffer = new byte[bufferSize];
int bytesRead;
while ((bytesRead = inputStream.read(buffer)) != -1) {
outputStream.write(buffer, 0, bytesRead);
}
} catch (MessagingException | IOException e) {
// 下载邮件失败 抛出自定义邮件下载异常
status = EmailLogManager.STATUS_ERROR;
@ -604,7 +595,11 @@ public class EmailServiceManager {
(Objects.isNull(emlFile)?" ":emlFile.getAbsolutePath()));
EmailLogManager.getInstance().offer(Thread.currentThread().getId(),event);
try {
if (Objects.nonNull(inputStream)) {
inputStream.close();
}
if (Objects.nonNull(outputStream)) {
outputStream.flush();
outputStream.close();
}
} catch (IOException e) {

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

@ -37,10 +37,6 @@ public class AutoProcessManager{
* 邮件Map数据锁
*/
private final Object lock = new Object();
/**
* 邮箱接收邮件锁
*/
private final Object receive = new Object();
/**
* 以邮件Id为key，邮件信息为value
*/
@ -103,7 +99,7 @@ public class AutoProcessManager{
boolean testFlag = emailServiceManager.testConnectEmailServer();
if(testFlag){
EmailParsingActuator emailParsingActuator = new EmailParsingActuator();
emailParsingActuator.init(next,spectrumServiceQuotes,emailCounter,systemStartupTime, receive);
emailParsingActuator.init(next,spectrumServiceQuotes,emailCounter,systemStartupTime);
emailParsingActuator.setName(next.getUsername()+"-email-monitor");
emailParsingActuator.start();
//把邮件监测执行线程加入管理队列
@ -237,9 +233,15 @@ public class AutoProcessManager{
if(databaseEmail.getEnabled().equals(SysMailEnableType.ENABLE.getMailEnableType())){
final boolean testFlag = testConnectEmailServer(databaseEmail);
if(testFlag){
databaseEmail.setNewEmailFlag(true);
if (emailExecThreadMap.containsKey(databaseEmail.getId())) {
EmailParsingActuator actuator = emailExecThreadMap.get(databaseEmail.getId());
actuator.setStop(false);
log.info("{}邮箱重新加入监测队列",databaseEmail.getUsername());
} else {
databaseEmail.setNewEmailFlag(true);
log.info("{}邮箱加入监测队列，设置新增标记",databaseEmail.getUsername());
}
emailMap.put(databaseEmail.getId(),databaseEmail);
log.info("{}邮箱加入监测队列，设置新增标记",databaseEmail.getUsername());
}
}
}

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

@ -36,20 +36,18 @@ public class EmailParsingActuator extends Thread{
private SpectrumServiceQuotes spectrumServiceQuotes;
private EmailCounter emailCounter;
private Date systemStartupTime;
private Object receive;
@Setter @Getter
private boolean isStop;
@Setter @Getter
private Date stopTime;
public void init(EmailProperties emailProperties,SpectrumServiceQuotes spectrumServiceQuotes,
EmailCounter emailCounter,Date systemStartupTime, Object receive){
EmailCounter emailCounter,Date systemStartupTime){
this.emailProperties = emailProperties;
this.spectrumServiceQuotes = spectrumServiceQuotes;
this.taskProperties = spectrumServiceQuotes.getTaskProperties();
this.emailCounter = emailCounter;
this.systemStartupTime = systemStartupTime;
this.receive = receive;
//获取机器可用核心数
int systemCores = spectrumServiceQuotes.getMaximumPoolSizeProperties().getAuto();
@ -76,21 +74,21 @@ public class EmailParsingActuator extends Thread{
long start = System.currentTimeMillis();
final EmailServiceManager emailServiceManager = EmailServiceManager.getInstance();
emailServiceManager.init(this.emailProperties,this.taskProperties.getReceiveNum(),this.taskProperties.getTemporaryStoragePath(),
this.systemStartupTime, spectrumServiceQuotes.getSpectrumPathProperties(), spectrumServiceQuotes.getTaskProperties(), spectrumServiceQuotes.getRedisUtil(), receive);
this.systemStartupTime, spectrumServiceQuotes.getSpectrumPathProperties(), spectrumServiceQuotes.getTaskProperties(), spectrumServiceQuotes.getRedisUtil());
List<String> messageIds = new ArrayList<>();
try {
Message[] messages = emailServiceManager.receiveMail();
if(ArrayUtils.isNotEmpty(messages)){
log.info("EmailParsingActuator本次获取邮件数量为：{}",messages.length);
log.info("EmailParsingActuator本次{}获取邮件数量为：{}", Thread.currentThread().getName(), messages.length);
//检验获取的邮件是否在之前删除失败列表中，若在直接调用邮件API删除，并且此次数组里元素也删除
for(int i=messages.length-1;i>=0;i--){
if (!messages[i].isExpunged()){
String messageId = ((MimeMessage) messages[i]).getMessageID();
final boolean exist = emailServiceManager.check(messages[i],messageId);
// final boolean exist = emailServiceManager.check(messages[i],messageId);
messageIds.add(messageId);
if(exist){
messages = ArrayUtils.remove(messages,i);
}
// if(exist){
// messages = ArrayUtils.remove(messages,i);
// }
}
}
log.info("EmailParsingActuator本次真实执行邮件数量为：{}",messages.length);
@ -107,11 +105,12 @@ public class EmailParsingActuator extends Thread{
taskLatch.await();
}
}
}catch (InterruptedException e) {
} catch (InterruptedException e) {
e.printStackTrace();
} catch (MessagingException e) {
throw new RuntimeException(e);
} finally {
log.info(Thread.currentThread().getName() +"执行完毕！！！！");
//清除本批次邮件日志缓存
EmailLogManager.getInstance().clear();
//保存本批次所有能谱日志

82
jeecg-module-auto-process/src/main/java/org/jeecg/modules/spectrum/SpectrumParsingActuator.java
View File

@ -110,47 +110,47 @@ public class SpectrumParsingActuator implements Runnable{
//保存邮件日志到PG数据库
this.spectrumServiceQuotes.getMailLogService().create(message,emailProperties);
//获取邮件内容
StringBuilder mailContent = new StringBuilder();
if(Objects.nonNull(emlFile) && emlFile.length() > 0){
mailContent.append(FileUtil.readUtf8String(emlFile));
}
//判断是否是IMS2.0协议文件
// 如果邮件内容校验成功 将文件保存到eml目录 并删除邮件对象
if(checkMailContent(mailContent,subject)){
AbstractSpectrumHandler spectrumHandler = new SamplephdSpectrum();
spectrumHandler.init(mailContent.toString(),emlFile.getName(),spectrumServiceQuotes,new StringBuilder(),SpectrumSource.FORM_EMAIL_SERVICE.getSourceType(),batchesCounter);
final boolean matchResult = spectrumHandler.saveEmailToLocal();
if(matchResult){
try {
//开始解析
spectrumHandler.handler();
spectrumServiceQuotes.getRedisUtil().del(key);
} catch (Exception e) {
//如果是gamma谱的分析异常
if (e instanceof AnalySpectrumException) {
// 如果邮件内容校验失败(邮件内容不完整) 将错误邮件从eml移动到emlError
if (Objects.nonNull(emlFile) && emlFile.exists()){
moveEmail(emlFile, key);
}
//删除邮件
emailServiceManager.removeMail(message,batchesCounter);
} else {
throw e;
}
}
}else{
log.warn("This email {} parsing failed and is not listed in the Met, Alert, SOH, Sample, Detbkphd, QC, Gasbkphd spectra.",subject);
}
emailServiceManager.removeMail(message,batchesCounter);
}else {
// 如果邮件内容校验失败(邮件内容不完整) 将错误邮件从eml移动到emlError
if (Objects.nonNull(emlFile) && emlFile.exists()){
moveEmail(emlFile, key);
throw new DownloadEmailException("邮件移走后手动抛出DownloadEmailException");
}
}
// //获取邮件内容
// StringBuilder mailContent = new StringBuilder();
// if(Objects.nonNull(emlFile) && emlFile.length() > 0){
// mailContent.append(FileUtil.readUtf8String(emlFile));
// }
//
// //判断是否是IMS2.0协议文件
// // 如果邮件内容校验成功 将文件保存到eml目录 并删除邮件对象
// if(checkMailContent(mailContent,subject)){
// AbstractSpectrumHandler spectrumHandler = new SamplephdSpectrum();
// spectrumHandler.init(mailContent.toString(),emlFile.getName(),spectrumServiceQuotes,new StringBuilder(),SpectrumSource.FORM_EMAIL_SERVICE.getSourceType(),batchesCounter);
// final boolean matchResult = spectrumHandler.saveEmailToLocal();
// if(matchResult){
// try {
// //开始解析
// spectrumHandler.handler();
// spectrumServiceQuotes.getRedisUtil().del(key);
// } catch (Exception e) {
// //如果是gamma谱的分析异常
// if (e instanceof AnalySpectrumException) {
// // 如果邮件内容校验失败(邮件内容不完整) 将错误邮件从eml移动到emlError
// if (Objects.nonNull(emlFile) && emlFile.exists()){
// moveEmail(emlFile, key);
// }
// //删除邮件
// emailServiceManager.removeMail(message,batchesCounter);
// } else {
// throw e;
// }
// }
// }else{
// log.warn("This email {} parsing failed and is not listed in the Met, Alert, SOH, Sample, Detbkphd, QC, Gasbkphd spectra.",subject);
// }
// emailServiceManager.removeMail(message,batchesCounter);
// }else {
// // 如果邮件内容校验失败(邮件内容不完整) 将错误邮件从eml移动到emlError
// if (Objects.nonNull(emlFile) && emlFile.exists()){
// moveEmail(emlFile, key);
// throw new DownloadEmailException("邮件移走后手动抛出DownloadEmailException");
// }
// }
} catch (Exception e) {
// 如果不是下载导致的失败 并且 下载成功，则删除下载的邮件对象
if(!(e instanceof DownloadEmailException) && downloadFlag){