当数据库不可用或下游服务关闭时，Spring集成暂停轮询

Question

我正在做一个项目，在该项目中轮询数据库中的事件，如果检测到有效事件，则下游调用服务激活器以获取有关事件的信息。执行一些处理，然后将结果写回数据库。

我想要实现的是，在数据库关闭或任何下游服务不可用的情况下，我的微服务将暂停轮询一段可配置的时间，之后它将重新启动。

到目前为止，我已经研究了 CircuitBreakerAdvice 和 RetryAdvice，但这些似乎适用于服务激活器而不是入站通道适配器。我也知道 Resilience4j 提供了一个全面的断路器机制，但我没有找到将它实施到我的项目中的方法。

我想出的解决方案是实现一个 ReceiveMessageAdvice，它将轮询设置为活动并将其传递给轮询器。错误通道将跟踪累积的错误数量，当达到配置的阈值时，它将 pollingActive 属性设置为 false。至于重新激活投票，我有点卡住了。我的猜测是安排一个任务在一段时间后将值更改回 true，但不确定在哪里或如何做。

投票渠道

@Bean
public IntegrationFlow readDBMessage() {
    return IntegrationFLows.fromSupplier(
            () -> dbService.readMessage(),
                  channelAdapter ->
                         channelAdapter.poller(
                                pollerSpec ->
                                      pollerSpec.fixedDelay(
                                                 \\polling period)
                                                .advice(messagePollingControlAdvice())
            .channel("apiCallChannel")
            .get();
}

MessagePollingControlAdvice

   public static class MessagePollingControlAdvice implements ReceiveMessageAdvice {
    private volatile boolean pollingActive = false;

    @Override
    public boolean beforeReceive(Object source) {
        return pollingActive;
    }

    @Override
    public Message<?> afterReceive(Message<?> result, Object source) {
        return result;
    }

    public boolean isPollingActive() {
        return pollingActive;
    }

    //call this method from whatever place in your code to activate/deactivate poller
    public void setPollingActive(boolean pollingActive) {
        this.pollingActive = pollingActive;
    }
   } 

取自How to stop OR change delay of Spring Integration Poller

关于我应该如何继续这样做的任何建议？ 我在文档中遗漏了什么吗？

更新 谢谢阿特姆！

我已经实施了 Artem 给出的建议。以下是供其他人遇到此问题时参考的代码。

MessagePollingControlAdvice

   public static class MessagePollingControlAdvice implements ReceiveMessageAdvice {
    private volatile boolean pollingActive = false;
    private volatile Long pollingDeactivatedTime = Instant.now().getEpochSecond();

    @Override
    public boolean beforeReceive(Object source) {
        
        // Get the desired time from configuration file
        if (!pollingActive && (Instant.now().getEpochSecond() - pollingDeactivatedTime) > 30) {
            pollingActive = true;

        }
        return pollingActive;
    }

    @Override
    public Message<?> afterReceive(Message<?> result, Object source) {
        return result;
    }

    public boolean isPollingActive() {
        return pollingActive;
    }

    //call this method from whatever place in your code to activate/deactivate poller
    public void setPollingActive(boolean pollingActive) {
        this.pollingDeactivatedTime = Instant.now().getEpochSecond();
        this.pollingActive = pollingActive;
    }
   } 

我已经查看了 SimpleActiveIdleReceiveMessageAdvice 并且肯定会在我的代码中实现一些逻辑。

作为后续问题：据我了解，即使在轮询期间发生错误，建议中的代码也会执行，因此是否可以跟踪此类中的错误并扩展逻辑以从其中停用轮询?

Answer 1

您可以在MessagePollingControlAdvice 中进行超时检查。不需要额外的计划任务，因为通道适配器上的这个轮询器仍在计划中。

因此，在 beforeReceive() 中，您不仅可以跟踪当前状态，还可以跟踪上次打开更改的时间。

SimpleActiveIdleReceiveMessageAdvice 也应该是您整个算法的一个很好的补充。