java - ReplyingKafkaTemplate/KafkaTemplate 未发送/接收 key

Question

我有一个如下所示的模板:

@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=MessageBDto 并在监听器中接收 key 吗？

监听器似乎没有获取 key 和/或它似乎是 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 :

@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;
}

当我查看监听器的调试器时，它显示该键是 MessageBDto 的空实例？？？

版本:

• Apache Zookeeper 3.5.7
• Apache 卡夫卡 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>
  

Answer 1

我不确定你的代码发生了什么。这是一个工作示例...

@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 INFO 16591 --- [o60384112-0-C-1] com.example.demo.So60384112Application : Foo [field=foo]:Bar [field=bar]

2020-02-24 18:00:47.915 INFO 16591 --- [ main] com.example.demo.So60384112Application : Foo [field=foo]:Bar [field=BAR]

要返回 key ，您必须返回 Message<?> 。不幸的是，您还必须为回复主题和相关性设置 header 。