多线程环境下JavaMailSenderImpl的重连机制及批量发送优化问询

我们每日通过JavaMailSenderImpl发送数千封邮件，此前采用私有方法实例化该类发送单封邮件，现已改为Spring Bean形式使用，因存在多个邮件主机，配置了多个Bean。

现需咨询两个核心问题：

1. 多线程场景下，批量发送邮件前如何正确检查Session/Transport并执行重连？
2. 若批量发送过程中JavaMailSenderImpl的Session意外断开，最优处理方案是什么？

相关代码 #

@Bean("myMailSender")
public JavaMailSender myMailSender(){
   JavaMailSenderImpl javaMailSender = new JavaMailSenderImpl();
   javaMailSender.setHost(host);
   javaMailSender.setPort(port);
   javaMailSender.setUsername(username); // 修正原代码拼写错误
   javaMailSender.setPassword(password);
   return javaMailSender;
}

ThreadPoolExecutor executorService = (ThreadPoolExecutor) Executors.newFixedThreadPool(10, new NamedThreadFactory("ParallelEmail-"));

try {
   Session session = mailSender.getSession();
   if (session.getTransport() == null || !session.getTransport().isConnected()) {
       Transport t = null;
       try {
           t = session.getTransport();
           t.connect(mailSender.getHost(), mailSender.getPort(), mailSender.getUsername(), mailSender.getPassword());
       } catch (MessagingException e){
           log.error("There was an error when re-connecting to mailhost " + mailSender.getHost());
       }
   }
} catch (NoSuchProviderException e1){
   log.error("There was an error when fetching an existing session for mailhost " + mailSender.getHost());
}

for (Email msg: getFromDb()) {
   executorService.execute(
           () -> {
               MimeMessagePreparator messagePreparator = getMsgPrep(msg);
               try{
                   mailSender.send(messagePreparator);
               } catch (MailException exception){
                   log.error(exception.getMessage());
                   //Reattempt for second time?
                   //I don’t want all threads to start reconnecting, just the first one which gets to this line.
               }
           }
   );
}

问题解答 #

一、批量发送前的Session/Transport检查与重连优化 #

你的前置检查逻辑存在多线程安全问题：多个线程可能同时进入检查逻辑，重复执行连接操作，而且session.getTransport()本身并非线程安全的调用。正确的做法是：

• 避免手动提前连接Session：JavaMailSenderImpl的send方法本身会自动处理Session的连接与复用，手动提前连接反而容易引发多线程冲突。如果确实需要预检查，必须加线程同步控制：

  // 给每个mailSender Bean加专属锁，避免不同主机的Bean互相阻塞
  private final Object lock = new Object();
  
  // 预检查逻辑改为同步块
  synchronized (lock) {
      Session session = mailSender.getSession();
      Transport transport = null;
      try {
          transport = session.getTransport();
          if (transport != null && !transport.isConnected()) {
              transport.connect(mailSender.getHost(), mailSender.getPort(), mailSender.getUsername(), mailSender.getPassword());
          }
      } catch (MessagingException e) {
          log.error("连接邮件服务器{}失败", mailSender.getHost(), e);
          // 预连接失败直接抛出异常，终止批量任务，避免后续无效发送
          throw new RuntimeException("邮件服务器预连接失败", e);
      } finally {
          // 注意：不要手动关闭transport，JavaMailSender会自行管理
      }
  }
  

• 利用JavaMail的内置连接池：通过Session的属性配置开启连接池，让JavaMail自动管理连接的复用与重连，无需手动干预：

  @Bean("myMailSender")
  public JavaMailSender myMailSender(){
      JavaMailSenderImpl javaMailSender = new JavaMailSenderImpl();
      javaMailSender.setHost(host);
      javaMailSender.setPort(port);
      javaMailSender.setUsername(username);
      javaMailSender.setPassword(password);
      
      // 配置JavaMail连接池属性
      Properties props = new Properties();
      props.put("mail.smtp.connectiontimeout", 5000);
      props.put("mail.smtp.timeout", 5000);
      props.put("mail.smtp.writetimeout", 5000);
      props.put("mail.smtp.auth", "true");
      // 开启连接池
      props.put("mail.smtp.socketFactory.class", "javax.net.ssl.SSLSocketFactory");
      props.put("mail.smtp.keepalive", "true"); // 保持连接
      props.put("mail.smtp.maxconnections", 10); // 对应线程池大小的连接数
      
      javaMailSender.setJavaMailProperties(props);
      return javaMailSender;
  }
  

二、Session意外断开的最优处理方案 #

当批量发送中Session断开时，核心目标是避免所有线程重复重连、保证消息不丢失、快速恢复发送：

1. 异常分层处理，加全局重连锁：
   在异常捕获块中，使用一个全局的原子标记来控制只有第一个遇到断开异常的线程执行重连操作：

   // 类级别原子变量，标记是否正在重连
   private final AtomicBoolean isReconnecting = new AtomicBoolean(false);
   
   // 发送逻辑中的异常处理
   try{
       mailSender.send(messagePreparator);
   } catch (MailException exception){
       // 判断是否是连接断开类异常（比如NoSuchProviderException、ConnectException等）
       if (isConnectionFailure(exception)) {
           // 只有第一个线程能进入重连逻辑
           if (isReconnecting.compareAndSet(false, true)) {
               try {
                   // 执行重连操作
                   reconnectMailSender(mailSender);
                   // 重连成功后，重试当前消息
                   mailSender.send(messagePreparator);
               } catch (Exception e) {
                   log.error("重连邮件服务器{}失败，消息{}发送失败", mailSender.getHost(), msg.getId(), e);
                   // 重连失败，将消息放入死信队列或数据库待重试
                   saveToRetryQueue(msg);
               } finally {
                   isReconnecting.set(false);
               }
           } else {
               // 其他线程等待重连完成后重试
               try {
                   Thread.sleep(1000); // 等待1秒后重试
                   mailSender.send(messagePreparator);
               } catch (Exception e) {
                   log.error("等待重连后发送消息{}失败", msg.getId(), e);
                   saveToRetryQueue(msg);
               }
           }
       } else {
           // 非连接异常，直接记录并放入重试队列
           log.error("发送消息{}失败，非连接异常", msg.getId(), exception);
           saveToRetryQueue(msg);
       }
   }
   

   其中isConnectionFailure方法用于判断异常类型，可通过捕获MessagingException的嵌套异常来识别连接类错误。

2. 消息可靠性保障：

   • 所有发送失败的消息必须持久化到数据库或消息队列，避免内存丢失；
   • 后续通过定时任务重试死信消息，重试次数限制在3次以内，超过则标记为失败并触发告警。

3. 避免手动管理Session/Transport：
   不要手动调用transport.connect()或transport.close()，JavaMailSenderImpl会在每次发送时自动检查连接状态，断开时会自动重连。手动干预反而会破坏其内置的连接管理逻辑。

内容的提问来源于stack exchange，提问作者Wrapper