Spring Integration JPA入站适配器重复发邮件问题排查求助

核心问题排查方向 #

1. 事务未覆盖完整处理链路 #

如果Spring Integration的JPA入站适配器仅在查询获取实体阶段开启事务，后续的邮件发送、实体状态更新操作处于事务之外，悲观锁会在查询事务提交后立即释放，其他实例的轮询任务就能再次抓取到同一实体，导致重复发送。

2. 悲观锁应用时机错误 #

若仅在实体加载后手动加锁，而非在查询语句层面直接加锁，会出现“间隙时间窗”：多个实例同时查询到同一PENDING状态的实体，之后才尝试加锁，此时已有实例完成加锁并处理，后续实例可能仍能拿到实体（取决于数据库锁机制）。

3. 锁配置未实际生效 #

JPA入站适配器的lockMode配置可能未正确传递到查询语句，导致悲观锁并未实际生效；或者事务隔离级别过低，无法阻止脏读或不可重复读。

关键配置验证与修正 #

确保事务覆盖全流程 #

需要让实体查询→邮件发送→状态更新的全链路处于同一个事务中。可通过给入站适配器绑定TransactionInterceptor，将整个消息处理逻辑纳入事务管理。

正确配置查询阶段的悲观锁 #

JPA入站适配器需在查询阶段就指定lockMode为PESSIMISTIC_WRITE，确保数据库在查询时就对匹配行加排他锁，阻止其他实例读取。

正确配置示例 #

Java配置方式 #

@Configuration
@EnableIntegration
public class JpaEmailPollerConfig {

    @Autowired
    private EntityManagerFactory entityManagerFactory;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Bean
    public JpaInboundChannelAdapter jpaEmailPoller() {
        JpaInboundChannelAdapter adapter = new JpaInboundChannelAdapter(jpaExecutor());
        adapter.setChannel(emailProcessingChannel());
        adapter.setAdviceChain(Collections.singletonList(transactionAdvice()));
        return adapter;
    }

    @Bean
    public JpaExecutor jpaExecutor() {
        JpaExecutor executor = new JpaExecutor(entityManagerFactory);
        executor.setJpqlQuery("SELECT e FROM EmailNotification e WHERE e.status = 'PENDING'");
        // 查询阶段直接加悲观写锁
        executor.setLockMode(LockModeType.PESSIMISTIC_WRITE);
        // 每次轮询只取1条，缩小锁范围
        executor.setMaxResults(1);
        return executor;
    }

    @Bean
    public MessageChannel emailProcessingChannel() {
        return new DirectChannel();
    }

    @Bean
    public TransactionInterceptor transactionAdvice() {
        DefaultTransactionAttribute txAttr = new DefaultTransactionAttribute();
        txAttr.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        txAttr.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        return new TransactionInterceptor(transactionManager, txAttr);
    }

    @Bean
    @ServiceActivator(inputChannel = "emailProcessingChannel")
    public MessageHandler emailSenderHandler() {
        return message -> {
            EmailNotification email = (EmailNotification) message.getPayload();
            // 执行邮件发送逻辑
            sendEmail(email);
            // 更新状态为已发送，事务内自动提交
            email.setStatus("SENT");
        };
    }

    private void sendEmail(EmailNotification email) {
        // 邮件发送实现
    }
}

XML配置方式 #

<int-jpa:inbound-channel-adapter id="emailPoller"
                                  channel="emailProcessingChannel"
                                  jpa-executor="emailJpaExecutor"
                                  auto-startup="true">
    <int:poller fixed-delay="5000">
        <int:advice-chain>
            <ref bean="transactionAdvice"/>
        </int:advice-chain>
    </int:poller>
</int-jpa:inbound-channel-adapter>

<bean id="emailJpaExecutor" class="org.springframework.integration.jpa.core.JpaExecutor">
    <property name="entityManagerFactory" ref="entityManagerFactory"/>
    <property name="jpqlQuery" value="SELECT e FROM EmailNotification e WHERE e.status = 'PENDING'"/>
    <property name="lockMode" value="PESSIMISTIC_WRITE"/>
    <property name="maxResults" value="1"/>
</bean>

<bean id="transactionAdvice" class="org.springframework.transaction.interceptor.TransactionInterceptor">
    <property name="transactionManager" ref="transactionManager"/>
    <property name="transactionAttributes">
        <props>
            <prop key="*">PROPAGATION_REQUIRED,ISOLATION_READ_COMMITTED</prop>
        </props>
    </property>
</bean>

<int:channel id="emailProcessingChannel"/>

<int:service-activator input-channel="emailProcessingChannel" ref="emailSenderService"/>

<bean id="emailSenderService" class="com.example.EmailSenderService"/>

额外优化建议 #

• 每次轮询仅抓取1条实体（maxResults=1），缩小锁范围，减少锁竞争。
• 邮件发送完成后立即更新实体状态为SENT或PROCESSING，确保后续轮询不会再抓取到该实体。
• 给邮件实体添加唯一业务标识（如messageId），在发送前做幂等校验，即使重复抓取也不会重复发送。

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