ReplyingKafkaTemplate/KafkaTemplate不是发送/接收密钥 [英] ReplyingKafkaTemplate / KafkaTemplate not sending / receiving key

问题描述

我有一个模板，如下所示：

@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<?>。遗憾的是，您还必须为回复主题和相关性设置标头。

这篇关于ReplyingKafkaTemplate/KafkaTemplate不是发送/接收密钥的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！