在 Kafka 中发送 ProducerRecord 所需的 Serilizer 类型

Question

我正在使用 spring kafka (KafkaTemplate) 发送字符串消息。但为了使其与旧代码兼容，我需要在消息中附加额外的 CorrelationId。所以我创建了 ProducerRecord 对象，以我的消息作为它的值，并在它的 HeaderRecord 中设置 CorrelationId。

        producerRecord = new ProducerRecord<>(
            kafkaTemplate_.getDefaultTopic(),
            null,
            null,
            null,
            myStringMessage,
            Collections.singletonList(new RecordHeader("CorrelationID", someIdAsBytes)));
    kafkaTemplate_.sendDefault(producerRecord);

键和值序列化器设置为 StringSerializer，但 about 代码无法说明 ProduderRecord 不是 String 或 StringSerializer 类型。

如果我像下面那样执行 toString()，它会起作用。但是在MessageListener 端，接收到的ConsumerRecord 没有correlationId 作为它的RecordHeader，因为correlationId 是作为ProducerRecord 的RecordHeader 附加的。所以我必须在 MessageListener onMessage(Object msg) 上进行类型转换，例如从 Object 转换为 ConsumerRecord（这是 ProduderRecord 的字符串序列化版本），从 ConsumerRecord.value() 字段解析 ProducerRecord，然后从 ProcuderRecord 标头获取 CorrelationId，并从生产者记录。这看起来很麻烦。我的发送和接收逻辑正常吗？

kafkaTemplate_.sendDefault(producerRecord.toString());

Answer 1

看起来您只是缺少以下区别：

/**
 * Send the data to the default topic with no key or partition.
 * @param data The data.
 * @return a Future for the {@link SendResult}.
 */
ListenableFuture<SendResult<K, V>> sendDefault(V data);

和

/**
 * Send the provided {@link ProducerRecord}.
 * @param record the record.
 * @return a Future for the {@link SendResult}.
 * @since 1.3
 */
ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);

看，你在ProducerRecord 中有一个topic 属性，所以，根本没有理由考虑sendDefault()。另外，正如您所见，这让您感到困惑。