如何在Java企业应用中通过编程方式启动Kafka Broker与ZooKeeper

一、编程启动ZooKeeper #

ZooKeeper提供原生Java API可直接启动服务，无需调用命令行脚本，步骤如下：

• 引入依赖（以Maven为例）：

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>你的ZooKeeper版本号</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>

• 编写启动代码：

import org.apache.zookeeper.server.ServerConfig;
import org.apache.zookeeper.server.ZooKeeperServerMain;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig;

import java.io.File;
import java.io.IOException;

public class EmbeddedZooKeeper {
    private ZooKeeperServerMain zkServer;

    public void start(String configPath) throws Exception {
        // 加载zookeeper.properties配置
        QuorumPeerConfig quorumConfig = new QuorumPeerConfig();
        quorumConfig.parse(new File(configPath));
        
        ServerConfig serverConfig = new ServerConfig();
        serverConfig.readFrom(quorumConfig);
        
        // 异步启动ZooKeeper，避免阻塞主线程
        zkServer = new ZooKeeperServerMain();
        new Thread(() -> {
            try {
                zkServer.runFromConfig(serverConfig);
            } catch (IOException e) {
                throw new RuntimeException("ZooKeeper启动失败", e);
            }
        }).start();
    }

    public void stop() {
        if (zkServer != null) {
            zkServer.shutdown();
        }
    }
}

调用时传入你的zookeeper.properties文件路径即可。

二、编程启动Kafka Broker #

Kafka支持嵌入式启动，核心是调用其内部启动类，步骤如下：

• 引入依赖（以Maven为例）：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.13</artifactId>
    <version>你的Kafka版本号</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>

• 编写启动代码：

import kafka.server.KafkaServerStartable;
import scala.collection.JavaConverters;

import java.io.FileInputStream;
import java.util.Properties;

public class EmbeddedKafka {
    private KafkaServerStartable kafkaStartable;

    public void start(String configPath) throws Exception {
        // 加载server.properties配置
        Properties props = new Properties();
        props.load(new FileInputStream(configPath));
        
        // 转换为Kafka所需的Scala类型配置
        scala.collection.immutable.Map<String, String> scalaProps = 
                JavaConverters.mapAsScalaMapConverter(props).asScala().toMap();
        
        kafkaStartable = KafkaServerStartable.fromProps(scalaProps);
        kafkaStartable.startup();
    }

    public void stop() {
        if (kafkaStartable != null) {
            kafkaStartable.shutdown();
        }
    }
}

注意：Kafka内部API可能随版本变动，建议使用与你手动启动时一致的Kafka版本，避免兼容性问题。

三、关键注意事项 #

• 启动顺序：必须先启动ZooKeeper，待其完全就绪后再启动Kafka Broker，可通过检查ZooKeeper端口的连通性判断就绪状态。
• 配置权限：确保zookeeper.properties和server.properties中指定的数据目录、日志目录在Java应用运行时有读写权限。
• 场景限制：这种嵌入式启动方式更适合本地开发、自动化测试场景，生产环境不建议使用，生产环境应采用独立部署的Kafka与ZooKeeper集群。

内容的提问来源于stack exchange，提问作者Shiva Kumar