我正在遵循一个简单的指南文档,将Kafka流与弹簧引导一起使用(弹簧指南)
我很清楚如何进出消息,然后在中间我能够进行一些处理,替换@KafkaListener
和kafkaTemplate.send()
所以我做了一个超级简单的基本类,如下所示:
@EnableBinding(Processor.class)
public static class UppercaseTransformer {
@StreamListener
@Input(Processor.INPUT)
public void receive(String input) {
System.out.println(input);
}
}
然后(也许这是我的错误),我从控制器执行此操作:
template.send("my-topic","hello world");
我正在使用具有以下配置的春季云流:
spring:
cloud:
stream:
bindings:
input:
destination: my-topic
group: ${spring.application.name}
consumer:
concurrency: ${KAFKA_CONSUMER_CONCURRENCY:3}
output:
destination: my-topic
kafka:
binder:
brokers: localhost:9092
auto-create-topics: false
required-acks: all
transaction:
transaction-id-prefix: ${spring.application.name}-
producer:
configuration:
retries: 3
bindings:
input:
consumer:
configuration:
isolation.level: read_committed
enable-dlq: true
dlq-name: some-name
也与消费者和听众一起尝试过这个
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
每次我尝试发送消息时,我都会收到以下内容:
class java.lang.String cannot be cast to class [B (java.lang.String and [B are in module java.base of loader 'bootstrap')
不知道出了什么问题,为什么从普通听众到这个版本有这么大的变化......想法?
我刚刚从 start.spring.io 创建了一个应用程序,并选择了"Cloud Stream"和"Kafka"。生成项目并将其添加到主类中(使用上面提供的相同配置)。
@SpringBootApplication
@EnableBinding(Processor.class)
public class So54408906Application {
public static void main(String[] args) {
SpringApplication.run(So54408906Application.class, args);
}
@StreamListener(Processor.INPUT)
public void receive(String input) {
System.out.println(input);
}
}
然后运行 kafka 控制台制作器脚本。
kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic
脚本中提供的文本正在记录在应用程序的控制台上。