使用IntegrationFlows构建的RabbitMQ监听器运行时启停方案问询

Spring Integration 中所有通过 IntegrationFlows 构建的流及内部端点组件都实现了 SmartLifecycle 接口，本身就支持运行时启停控制，不需要依赖 @RabbitListener 专属的 RabbitListenerEndpointRegistry，可通过以下两种方式实现操控：

方式1：直接操控IntegrationFlow实例 #

你定义的 IntegrationFlow 本身就是Spring容器中的生命周期Bean，默认Bean名称为定义Flow的方法名，直接注入后即可调用启停方法：

流定义示例 #

@Bean
IntegrationFlow registerDevices() {
   return IntegrationFlows
      .from(adapter)
      .channel(channel)
      .get();
}

启停代码示例 #

@RestController
public class FlowController {
    // 按Bean名称注入目标Flow
    @Autowired
    @Qualifier("registerDevices")
    private SmartLifecycle registerDevicesFlow;

    // 停止流
    @PostMapping("/flow/register/stop")
    public void stopRegisterFlow() {
        if (registerDevicesFlow.isRunning()) {
            registerDevicesFlow.stop();
        }
    }

    // 启动流
    @PostMapping("/flow/register/start")
    public void startRegisterFlow() {
        if (!registerDevicesFlow.isRunning()) {
            registerDevicesFlow.start();
        }
    }
}

方式2：精准操控流内的监听器组件 #

如果不需要停止整个流，只想单独控制Rabbit监听器的启停，可以给入站适配器指定自定义ID，通过ID从Spring上下文获取实例后操控：

流定义示例（给适配器加ID） #

@Bean
IntegrationFlow registerDevices() {
   return IntegrationFlows
      // 给Rabbit入站适配器指定ID
      .from(adapter, e -> e.id("registerRabbitAdapter"))
      .channel(channel)
      .get();
}

启停代码示例 #

@Component
public class RabbitAdapterOperation {
    @Autowired
    private ApplicationContext context;

    public void stopRabbitListener() {
        SmartLifecycle adapter = context.getBean("registerRabbitAdapter", SmartLifecycle.class);
        if (adapter.isRunning()) {
            adapter.stop();
        }
    }

    public void startRabbitListener() {
        SmartLifecycle adapter = context.getBean("registerRabbitAdapter", SmartLifecycle.class);
        if (!adapter.isRunning()) {
            adapter.start();
        }
    }
}

注意事项 #

• 调用stop()方法为优雅停机，会等待正在处理的消息执行完成后再停止监听器，不会丢失运行中的消息
• 如果配置了autoStartup = false，应用启动时对应组件/流不会自动启动，需要手动调用start()方法触发启动

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