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

修改获取邮件方式

Browse Source
This commit is contained in:
duwenyuan 2025-07-16 09:06:32 +08:00
parent a0cff20da8
commit 9d8dca0948
3 changed files with 236 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,11 +1,17 @@
package org.jeecg.common.email;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
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 com.google.common.collect.Lists;
import com.sun.mail.imap.IMAPStore;
import com.sun.mail.pop3.POP3Folder;
import com.sun.mail.smtp.SMTPAddressFailedException;
import lombok.extern.slf4j.Slf4j;
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;
@ -30,7 +36,9 @@ import javax.mail.search.SearchTerm;
import java.io.*;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.file.Files;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
@ -61,7 +69,8 @@ public class EmailServiceManager {
/**
* smtp协议的存储对象
*/
private IMAPStore store = null;
//private IMAPStore store = null;
private Store store = null;
/**
* 邮件附件临时存储路径
*/
@ -70,7 +79,8 @@ public class EmailServiceManager {
* 收件箱
*/
private Folder folder = null;
// 判断协议类型
private String protocol;
private RedisUtil redisUtil;
private Object downloadEmlLocal = new Object();
@ -135,7 +145,7 @@ public class EmailServiceManager {
/**
* 接收邮件
*/
public Message[] receiveMail() throws Exception {
public Message[] receiveMail1() throws Exception {
String status = EmailLogManager.STATUS_SUCCESS;
try {
//配置邮件服务属性
@ -167,9 +177,11 @@ public class EmailServiceManager {
//连接
store.connect(email.getUsername(), email.getPassword());
if (email.getEmailServerType() == 1) {
// 解决163普通邮箱无法建立连接问题
store.id(IAM);
((IMAPStore) store).id(IAM);
}
//获取收件箱
folder = store.getFolder("INBOX");//INBOX
folder.open(Folder.READ_WRITE);
@ -213,6 +225,134 @@ public class EmailServiceManager {
return null;
}
public Message[] receiveMail() throws Exception {
String status = EmailLogManager.STATUS_SUCCESS;
try {
store = setStoreProperties();
//获取收件箱
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 {
if (o1.getReceivedDate() == null && o2.getReceivedDate() == null) {
return o1.getSentDate().compareTo(o2.getSentDate());
} else {
return o1.getReceivedDate().compareTo(o2.getReceivedDate());
}
} catch (MessagingException e) {
e.printStackTrace();
log.error(e.getMessage(), e);
}
return 0;
});
if (this.receiveNum >= messages.length) {
return messages;
} else {
return Arrays.copyOfRange(messages, 0, this.receiveNum - 1);
}
}
} catch (Exception e) {
status = EmailLogManager.STATUS_ERROR;
log.error("Email connection is abnormal, account is {}, service is {},the reason is {}.", email.getName(), email.getEmailServerAddress(), e.getMessage());
throw e;
} 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;
}
private Store setStoreProperties() throws MessagingException {
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");
Properties properties = getProperties();
//获取邮件回话
Session session = Session.getInstance(properties);
//获取smtp协议的存储对象
Store store = session.getStore();
log.info("连接 {}:{}:{}", email.getUsername(), email.getPassword(), email.getPort());
//连接
store.connect(email.getUsername(), email.getPassword());
if (email.getEmailServerType() == 1) {
// 解决163普通邮箱无法建立连接问题
((IMAPStore) store).id(IAM);
}
protocol = store.getURLName().getProtocol();
return store;
}
@NotNull
private Properties getProperties() {
Properties properties = new Properties();
switch (email.getPort()) {
case 110:
properties.put("mail.store.protocol", "pop3");
properties.put("mail.pop3.host", email.getEmailServerAddress());
properties.put("mail.pop3.port", email.getPort());
properties.put("mail.pop3.starttls.enable", "true");
// 超时设置(毫秒)
properties.put("mail.pop3.connectiontimeout", "10000");
properties.put("mail.pop3.timeout", "15000");
break;
case 995:
properties.put("mail.store.protocol", "pop3");
properties.put("mail.pop3.host", email.getEmailServerAddress());
properties.put("mail.pop3.port", email.getPort());
properties.put("mail.pop3.starttls.enable", "true");
properties.put("mail.pop3.connectiontimeout", "10000");
properties.put("mail.pop3.timeout", "15000");
properties.put("mail.pop3.auth", "true");
break;
case 143:
properties.put("mail.store.protocol", "imap");
properties.put("mail.imap.host", email.getEmailServerAddress());
properties.put("mail.imap.port", email.getPort());
properties.put("mail.imap.ssl.enable", "false");
properties.put("mail.imap.starttls.enable", "true");
properties.put("mail.imap.auth", "true");
// 超时设置(毫秒)
properties.put("mail.imap.connectiontimeout", "10000");
properties.put("mail.imap.timeout", "30000");
break;
case 993:
default:
properties.put("mail.store.protocol", "imap");
properties.put("mail.imap.host", email.getEmailServerAddress());
properties.put("mail.imap.port", email.getPort());
properties.put("mail.imap.connectiontimeout", "3000");
properties.put("mail.imap.timeout", "3000");
if (email.getIsQiye() == 1) {
properties.put("mail.imap.ssl.enable", "true");
} else {
properties.put("mail.imap.ssl.enable", "false");
}
break;
}
return properties;
}
/*
* 测试收件邮箱账号是否可以正常使用
* */
@ -222,10 +362,7 @@ public class EmailServiceManager {
String password = email.getPassword();
String host = email.getEmailServerAddress();
Properties props = new Properties();
props.put("mail.store.protocol", "imap");
props.put("mail.imap.host", host);
props.put("mail.imap.port", port);
Properties props = getProperties();
Session session = Session.getInstance(props, new Authenticator() {
@Override
@ -477,6 +614,31 @@ public class EmailServiceManager {
return subject;
}
public String getMessagesID(Message message, Integer batchesCounter) throws MessagingException {
String messageId = null;
String status = EmailLogManager.STATUS_SUCCESS;
try {
if (null == message.getHeader("Message-ID")) {
String subject = message.getSubject().replace(" ", StringConstant.UNDER_LINE);
Date date = message.getReceivedDate() == null ? message.getSentDate() : message.getReceivedDate();
String receivedStr = DateUtil.format(date, DatePattern.NORM_DATETIME_MINUTE_PATTERN);
messageId = subject + StringConstant.UNDER_LINE + receivedStr;
} else {
messageId = ((MimeMessage) message).getMessageID();
}
} catch (MessagingException e) {
status = EmailLogManager.STATUS_ERROR;
log.error(e.getMessage());
throw e;
}finally {
EmailLogEvent event = new EmailLogEvent(batchesCounter, Thread.currentThread().getId(), EmailLogManager.GS_TYPE_GET, status, EmailLogManager.GETIDHEADER);
EmailLogManager.getInstance().offer(Thread.currentThread().getId(), event);
}
return messageId;
}
/**
* 获取邮件内容
*
@ -550,8 +712,8 @@ public class EmailServiceManager {
File emlFile = null;
String status = EmailLogManager.STATUS_SUCCESS;
Date receivedDate = null;
InputStream inputStream = null;
BufferedOutputStream outputStream = null;
//InputStream inputStream = null;
//BufferedOutputStream outputStream = null;
try {
//获取发件人
final String address = ((InternetAddress) message.getFrom()[0]).getAddress();
@ -564,7 +726,7 @@ public class EmailServiceManager {
if (subject.indexOf(StringConstant.COLON) != -1) {
subject = StringUtils.replace(subject, StringConstant.COLON, "");
}
receivedDate = message.getReceivedDate();
receivedDate = message.getReceivedDate() == null ? message.getSentDate() : message.getReceivedDate();
StringBuilder fileName = new StringBuilder();
fileName.append(from);
fileName.append(StringConstant.UNDER_LINE);
@ -585,19 +747,21 @@ public class EmailServiceManager {
final String rootPath = spectrumPathProperties.getRootPath();
final String emlPath = spectrumPathProperties.getEmlPath();
emlFile = new File(rootPath + emlPath + File.separator + fileName);
// 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);
try (InputStream inputStream = message.getInputStream();
BufferedOutputStream outputStream = new BufferedOutputStream(new FileOutputStream(emlFile),bufferSize)) {
byte[] buffer = new byte[1024*1024];
int bytesRead;
while ((bytesRead = inputStream.read(buffer)) != -1) {
outputStream.write(buffer, 0, bytesRead);
}
outputStream.flush(); // 显式刷新
} catch (IOException e) {
FileUtils.copyInputStreamToFile(message.getInputStream(), emlFile);
}
if (emlFile.length() <= 0) {
FileUtils.copyInputStreamToFile(message.getInputStream(), emlFile);
}
} catch (MessagingException | IOException e) {
// 下载邮件失败 抛出自定义邮件下载异常
status = EmailLogManager.STATUS_ERROR;
@ -606,21 +770,11 @@ public class EmailServiceManager {
throw new DownloadEmailException(errorMsg);
} catch (Exception e) {
log.error("", e);
throw new RuntimeException(e);
} finally {
EmailLogEvent event = new EmailLogEvent(batchesCounter, Thread.currentThread().getId(), EmailLogManager.GS_TYPE_GET, status, EmailLogManager.GETIDEML, subject, DateUtils.formatDate(receivedDate, "yyyy-MM-dd HH:mm:ss:SSS"),
(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) {
throw new RuntimeException(e);
}
}
return emlFile;
}
@ -679,7 +833,7 @@ public class EmailServiceManager {
if (subject.contains(StringConstant.COLON)) {
subject = StringUtils.replace(subject, StringConstant.COLON, "");
}
receivedDate = message.getReceivedDate();
receivedDate = message.getReceivedDate() == null ? message.getSentDate() : message.getReceivedDate();
StringBuilder fileName = new StringBuilder();
fileName.append(from);
fileName.append(StringConstant.UNDER_LINE);
@ -738,13 +892,14 @@ public class EmailServiceManager {
Date receivedDate = null;
try {
subject = MimeUtility.decodeText(message.getSubject());
receivedDate = message.getReceivedDate();
receivedDate = message.getReceivedDate() == null ? message.getSentDate() : message.getReceivedDate();
message.setFlag(Flags.Flag.DELETED, true);
// log.info("EmailServiceManager: Remove Email:{},receiveTime:{}",message.getSubject(), DateUtils.formatDate(message.getReceivedDate(),"yyyy-MM-dd HH:mm:ss"));
} catch (MessagingException | UnsupportedEncodingException e) {
status = EmailLogManager.STATUS_ERROR;
log.error("Email deletion failed, the subject of the email is :{}, the reason is :{}.", subject, e.getMessage());
e.printStackTrace();
} catch (Exception e) {
log.error(e.getMessage(), e);
} 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"));
@ -761,21 +916,22 @@ public class EmailServiceManager {
public void close(List<String> messageIds) {
try {
if (null != folder) {
folder.expunge();
if ("imap".equalsIgnoreCase(protocol)) {
folder.expunge();
}
folder.close();
}
if (null != store) {
store.close();
}
log.info(Thread.currentThread().getName() + ",EmailServiceManage资源关闭完成.");
// for(String messageId : messageIds){
// String key = RedisConstant.EMAIL_MSG_ID+StringConstant.COLON+messageId;
// redisUtil.del(key);
// }
log.info("{}: EmailServiceManage资源关闭完成.", Thread.currentThread().getName());
for (String messageId : messageIds) {
String key = RedisConstant.EMAIL_MSG_ID + StringConstant.COLON + messageId;
redisUtil.del(key);
}
} catch (MessagingException e) {
log.error("Email closure failed, email address is: {}, reason is: {}", email.getUsername(), e.getMessage());
e.printStackTrace();
log.error(e.getMessage(), e);
}
}
@ -793,7 +949,8 @@ public class EmailServiceManager {
// exist = redisUtil.hasKey(key);
if (numberKey >= taskProperties.getForceDeletedNumber()) {
exist = true;
log.info("Check: Remove Email:{},receiveTime:{}", message.getSubject(), DateUtils.formatDate(message.getReceivedDate(), "yyyy-MM-dd HH:mm:ss"));
Date dateUtils = message.getReceivedDate() == null ? message.getSentDate() : message.getReceivedDate();
log.info("Check: Remove Email:{},receiveTime:{}", message.getSubject(), DateUtils.formatDate(dateUtils, "yyyy-MM-dd HH:mm:ss"));
message.setFlag(Flags.Flag.DELETED, true);
redisUtil.del(key);
}

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

@ -17,6 +17,7 @@ import org.jeecg.modules.spectrum.SpectrumLogManager;
import org.jeecg.modules.spectrum.SpectrumParsingActuator;
import org.jeecg.modules.spectrum.SpectrumServiceQuotes;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;
@ -27,7 +28,7 @@ import java.util.concurrent.*;
* 邮件解析执行器
*/
@Slf4j
public class EmailParsingActuator extends Thread{
public class EmailParsingActuator extends Thread {
private TaskProperties taskProperties;
@Getter
@ -36,15 +37,18 @@ public class EmailParsingActuator extends Thread{
private SpectrumServiceQuotes spectrumServiceQuotes;
private EmailCounter emailCounter;
private Date systemStartupTime;
@Setter @Getter
@Setter
@Getter
private boolean isStop;
@Setter @Getter
@Setter
@Getter
private boolean threadSleep;
@Setter @Getter
@Setter
@Getter
private Date stopTime;
public void init(EmailProperties emailProperties,SpectrumServiceQuotes spectrumServiceQuotes,
EmailCounter emailCounter,Date systemStartupTime){
public void init(EmailProperties emailProperties, SpectrumServiceQuotes spectrumServiceQuotes,
EmailCounter emailCounter, Date systemStartupTime) {
this.emailProperties = emailProperties;
this.spectrumServiceQuotes = spectrumServiceQuotes;
this.taskProperties = spectrumServiceQuotes.getTaskProperties();
@ -53,11 +57,11 @@ public class EmailParsingActuator extends Thread{
//获取机器可用核心数
int systemCores = spectrumServiceQuotes.getMaximumPoolSizeProperties().getAuto();
int maximumPoolSize = taskProperties.getReceiveNum() > systemCores?taskProperties.getReceiveNum():systemCores;
int maximumPoolSize = taskProperties.getReceiveNum() > systemCores ? taskProperties.getReceiveNum() : systemCores;
//初始化线程池
ThreadFactory threadFactory = new CustomizableThreadFactory("mail-parsing-");
poolExecutor = new ThreadPoolExecutor(taskProperties.getReceiveNum(),maximumPoolSize,5, TimeUnit.SECONDS, new LinkedBlockingQueue<>(),threadFactory);
poolExecutor = new ThreadPoolExecutor(taskProperties.getReceiveNum(), maximumPoolSize, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), threadFactory);
}
public void updateEmail(EmailProperties emailProperties) {
@ -66,10 +70,10 @@ public class EmailParsingActuator extends Thread{
@Override
public void run() {
for(;;){
for (; ; ) {
String nowDate = DateUtils.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss");
if (threadSleep) {
log.info(nowDate + " " +this.emailProperties.getName()+" EmailParsingActuator is sleep!");
log.info(nowDate + " " + this.emailProperties.getName() + " EmailParsingActuator is sleep!");
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
@ -78,49 +82,50 @@ public class EmailParsingActuator extends Thread{
continue;
}
if (isStop) {
log.info(nowDate + " " +this.emailProperties.getName()+" EmailParsingActuator is Stop!");
log.info(nowDate + " " + this.emailProperties.getName() + " EmailParsingActuator is Stop!");
closeResource();
return;
}
long start = System.currentTimeMillis();
final EmailServiceManager emailServiceManager = EmailServiceManager.getInstance();
emailServiceManager.init(this.emailProperties,this.taskProperties.getReceiveNum(),this.taskProperties.getTemporaryStoragePath(),
emailServiceManager.init(this.emailProperties, this.taskProperties.getReceiveNum(), this.taskProperties.getTemporaryStoragePath(),
this.systemStartupTime, spectrumServiceQuotes.getSpectrumPathProperties(), spectrumServiceQuotes.getTaskProperties(), spectrumServiceQuotes.getRedisUtil());
List<String> messageIds = new ArrayList<>();
try {
Message[] messages = emailServiceManager.receiveMail();
log.info("EmailParsingActuator本次{}获取邮件数量为：{}", Thread.currentThread().getName(), ArrayUtils.isEmpty(messages) ? 0 : messages.length);
if(ArrayUtils.isNotEmpty(messages)){
if (ArrayUtils.isNotEmpty(messages)) {
//检验获取的邮件是否在之前删除失败列表中，若在直接调用邮件API删除，并且此次数组里元素也删除
for(int i=messages.length-1;i>=0;i--){
for (int i = messages.length - 1; i >= 0; i--) {
String messageId = null;
if (null == messages[i].getHeader("Message-ID")) {
// 有些邮箱拿不到 message-ID，换成主题+接收时间
String subject = messages[i].getSubject().replace(" ", StringConstant.UNDER_LINE);
String receivedStr = DateUtil.format(messages[i].getReceivedDate(), DatePattern.NORM_DATETIME_MINUTE_PATTERN);
Date date = messages[i].getReceivedDate() == null ? messages[i].getSentDate() : messages[i].getReceivedDate();
String receivedStr = DateUtil.format(date, DatePattern.NORM_DATETIME_MINUTE_PATTERN);
messageId = subject + StringConstant.UNDER_LINE + receivedStr;
// messages = ArrayUtils.remove(messages, i);
// continue;
} else {
messageId = ((MimeMessage) messages[i]).getMessageID();
}
if (!messages[i].isExpunged()){
if (!messages[i].isExpunged()) {
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);
if(messages.length > 0){
log.info("EmailParsingActuator本次真实执行邮件数量为：{}", messages.length);
if (messages.length > 0) {
//本批次邮件号
final Integer batchesCounter = spectrumServiceQuotes.getBatchesCounter().getCurrValue();
CountDownLatch taskLatch = new CountDownLatch(messages.length);
for(Message message : messages){
for (Message message : messages) {
SpectrumParsingActuator spectrumParsingActuator = new SpectrumParsingActuator();
spectrumParsingActuator.init(message,emailProperties,emailServiceManager,
taskLatch,spectrumServiceQuotes,emailCounter,batchesCounter);
spectrumParsingActuator.init(message, emailProperties, emailServiceManager,
taskLatch, spectrumServiceQuotes, emailCounter, batchesCounter);
poolExecutor.execute(spectrumParsingActuator);
}
taskLatch.await();
@ -144,9 +149,9 @@ public class EmailParsingActuator extends Thread{
emailServiceManager.close(messageIds);
}
long end = System.currentTimeMillis();
long sleepTime = taskProperties.getMailThreadExecCycle() - (end-start);
long sleepTime = taskProperties.getMailThreadExecCycle() - (end - start);
//如果sleepTime > 0 需要睡眠到指定时间，否则继续下次获取邮件
if(sleepTime > 0){
if (sleepTime > 0) {
try {
//如果本次
TimeUnit.MILLISECONDS.sleep(sleepTime);
@ -162,8 +167,8 @@ public class EmailParsingActuator extends Thread{
/**
* 立即关闭线程池
*/
protected void closeResource(){
if(Objects.nonNull(poolExecutor)) {
protected void closeResource() {
if (Objects.nonNull(poolExecutor)) {
poolExecutor.shutdownNow();
}
}

8
setup/SelfParameter.xml
View File

@ -1,8 +1,8 @@
<config>
<nuclide>
<item nuclide_name="Xe131M" energy="29.7" yield="44.362" half_life="11.934" />
<item nuclide_name="Xe133M" energy="29.7" yield="46.049" half_life="2.19" />
<item nuclide_name="Xe133" energy="81.0" yield="38.0" half_life="5.243" />
<item nuclide_name="Xe135" energy="249.8" yield="90.0" half_life="0.380832" />
<item nuclide_name="Xe131M" energy="29.6" yield="44.362" half_life="11.934" range="0.2"/>
<item nuclide_name="Xe133M" energy="29.6" yield="46.049" half_life="2.19" range="0.2"/>
<item nuclide_name="Xe133" energy="81.0" yield="38.0" half_life="5.243" range="0.5"/>
<item nuclide_name="Xe135" energy="249.8" yield="90.0" half_life="0.380832" range="0.5"/>
</nuclide>
</config>