问Kafka与Spring框架的连接

EN

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 id="sampleStream" class="SampleStream"/>