java - 通过java代码启动Zookeeper和Kafka

Question

昨天我问了this question我正在尝试遵循我找到的其他一些答案，但它们无济于事。我真的不明白如何从代码中正确设置 Zookeeper，然后设置 Kafka 服务器。到目前为止我所做的是:

Properties prop = new Properties();

prop.setProperty("dataDir","C:\\kafka_2.12-2.2.0\\config\\zookeeper.properties");
prop.setProperty("bootstrap.servers", "localhost:2181");

QuorumPeerConfig quorumConfiguration = new QuorumPeerConfig();

try {
     quorumConfiguration.parseProperties(prop);
} catch(Exception e) {
   throw new RuntimeException(e);
}
            
ZooKeeperServerMain zookeeper = new ZooKeeperServerMain();
            
final ServerConfig configuration = new ServerConfig();

configuration.readFrom(quorumConfiguration);
            
            new Thread() {
                public void run() {
                    try {
                        zookeeper.runFromConfig(configuration);
                    } catch (IOException e) {
                        
                    }
                }
            }.start();
            
                    
             Properties props = new Properties();
             props.put("bootstrap.servers", "localhost:9092");
             props.put("acks", "all");
             props.put("retries", 0);
             props.put("batch.size", 16384);
             props.put("linger.ms", 1);
             props.put("buffer.memory", 33554432);
             props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
             props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
             KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
             for (int i = 0; i < 100; i++)
                 producer.send(new ProducerRecord<String, String>("info", Integer.toString(i), Integer.toString(i)));

producer.close();

我找不到比这更多的东西了。

Answer 1

强烈建议使用*-server-start命令...它们正确设置了类路径。

否则，看起来您只是在尝试编写简单的测试，并且嵌入式 Kafka 的库已经存在。例如https://github.com/embeddedkafka/embedded-kafka或using Docker