如何在运行时调整Spring Integration Http出站网关的轮询频率及启停集成流

1. 现有配置必要性说明 #

你当前拆分source流+httpOutRequest通道+outbound流的配置是合法的，但不是必须的。可以直接将轮询器绑定到HTTP出站网关的上游，省略中间通道和拆分的流，简化配置示例如下：

@Bean
public IntegrationFlow combinedFlow() {
    return IntegrationFlows.from(
                    () -> new GenericMessage<String>(""),
                    e -> e.poller(p -> p.fixedRate(5000).id("httpPoller"))
            )
            .handle(Http.outboundGateway("http://localhost:8080/harry-potter-service/next/request")
                    .httpMethod(HttpMethod.GET)
                    .expectedResponseType(TransferRequest.class)
            )
            .handle(handler())
            .get();
}

2. 运行时调整轮询频率+停止集成流实现 #

你原有代码直接修改period字段无效，因为Spring初始化时已经将初始值写入触发器实例，修改字段不会同步到实际运行的轮询规则中，可通过以下方式实现：

2.1 前置配置修改 #

首先给需要控制的端点、轮询器加ID，同时声明控制总线（Control Channel）用于发送控制命令：

// 声明控制总线
@Bean
public IntegrationFlow controlBus() {
    return IntegrationFlows.from("controlChannel")
            .controlBus()
            .get();
}

// 给source流的端点和轮询器加ID
@Bean
public IntegrationFlow source() {
    return IntegrationFlows.from(
                    () -> new GenericMessage<String>(""),
                    e -> e.id("sourceEndpoint")
                            .poller(p -> p.id("httpPoller")
                                    .fixedRate(period))
            )
            .channel("httpOutRequest")
            .get();
}

2.2 异常处理逻辑改造 #

在errorHandler中注入controlChannel，通过发送控制命令实现规则调整：

@Autowired
private MessageChannel controlChannel;

@Bean
@ServiceActivator(inputChannel = "errorChannel")
public MessageHandler errorHandler() {
    return new MessageHandler() {
        @Override
        public void handleMessage(Message<?> message) throws MessagingException {
            LOG.error("message.payload: " + message.getPayload());
            MessageHandlingException e = (MessageHandlingException) message.getPayload();
            LOG.error("Exception: " + e);
            LOG.debug("exception counter = " + (++problemCounter));

            if (problemCounter >= MAX_COUNT_TO_REDUCE_POLLING) {
                LOG.debug("调整轮询频率/停止集成流");
                // 方式1：下调轮询间隔为60s
                controlChannel.send(new GenericMessage<>("@httpPoller.getTrigger().setPeriod(60000)"));
                // 方式2：直接停止source集成流（轮询逻辑在source流中，停止outbound流无效）
                // controlChannel.send(new GenericMessage<>("@source.stop()"));
            }
        }
    };
}

3. 补充问题解答 #

你提到的Lambda参数p是PollerSpec实例，配置时给poller指定ID后，即可在Spring容器中获取对应的PollerMetadata实例，其内部的trigger属性就是你配置的PeriodicTrigger。
通过Control Channel修改fixedRate的逻辑就是上述2.2中的控制命令，调用trigger的setPeriod方法即可更新轮询间隔，无需额外手动获取参数实例。
如果不使用控制总线，也可以直接注入PollerMetadata和source流实例，直接调用对应方法：

@Autowired
@Qualifier("httpPoller")
private PollerMetadata httpPoller;

@Autowired
private IntegrationFlow source;

// 调整逻辑
// 改间隔
((PeriodicTrigger) httpPoller.getTrigger()).setPeriod(60000);
// 停流
source.stop();

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