有人知道Spring Boot与Kafka Connect的任何集成吗?有一个 spring-kafka 项目,我认为它与 Kafka 客户端很好地集成,但不能连接和流式传输 API。
> KConnect 本身是一个不同的客户端,你不能给它添加 spring。通过这样做,您最终将创建自己的客户端,该客户端不能称为KConnect。如果您打算将 KConnect 插件与 spring 集成,这是一项艰巨的任务,但应该仍然可以,但我不推荐它,因为插件在初始化和运行时方面应该很轻。此外,KConnect并不意味着拥有任何业务逻辑,如果没有业务逻辑,那么标准插件应该能够完全满足您的需求。
但是,KStreams可以与弹簧集成。这就像创建一个 bean 对象一样简单。这是一个示例
public class SampleStream {
@Autowired
CustomBean myBean;
private static final Logger LOG = LoggerFactory.getLogger(SampleStream.class);
public SampleStream() {
KafkaStreams stream = getStream();
LOG.info("Starting Stream {}", stream);
stream.start();
Runtime.getRuntime().addShutdownHook(new Thread(stream::close));
}
private KafkaStreams getStream(){
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> requestStream = builder.stream("REQUEST_TOPIC");
KStream<String, String> responseStream = requestStream.flatMap((key, request) -> {
myBean.process(request)
//custom logic
});
responseStream.to("RESPONSE_TOPIC");
return new KafkaStreams(builder.build(), getStreamProperties());
}
private Properties getStreamProperties() {
String defaultKeySerdes = Serdes.StringSerde.class.getName();
String defaultValueSerdes = Serdes.StringSerde.class.getName();
String defaultExceptionHandler = LogAndContinueExceptionHandler.class.getName();
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getClass().toString());
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka.com:9092");
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, defaultKeySerdes);
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, defaultValueSerdes);
properties.put(GenericMessagingConstants.KAFKA_DESERIALIZER_VALUE_CLASS_CONFIG, String.class.getName());
properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, defaultExceptionHandler);
....
return properties;
}
}
在应用程序上下文中创建 Bean,如下所示,
<bean id="sampleStream" class="SampleStream"/>
回答这个问题,因为需要一些时间来弄清楚这一点。