ReplyingKafkaTemplate/KafkaTemplate不是发送/接收密钥
我有一个模板，如下所示：
@Autowired
private ReplyingKafkaTemplate<ItemId, MessageBDto, MessageBDto> xxx2ReplyingKafkaTemplate;
我的发送包装方法如下所示：
public RequestReplyFuture<ItemId, MessageBDto, MessageBDto> sendAndReceiveMessageB(MessageBDto message) {
ProducerRecord<ItemId, MessageBDto> producerRecord = new ProducerRecord<>(KafkaTopicConfig.xxx2_TOPIC, new ItemId(message.getCount()), message);
producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, KafkaTopicConfig.xxx2_REPLY_TOPIC.getBytes()));
return this.xxx2ReplyingKafkaTemplate.sendAndReceive(producerRecord);
}
这是我的听众：
@SendTo
@KafkaListener(topics=KafkaTopicConfig.xxx2_TOPIC, containerFactory="xxx2ListenerContainerFactory")
public MessageBDto xxx2Listener(ConsumerRecord<ItemId, MessageBDto> message) {
System.out.println("xxx2(value): " + message.value().getMessage() + ", " + message.value().getCount());
message.value().setCount(message.value().getCount() * 2);
return message.value();
}
这不是应该将key=ItemID，Value=MessageBD发送到侦听器并在侦听器中接收密钥吗？
侦听器似乎未获得密钥和/或它似乎是MessageBDto的另一个实例。
我是否误解了这应该如何工作？
编辑：
生产者Bean：
@Bean
public ProducerFactory<ItemId, MessageBDto> xxx2ProducerFactory() {
return new DefaultKafkaProducerFactory<ItemId, MessageBDto>(super.producerConfigs(),
new JsonSerializer<ItemId>(),
new JsonSerializer<MessageBDto>());
}
@Bean
public ConsumerFactory<ItemId, MessageBDto> xxx2ConsumerFactory() {
return new DefaultKafkaConsumerFactory<>(super.consumerConfigs(),
trustingDeserializer(ItemId.class),
trustingDeserializer(MessageBDto.class));
}
@Bean
public KafkaMessageListenerContainer<ItemId, MessageBDto> dtms2MessageListenerContainer() {
return new KafkaMessageListenerContainer<>(xxx2ConsumerFactory(),
new ContainerProperties(KafkaTopicConfig.xxx2_REPLY_TOPIC));
}
@Bean
public ReplyingKafkaTemplate<ItemId, MessageBDto, MessageBDto> xxx2ReplyingKafkaTemplate() {
return new ReplyingKafkaTemplate<>(xxx2ProducerFactory(), xxx2MessageListenerContainer());
}
private <T> JsonDeserializer<T> trustingDeserializer(Class<T> targetType) {
JsonDeserializer<T> deserializer = new JsonDeserializer<>(targetType);
deserializer.addTrustedPackages("*");
return deserializer;
}
消费类豆类：
@Bean
public KafkaTemplate<ItemId, MessageBDto> xxx2KafkaTemplate() {
return new KafkaTemplate<>(xxx2ProducerFactory());
}
@Bean
public KafkaListenerContainerFactory<ItemId, MessageBDto> xxxListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<ItemId, MessageBDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(dtms2ConsumerFactory());
factory.setReplyTemplate(xxx2KafkaTemplate());
return factory;
}
当我查看监听程序的调试器时，它显示key是MessageBDto？
的空实例版本：
- ApacheZooKeeper 3.5.7
- 阿帕奇·卡夫卡2.12-2.4.0
Spring Boot 2.2.4
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.4.2.RELEASE</version> <!-- $NO-MVN-MAN-VER$ --> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-streams</artifactId> <version>2.4.0</version> <!--$NO-MVN-MAN-VER$--> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.0</version> <!--$NO-MVN-MAN-VER$--> </dependency>
解决方案
我不确定您的代码出了什么问题。以下是一个工作示例...
@SpringBootApplication
public class So60384112Application {
private static final Logger LOG = LoggerFactory.getLogger(So60384112Application.class);
public static void main(String[] args) {
SpringApplication.run(So60384112Application.class, args).close();
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("so60384112").partitions(1).replicas(1).build();
}
@Bean
public NewTopic replies() {
return TopicBuilder.name("so60384112replies").partitions(1).replicas(1).build();
}
@KafkaListener(id = "so60384112", topics = "so60384112")
@SendTo
public Message<?> listen(ConsumerRecord<Foo, Bar> record) {
LOG.info(record.key().toString() + ":" + record.value().toString());
return MessageBuilder.withPayload(new Bar(record.value().getField().toUpperCase()))
.setHeader(KafkaHeaders.MESSAGE_KEY, record.key())
.setHeader(KafkaHeaders.CORRELATION_ID, record.headers().lastHeader(KafkaHeaders.CORRELATION_ID).value())
.setHeader(KafkaHeaders.TOPIC, record.headers().lastHeader(KafkaHeaders.REPLY_TOPIC).value())
.build();
}
@Bean
public ReplyingKafkaTemplate<Foo, Bar, Bar> replyer(ProducerFactory<Foo, Bar> pf,
ConcurrentKafkaListenerContainerFactory<Foo, Bar> containerFactory) {
containerFactory.setReplyTemplate(kafkaTemplate(pf));
ConcurrentMessageListenerContainer<Foo, Bar> container = containerFactory.createContainer("so60384112replies");
container.getContainerProperties().setGroupId("so60384112replies");
ReplyingKafkaTemplate<Foo, Bar, Bar> replyer = new ReplyingKafkaTemplate<>(pf, container);
return replyer;
}
@Bean
public KafkaTemplate<Foo, Bar> kafkaTemplate(ProducerFactory<Foo, Bar> pf) {
return new KafkaTemplate<>(pf);
}
@Bean
public ApplicationRunner runner(ReplyingKafkaTemplate<Foo, Bar, Bar> template) {
return args -> {
RequestReplyFuture<Foo, Bar, Bar> future =
template.sendAndReceive(new ProducerRecord<>("so60384112", 0, new Foo("foo"), new Bar("bar")));
ConsumerRecord<Foo, Bar> record = future.get(10, TimeUnit.SECONDS);
LOG.info(record.key().toString() + ":" + record.value().toString());
};
}
}
class Foo {
private String field;
public Foo() {
}
public Foo(String field) {
this.field = field;
}
public String getField() {
return this.field;
}
public void setField(String field) {
this.field = field;
}
@Override
public String toString() {
return getClass().getSimpleName() + " [field=" + this.field + "]";
}
}
class Bar {
private String field;
public Bar() {
}
public Bar(String field) {
this.field = field;
}
public String getField() {
return this.field;
}
public void setField(String field) {
this.field = field;
}
@Override
public String toString() {
return getClass().getSimpleName() + " [field=" + this.field + "]";
}
}
spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.properties.spring.json.trusted.packages=*
spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
2020-02-24 18：00：47.904信息2020-[o60384112-0-C-1]com.example.demo.So60384112应用程序：FOO[FIELD=FOO]：BAR[FIELD=BAR]
2020-02-24 18：00：47.915信息16591-[Main]com.example.demo.So60384112应用程序：foo[field=foo]：bar[field=bar]
要返回密钥，您必须返回Message<?>。遗憾的是，您还必须为回复主题和相关性设置标头。
相关文章