kafka消费者使用json？

vatpfxk5 于 2021-07-16 发布在 Java

关注(0)|答案(1)|浏览(352)

有没有可能使用kafka从post-http请求中获取json对象，将它们放入topic中，然后发送给消费者（数据库）？
顺便说一句，这是我的Kafka图型课程：

@EnableKafka
@Configuration
public class KafkaConfig {

    @Bean
    public KafkaTemplate<String, User> kafkaTemplate(){
        return new KafkaTemplate<>(producerFactory());
    }
    @Bean
    static public ProducerFactory<String,User> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public ConsumerFactory<String,User> consumerFactory(){
        Map<String, Object> config =  new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG,"group_id");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(config);
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String,User> kafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String,User> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}

Java spring

来源：https://stackoverflow.com/questions/66664874/kafka-consumer-with-json

1条答案

按热度按时间

vlf7wbxs1#

我假设您知道如何使用spring项目创建post rest点。基本上，在从端点获得json输入之后，就可以使用kafkatemplate引用将json对象发送到kafka。像这样的伪代码

@RestController
class ExampleController

@Autowired
private final KafkaTemplate kafkaTemplate;

@PostMapping("/anyPath")
public void post(final ObjectAsJson yourObject) {

   kafkaTemplate.doSend(// here map your object to a Producer Record)
   // depending on your use you can return a custom success response
}

然后，您可以将一个方法与kafkalistener注解连接起来，以使用它并将其写入数据库。

@KafkaListener(topics = "topicName", groupId = "foo", containerFactory = "kafkaListenerContainerFactory")
public void listen(YourCustomObject message) {
    // with kafkaListenerContainerFactory it should deserialise it to your desired object and here you can just write your database insertion here
}

我还想看看kafka connect，它有助于像这样的集成，你想实现http作为源和数据库作为接收器和kafka之间的主题。
希望对你有帮助。

赞(0）回复(0）举报 2021-07-16

我来回答

kafka消费者使用json？

1条答案

相关问题

热门标签

最新问答