spring-kafka请求回复:不同类型的请求和回复

Question

ReplyingKafkaTemplate提供请求-回复支持的文档(在Spring Kafka 2.1.3中介绍(表明，请求和回复可以使用不同类型：

ReplyingKafkaTemplate<K, V, R>

其中参数化类型K表示消息Key，V表示Value(即请求(，R表示Reply。

到目前为止还不错。但是，用于实现服务器端Request Reply的相应支持类似乎不支持V、R的不同类型。文档建议使用添加了@SendTo注释的KafkaListener，该注释在后台使用MessageListenerContainer上配置的replyTemplate。但是AbstractKafkaListenerEndpoint只支持侦听器和replyTemplate:的单一类型

public abstract class AbstractKafkaListenerEndpoint<K, V>
implements KafkaListenerEndpoint, BeanFactoryAware, InitializingBean {
...
/**
* Set the {@link KafkaTemplate} to use to send replies.
* @param replyTemplate the template.
* @since 2.0
*/
public void setReplyTemplate(KafkaTemplate<K, V> replyTemplate) {
this.replyTemplate = replyTemplate;
}
...
}

因此V和R需要是相同的类型。

文档中使用的示例确实将String用于Request和Reply。

我是遗漏了什么，还是这是Spring Kafka请求回复支持中的一个设计缺陷，应该报告并更正？

Answer 1

这在2.2版本中得到了修复。

对于早期版本，只需注入一个原始KafkaTemplate(没有泛型(。

编辑

@SpringBootApplication
public class So53151961Application {
public static void main(String[] args) {
SpringApplication.run(So53151961Application.class, args);
}
@KafkaListener(id = "so53151961", topics = "so53151961")
@SendTo
public Bar handle(Foo foo) {
System.out.println(foo);
return new Bar(foo.getValue().toUpperCase());
}
@Bean
public ReplyingKafkaTemplate<String, Foo, Bar> replyingTemplate(ProducerFactory<String, Foo> pf,
ConcurrentKafkaListenerContainerFactory<String, Bar> factory) {
ConcurrentMessageListenerContainer<String, Bar> replyContainer =
factory.createContainer("so53151961-replyTopic");
replyContainer.getContainerProperties().setGroupId("so53151961.reply");
ReplyingKafkaTemplate<String, Foo, Bar> replyingKafkaTemplate = new ReplyingKafkaTemplate<>(pf, replyContainer);
return replyingKafkaTemplate;
}
@Bean
public KafkaTemplate<String, Bar> replyTemplate(ProducerFactory<String, Bar> pf,
ConcurrentKafkaListenerContainerFactory<String, Bar> factory) {
KafkaTemplate<String, Bar> kafkaTemplate = new KafkaTemplate<>(pf);
factory.setReplyTemplate(kafkaTemplate);
return kafkaTemplate;
}
@Bean
public ApplicationRunner runner(ReplyingKafkaTemplate<String, Foo, Bar> template) {
return args -> {
ProducerRecord<String, Foo> record = new ProducerRecord<>("so53151961", null, "key", new Foo("foo"));
RequestReplyFuture<String, Foo, Bar> future = template.sendAndReceive(record);
System.out.println(future.get(10, TimeUnit.SECONDS).value());
};
}
@Bean
public NewTopic topic() {
return new NewTopic("so53151961", 1, (short) 1);
}
@Bean
public NewTopic reply() {
return new NewTopic("so53151961-replyTopic", 1, (short) 1);
}
public static class Foo {
public String value;
public Foo() {
super();
}
public Foo(String value) {
this.value = value;
}
public String getValue() {
return this.value;
}
public void setValue(String value) {
this.value = value;
}
@Override
public String toString() {
return "Foo [value=" + this.value + "]";
}
}
public static class Bar {
public String value;
public Bar() {
super();
}
public Bar(String value) {
this.value = value;
}
public String getValue() {
return this.value;
}
public void setValue(String value) {
this.value = value;
}
@Override
public String toString() {
return "Bar [value=" + this.value + "]";
}
}
}

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.properties.spring.json.trusted.packages=com.example

结果

Foo [value=foo]
Bar [value=FOO]