当生产者以列表< object>的形式发送消息时,kafka消费者会收到不同格式的消息

p1iqtdky  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(345)
I am sending `List<Object>` from Kafka Producer to a topic. A consumer which listens to the topic receives the message but in a different format(i.e. `List<LinkedHashMap>` instead of `List<Object>`)

Any idea on how to receive the List<Object> messages in the consumer?I am sending `List<foo>`. But in the Consumer, data comes as `List<LinkedHashMap>`

只是把下面的例子改成列表https://memorynotfound.com/spring-kafka-json-serializer-deserializer-example/ 从链接中可以找到发送者和侦听者。唯一的变化是发送方将接受object而不是foo。

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

            @Autowired
            private FooSender sender;

            @Override
            public void run(String... strings) throws Exception {
            List<Foo> f = new ArrayList<>();

        f.add(new Foo("Spring Kafka", "sending and receiving JSON messages"))
        f.add(new Foo("Spring Kafka1", "sending and receiving11"))
                sender.send(f);
            }
        }
46qrfjad

46qrfjad1#

不要在注解中添加代码;很难阅读-改为编辑问题。
你用的是什么版本的SpringKafka?
您至少需要2.1.x和一个kafka代理>=0.11,这样类型信息就可以在头中传递。看这里。
对于早期版本,必须使用目标类型配置反序列化程序。
编辑
我看到了问题;由于类型擦除,序列化程序不知道容器(列表)的内容类型,因此它不会尝试内省集合以查找类型。
虽然我们可能可以解决像集合这样的简单情况,但更复杂的对象(例如Map的Map)将是一个挑战。
一个数组,而不是一个列表,可能会起作用。
同时,您可以使用消息转换器,在这里我们使用侦听器的参数类型来确定预期的类型。举个例子:

@SpringBootApplication
public class So53665459Application2 {

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

    @Bean
    public NewTopic topic() {
        return new NewTopic("so53665459", 1, (short) 1);
    }

    @Bean
    public MessageConverter converter() {
        return new StringJsonMessageConverter();
    }

    @KafkaListener(id = "so53665459", topics = "so53665459")
    public void listen(List<Foo1> foos) {
        System.out.println(foos);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, List<Object>> template) {
        return args -> template.send("so53665459", Collections.singletonList(new Foo1("baz1")));
    }

    public static class Foo1 {

        private String bar;

        public Foo1() {
            super();
        }

        Foo1(String bar) {
            this.bar = bar;
        }

        public String getBar() {
            return this.bar;
        }

        public void setBar(String bar) {
            this.bar = bar;
        }

        @Override
        public String toString() {
            return "Foo1 [bar=" + this.bar + "]";
        }

    }

}

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

# spring.kafka.consumer.value-deserializer= boot's default StringDeserializer

spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest

以及

[Foo1 [bar=baz1]]

或者,可以对序列化程序进行子类化,以设置具有适当类型的头,从而帮助反序列化程序。

相关问题