在 Spring Cloud Kafka 中转换消息的最简单方法是什么？答案

【问题标题】：What is the simplest way to transform messages in Spring Cloud Kafka?在 Spring Cloud Kafka 中转换消息的最简单方法是什么？
【发布时间】：2020-10-01 15:19:46
【问题描述】：

我有以下代码

@SpringBootApplication
public class EnrichmentProcessPOC {

    public static void main(String[] args) {
        SpringApplication.run(EnrichmentProcessPOC.class, args);
    }

    public static class EnrichmentProcessorApplication {

        public static final String INPUT_TOPIC = "PAYMENT_MSG";
        public static final String OUTPUT_TOPIC = "PAYMENT_MSG_CIF";

        @Bean
        public Function<KStream<Bytes, String>, KStream<Bytes, String>> process() {

            return input -> input.mapValues(value -> "foo");
        }
    }
}

我的期望是，这应该用“foo”替换每个使用的消息，但它所做的只是在生产者主题上复制消息。我在这里缺少什么，为什么要这样做？

我也尝试过使用 Transformer，但它的行为相同。将消息实际转换为“foo”的最简单方法是什么？

【问题讨论】：

标签： spring spring-boot apache-kafka-streams spring-kafka spring-cloud-stream

【解决方案1】：

好吧，我是个白痴，原始代码运行良好，但是当我启动 Kafka 控制台消费者和生产者时，我在同一个主题上启动它们（复制/粘贴错误）。

现在我将它们用于不同的主题，我的代码按预期运行。

感谢大家的意见。

【讨论】：

【解决方案2】：

你能展示你的配置吗？这对我来说按预期工作：

@SpringBootApplication
public class So64158395Application {

    public static void main(String[] args) {
        SpringApplication.run(So64158395Application.class, args);
    }

    @Bean
    public Function<KStream<byte[], byte[]>, KStream<byte[], byte[]>> process() {
        return input -> input.map((key, value) -> KeyValue.pair(key, "foo".getBytes()));
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            template.send("in", "bar", "baz");
        };
    }

    @KafkaListener(id = "outListener", topics = "out")
    void listen(String value, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) {
        System.out.println(key + ":" + value);
    }

}

spring.cloud.stream.bindings.process-in-0.destination=in
spring.cloud.stream.bindings.process-in-0.group=group
spring.cloud.stream.bindings.process-out-0.destination=out

结果：

bar:foo

【讨论】：

【解决方案3】：

您是否尝试过使用函数“mapValues”而不是地图？

根据文档，这是两者之间的区别-

地图

将输入流的每条记录转化为输出流中的一条新记录>（key和value类型都可以任意改变）。

地图值

将每个输入记录的值转换为输出记录的新值（可能具有新的 >type）。

【讨论】：