Spring Cloud Stream kafka 暂停/恢复活页夹

Question

我们使用 spring cloude stream 2.0 和 Kafka 作为消息代理。
对于目标系统（DB 或 3rd 方 API）不可用的情况，我们已经实现了一个停止应用程序上下文的断路器，如下所示：Stop Spring Cloud Stream @StreamListener from listening when target system is down

现在在 spring cloud stream 2.0 中有一种方法可以使用 actuator 来管理 binder 的生命周期：Binding visualization and control

是否可以从代码控制活页夹生命周期，这意味着如果目标服务器关闭，到pause 活页夹，当它启动时，到resume？

Answer 1

对不起，我看错了你的问题。

您可以自动连接 BindingsEndpoint，但不幸的是，它的 State 枚举是私有的，因此您不能以编程方式调用 changeState()。

我有opened an issue for this。

编辑

你可以用反射来做，但它有点难看......

@SpringBootApplication
@EnableBinding(Sink.class)
public class So53476384Application {

    public static void main(String[] args) {
        SpringApplication.run(So53476384Application.class, args);
    }

    @Autowired
    BindingsEndpoint binding;

    @Bean
    public ApplicationRunner runner() {
        return args -> {
            Class<?> clazz = ClassUtils.forName("org.springframework.cloud.stream.endpoint.BindingsEndpoint$State",
                    So53476384Application.class.getClassLoader());
            ReflectionUtils.doWithMethods(BindingsEndpoint.class, method -> {
                try {
                    method.invoke(this.binding, "input", clazz.getEnumConstants()[2]); // PAUSE
                }
                catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }, method -> method.getName().equals("changeState"));
        };
    }

    @StreamListener(Sink.INPUT)
    public void listen(String in) {

    }

}