首页
学习
活动
专区
工具
TVP
发布
精选内容/技术社群/优惠产品,尽在小程序
立即前往

在同一个Spring Kafka应用程序中读取JSON和String

,可以通过配置适当的反序列化器来实现。

首先,需要在Spring Kafka配置文件中配置Kafka消费者工厂,指定反序列化器。可以使用StringDeserializer来读取String类型的消息,使用JsonDeserializer来读取JSON类型的消息。

代码语言:txt
复制
@Configuration
@EnableKafka
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ConsumerFactory<String, String> stringConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConsumerFactory<String, MyJsonClass> jsonConsumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(MyJsonClass.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> stringKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(stringConsumerFactory());
        return factory;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, MyJsonClass> jsonKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, MyJsonClass> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(jsonConsumerFactory());
        return factory;
    }
}

然后,在消费者类中使用@KafkaListener注解来监听并处理消息。可以创建两个不同的方法,一个用于处理String类型的消息,另一个用于处理JSON类型的消息。

代码语言:txt
复制
@Component
public class KafkaConsumer {

    @KafkaListener(topics = "topic1", containerFactory = "stringKafkaListenerContainerFactory")
    public void consumeString(String message) {
        // 处理String类型的消息
        System.out.println("Received String message: " + message);
    }

    @KafkaListener(topics = "topic2", containerFactory = "jsonKafkaListenerContainerFactory")
    public void consumeJson(MyJsonClass message) {
        // 处理JSON类型的消息
        System.out.println("Received JSON message: " + message);
    }
}

以上代码示例中,MyJsonClass是一个自定义的Java类,用于表示JSON消息的结构。根据实际情况,可以根据需要定义自己的JSON类。

在这个例子中,我们使用了两个不同的topic来区分String类型和JSON类型的消息。stringKafkaListenerContainerFactory使用了stringConsumerFactory,用于处理String类型的消息;jsonKafkaListenerContainerFactory使用了jsonConsumerFactory,用于处理JSON类型的消息。

这样,同一个Spring Kafka应用程序就可以同时读取JSON和String类型的消息了。

推荐的腾讯云相关产品:腾讯云消息队列 CMQ、腾讯云云服务器 CVM、腾讯云云原生容器服务 TKE、腾讯云数据库 TencentDB、腾讯云对象存储 COS、腾讯云区块链服务 TBCS。

腾讯云产品介绍链接地址:

页面内容是否对你有帮助?
有帮助
没帮助

相关·内容

没有搜到相关的沙龙

领券