java - 卡夫卡中的空指针异常

Question

我想从 Kafka 中写入数据到 MySQL 数据库中。我已经实现了以下代码。

kafkaConsumer.subscribe(Arrays.asList(topicName));
    try {


        while (true) {
            ConsumerRecords<String, String> record = kafkaConsumer.poll(1000);
            for (ConsumerRecord<String, String> record1 : record) {
                if (record1.value().length() > 0) {
                    System.out.println(record1.value());
                    String value = record1.value();
                    String[] array = value.split(" ");
                    String sql = String.format("insert into data(timestamp, LogLevel,CityName,Detail) values ('%s', '%s','%s','%s')", array[0], array[1], array[2], array[3]);
                    mysqlConnector.statement.executeUpdate(sql);

                }
            }
        }

Answer 1

也许不值得重新发明轮子。最好的选择是使用众所周知且经过测试的工具:Kafka Connect 和 Confluent JDBC Sink Connector

https://docs.confluent.io/current/connect/kafka-connect-jdbc/sink-connector/index.html