Spring Integration多实例邮件消费者防重复处理策略

<int-mail:inbound-channel-adapter id="imapAdapter"
                                      store-uri="imaps://abc.com/INBOX"
                                      channel="receiveChannel"
                                      should-delete-messages="false"
                                      should-mark-messages-as-read="true"
                                      java-mail-properties="javaMailProperties"
                                      auto-startup="true">
        <int:poller max-messages-per-poll="1" fixed-rate="600000" />
    </int-mail:inbound-channel-adapter>

    <util:properties id="javaMailProperties">
        <prop key="mail.imap.socketFactory.class">javax.net.ssl.SSLSocketFactory</prop>
        <prop key="mail.imap.socketFactory.fallback">false</prop>
        <prop key="mail.store.protocol">imaps</prop>
        <prop key="mail.debug">false</prop>
        <prop key="mail.smtp.ssl.protocols">TLSv1.2</prop>
    </util:properties>

    <bean id="mailService" class="com.xpressbees.poller.EmailPoller"/>

    <int:service-activator id="serviceActivator" input-channel="receiveChannel" ref="mailService" method="handleMail"/>

@Configuration
@EnableIntegration
public class MailPollingConfig {

    // ... 其他邮件配置 bean ...

    @Bean
    public IntegrationFlow imapMailFlow(
            @Value("${mail.imap.uri}") String imapUri,
            @Qualifier("javaMailProperties") Properties javaMailProperties,
            EmailPoller mailService) {
        return IntegrationFlows.from(
                Mail.imapInboundAdapter(imapUri)
                        .javaMailProperties(javaMailProperties)
                        .shouldDeleteMessages(false)
                        .shouldMarkMessagesAsRead(true) // 即使有领导者选举，保留此设置也是一个好的实践
                        .autoStartup(false), // 初始不自动启动，由领导者选举控制
                e -> e.poller(Pollers.fixedRate(600000).maxMessagesPerPoll(1))
        )
        .handle(mailService, "handleMail")
        .get();
    }

    // 假设你已经配置了Spring Cloud Commons的Leader Election，例如通过Kubernetes或Zookeeper
    // 邮件适配器将通过LifecycleProcessor或自定义组件在成为领导者时启动
    // 详细实现请参考Spring Integration和Spring Cloud Commons的领导者选举文档
}

@Configuration
@EnableIntegration
public class IdempotentReceiverConfig {

    // ... 其他配置 ...

    // 定义一个消息存储，用于记录已处理的消息ID
    // 实际应用中，这通常是持久化的，例如Redis或JDBC
    @Bean
    public ConcurrentHashMapMessageStore messageStore() {
        return new ConcurrentHashMapMessageStore(); // 仅用于示例，生产环境请使用持久化存储
    }

    // 定义幂等接收器建议
    @Bean
    public IdempotentReceiverInterceptor idempotentReceiverInterceptor(MessageStore messageStore) {
        // 使用消息的 'Message-ID' 头部作为唯一标识符
        return new IdempotentReceiverInterceptor(new MessageIdExpression(), messageStore);
    }

    // 将幂等接收器建议应用到服务激活器
    @Bean
    public IntegrationFlow mailProcessingFlow(EmailPoller mailService,
                                              IdempotentReceiverInterceptor idempotentReceiverInterceptor) {
        return IntegrationFlows.from("receiveChannel") // 假设这是邮件适配器输出的通道
                .channel(c -> c.queue(10)) // 可以添加一个队列通道
                .handle(mailService, "handleMail", e -> e.advice(idempotentReceiverInterceptor)) // 应用幂等接收器建议
                .get();
    }

    // 假设EmailPoller的handleMail方法处理邮件，并可以从MessageHeader获取Message-ID
    // 例如：String messageId = (String) message.getHeaders().get("mail_message_id");
}

public class MessageIdExpression implements MessageProcessor<String> {
    @Override
    public String processMessage(Message<?> message) {
        // 邮件的Message-ID通常在'mail_message_id'头部
        String messageId = (String) message.getHeaders().get("mail_message_id");
        if (messageId == null) {
            // 如果没有Message-ID，可以使用其他唯一标识，或抛出异常
            throw new IllegalArgumentException("Mail message does not contain 'mail_message_id' header.");
        }
        return messageId;
    }
}