为每个kafka消息致电远程REST API

Question

我正在Kafka生产商中读取文件（逐行），并在流中读取每条记录（即每行），而Kafka消费者正在收到此记录。Further我需要将其发送给KAFKA消费者的REST API服务以进行一些处理我的回应将发送给其他话题，依此类推。 我不确定如何使用生产者-Compumer -Chafka API进行操作？ 需要帮助

Answer 1

这很简单。这就是您需要与Kafka Consumer Client一起开发Spring Boot应用程序的方式，该应用程序与任何后端REST API集成在一起。

1. 一个可以使消费者对象的类

package x.x.x.x.x.

@Service

public class KAFKAConsumer implements Runnable{

    public void run(String topicName) {
        try {
            Consumer<String, String> consumer = new KafkaConsumer<>();
            consumer.subscribe(Collections.singleton(topicName));
                while (true) {
                    try {
                        LOGGER.debug("Polling topic for new events...");
                        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(5000));
                        for (TopicPartition partition : consumerRecords.partitions()) {
                            List<ConsumerRecord<String, String>> partitionRecords = consumerRecords.records(partition);
                            for (ConsumerRecord<String, String> record : partitionRecords) {

                                // Calling REST API for each record

                                invokeRestAPIClass(Parameters....)

                                try {
                                    consumer.commitSync();
                                } catch (CommitFailedException e) {
                                }
                            }
                        }
                    } catch (SerializationException e1) {
                        continue;
                    } catch (Exception e2) {
                        continue;
                    }
                }

            } finally {
                consumer.close();
            }
        } catch (Exception e) {
        }
    }

2. 定义REST API模板的类，

您可以 https://howtodoinjava.com/spring-boot2/resttemplate/spring-resttul-client-client-resttemplate-example/ 作为构建REST API模板的参考），根据REST API请求和响应结构，通过POJO类映射您的请求。

3. 实施春季启动启动主类

package x.x.x.x.x.

@SpringBootApplication(scanBasePackages = "x.x.x.x.x")

@EnableAspectJAutoProxy

public class SpringBootStartup implements CommandLineRunner {

    @Inject
    private KAFKAConsumer consumer;

    @Override
    public void run(String... args){
        try {
            Thread springBootThread = new Thread(consumer);
            springBootThread.start();
        }catch(Exception ex)
        {
        }
    }

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

}