我正在尝试通过spring-cloud-stream-binder-kafka-streams库通过一个简单的Spring引导应用程序使用Kafka中某个主题的消息。

我在这里关注文档-https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/blob/master/docs/src/main/asciidoc/kafka-streams.adoc

我的代码如下：     软件包co.zip.poc.springbootkafkakotlin

import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.apache.kafka.streams.kstream.KStream
import org.springframework.context.annotation.Bean
import java.util.function.Consumer


@SpringBootApplication
class SpringBootKafkaConsumer {

  @Bean
  fun process(): Consumer<KStream<Any,Any>> {
    println("=======================> Creating a consumer for reading stuff<=====================")
    return Consumer { input -> input.foreach { key,value ->
      println("============key = $key")
      println("===========value = $value")
    }}
  }
}

fun main(args: Array<String>) {
  runApplication<SpringBootKafkaConsumer>(*args)
}

application.yml

spring:
  application:
    name: customer-balance
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              application:
                id: customer-balance
      bindings:
        process_in:
          destination: "dbserver1.inventory.customers"

logging:
  level:
    org.springframework.kafka.config: trace

我的本​​地kafka运行在端口9092上。我能够通过kafka控制台使用者使用消息-

kafka-console-consumer.sh \
    --bootstrap-server localhost:9092 \
    --from-beginning \
    --property print.key=true \
    --topic dbserver1.inventory.customers

上面在控制台上产生4条消息。但是，当我运行spring boot应用程序时，没有显示key = ..和value = print消息的任何内容。

我在做什么错了？