将自定义 kafka 消费者与 Spring 云配置客户端集成

Question

我们正在使用 spring 配置客户端动态刷新属性。我们在类路径上添加了 spring-cloud-starter-bus-kafka，一切正常。所有这些依赖项的 Pom 版本是 2.X。

我想要做的是，删除 spring-cloud-starter-bus-kafka 并添加我的自定义代码以获取消费者事件和刷新上下文，然后使用云配置客户端刷新属性。我相信 Spring 某处正在呼叫 ConfigServicePropertySourceLocator.locate。基本上，我只想复制 Spring 在 spring-cloud-starter-bus-kafka 中所做的事情，以便实时刷新属性。

我做这一切的原因是，我在内部使用旧版本的 kafka-clients。我们有一个自己开发的版本，它支持加密等等。问题来了，因为 spring-cloud-starter-bus-kafka 正在使用 2.X 版本的 kakfa-clients 而我们的家庭版本还没有准备好。因此，它们中的任何一个都在某个时间点工作。

有人可以告诉我一些关于使用来自 kafka 的刷新事件并刷新属性需要做什么的指示吗？我不认为这太复杂了。它应该消耗 kafka 事件并在某处调用 ConfigSourceLocator 来刷新属性。

Answer 1

比那更简单。如果您查看 RefreshListener class，您会发现它所做的只是

    Set<String> keys = this.contextRefresher.refresh();
    log.info("Received remote refresh request. Keys refreshed " + keys);

其中 contextRefresher 是 org.springframework.cloud.context.refresh.ContextRefresher。

这将触发自动处理查找配置的代码。

将自定义 kafka 消费者与 Spring 云配置客户端集成

Integrating custom kafka consumer with Spring cloud config client

spring-cloud

spring-cloud-stream

spring-cloud-config

spring-kafka