如果您使用spring-kafka和kafka-streams，我建议您将kafkastreams定义为@bean，这样spring将负责对象的生命周期，比如

@Configuration
public class KafkaConfig{
  @Bean
  KafkaStreams ingestAndAggregateStream(KafkaProperties kafkaProps){
  Topology t....
  KafkaStreams stream = new KafkaStreams(t,kafkaProps.buildStreamsProperties())
  stream.start();
  return stream;
  }
}

一旦您这样做了，您就可以在控制器中自动连接流，并使用它来检索和查询本地kvstore

@RestController
public class StreamingController {

  private final KafkaStreams ingestAndAggregateStream;
  public StreamingController(KafkaStreams ingestAndAggregateStream){
    this.ingestAndAggregateStream = ingestAndAggregateStream;
  }

  @RequestMapping(value = "/queryStore", method = RequestMethod.GET)
  @ResponseBody
  public Value getKVStoreValue(@RequestParam String key){
    ReadOnlyKeyValueStore<String, Value> store = ingestAndAggregateStream
        .store("KV_STORE_NAME",
            QueryableStoreTypes.<String, Value>keyValueStore());
    return store.get(key);
  }
}