mysql 同步kafka

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class MySQLToKafkaSync {
    public static void main(String[] args) {
        String mysqlUrl = "jdbc:mysql://localhost:3306/mydatabase";
        String mysqlUser = "user";
        String mysqlPassword = "password";
        String kafkaBootstrapServers = "localhost:9092";
        String kafkaTopic = "mytopic";

        try (Connection conn = DriverManager.getConnection(mysqlUrl, mysqlUser, mysqlPassword);
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery("SELECT * FROM mytable")) {

            KafkaProducer<String, String> producer = new KafkaProducer<>(getKafkaProperties(kafkaBootstrapServers));

            while (rs.next()) {
                String data = rs.getString("data");
                producer.send(new ProducerRecord<>(kafkaTopic, data));
            }

            producer.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static Properties getKafkaProperties(String bootstrapServers) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }
}