54

我们如何使用 API 从 IDE 在 Kafka 中创建主题，因为当我这样做时：

bin/kafka-create-topic.sh --topic mytopic --replica 3 --zookeeper localhost:2181

我得到错误：

bash: bin/kafka-create-topic.sh: No such file or directory

我按照开发人员的设置进行了操作。

4

11 回答 11

74

在 Kafka 0.8.1+（目前最新版本的 Kafka）中，您可以通过AdminCommand. 该问题的先前答案之一中提到的（旧版 Kafka 0.8.0 的一部分）的功能CreateTopicCommand已移至AdminCommand.

Kafka 0.8.1 的 Scala 示例：

import kafka.admin.AdminUtils
import kafka.utils.ZKStringSerializer
import org.I0Itec.zkclient.ZkClient

// Create a ZooKeeper client
val sessionTimeoutMs = 10000
val connectionTimeoutMs = 10000
// Note: You must initialize the ZkClient with ZKStringSerializer.  If you don't, then
// createTopic() will only seem to work (it will return without error).  The topic will exist in
// only ZooKeeper and will be returned when listing topics, but Kafka itself does not create the
// topic.
val zkClient = new ZkClient("zookeeper1:2181", sessionTimeoutMs, connectionTimeoutMs,
    ZKStringSerializer)

// Create a topic named "myTopic" with 8 partitions and a replication factor of 3
val topicName = "myTopic"
val numPartitions = 8
val replicationFactor = 3
val topicConfig = new Properties
AdminUtils.createTopic(zkClient, topicName, numPartitions, replicationFactor, topicConfig)

构建依赖关系，以 sbt 为例：

libraryDependencies ++= Seq(
  "com.101tec" % "zkclient" % "0.4",
  "org.apache.kafka" % "kafka_2.10" % "0.8.1.1"
    exclude("javax.jms", "jms")
    exclude("com.sun.jdmk", "jmxtools")
    exclude("com.sun.jmx", "jmxri"),
  ...
)

编辑：为 Kafka 0.9.0.0 添加了 Java 示例（截至 2016 年 1 月的最新版本）。

Maven依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.9.0.0</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.7</version>
</dependency>

代码：

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

import java.util.Properties;

import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

public class KafkaJavaExample {

  public static void main(String[] args) {
    String zookeeperConnect = "zkserver1:2181,zkserver2:2181";
    int sessionTimeoutMs = 10 * 1000;
    int connectionTimeoutMs = 8 * 1000;
    // Note: You must initialize the ZkClient with ZKStringSerializer.  If you don't, then
    // createTopic() will only seem to work (it will return without error).  The topic will exist in
    // only ZooKeeper and will be returned when listing topics, but Kafka itself does not create the
    // topic.
    ZkClient zkClient = new ZkClient(
        zookeeperConnect,
        sessionTimeoutMs,
        connectionTimeoutMs,
        ZKStringSerializer$.MODULE$);

    // Security for Kafka was added in Kafka 0.9.0.0
    boolean isSecureKafkaCluster = false;
    ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperConnect), isSecureKafkaCluster);

    String topic = "my-topic";
    int partitions = 2;
    int replication = 3;
    Properties topicConfig = new Properties(); // add per-topic configurations settings here
    AdminUtils.createTopic(zkUtils, topic, partitions, replication, topicConfig);
    zkClient.close();
  }

}

编辑 2：为 Kafka 0.10.2.0 添加了 Java 示例（截至 2017 年 4 月的最新版本）。

Maven依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.10.2.0</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.9</version>
</dependency>

代码：

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

import java.util.Properties;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

public class KafkaJavaExample {

  public static void main(String[] args) {
    String zookeeperConnect = "zkserver1:2181,zkserver2:2181";
    int sessionTimeoutMs = 10 * 1000;
    int connectionTimeoutMs = 8 * 1000;

    String topic = "my-topic";
    int partitions = 2;
    int replication = 3;
    Properties topicConfig = new Properties(); // add per-topic configurations settings here

    // Note: You must initialize the ZkClient with ZKStringSerializer.  If you don't, then
    // createTopic() will only seem to work (it will return without error).  The topic will exist in
    // only ZooKeeper and will be returned when listing topics, but Kafka itself does not create the
    // topic.
    ZkClient zkClient = new ZkClient(
        zookeeperConnect,
        sessionTimeoutMs,
        connectionTimeoutMs,
        ZKStringSerializer$.MODULE$);

    // Security for Kafka was added in Kafka 0.9.0.0
    boolean isSecureKafkaCluster = false;

    ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperConnect), isSecureKafkaCluster);
    AdminUtils.createTopic(zkUtils, topic, partitions, replication, topicConfig, RackAwareMode.Enforced$.MODULE$);
    zkClient.close();
  }

}
于 2014-04-29T09:00:25.243 回答
37

从 0.11.0.0 开始，您只需要：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.0</version>
</dependency>

该工件现在包含AdminClient( org.apache.kafka.clients.admin)。

AdminClient可以处理许多 Kafka 管理任务，包括主题创建：

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");

AdminClient admin = AdminClient.create(config);

Map<String, String> configs = new HashMap<>();
int partitions = 1;
int replication = 1;

admin.createTopics(asList(new NewTopic("topic", partitions, replication).configs(configs)));

此命令的输出是 a CreateTopicsResult，您可以使用它来获取Future整个操作或每个单独主题创建的 a：

  • 要获得整个操作的未来，请使用CreateTopicsResult#all().
  • 要Future单独获取所有主题的 s，请使用CreateTopicsResult#values().

例如：

CreateTopicsResult result = ...
KafkaFuture<Void> all = result.all();

或者：

CreateTopicsResult result = ...
for (Map.Entry<String, KafkaFuture<Void>> entry : result.values().entrySet()) {
    try {
        entry.getValue().get();
        log.info("topic {} created", entry.getKey());
    } catch (InterruptedException | ExecutionException e) {
        if (Throwables.getRootCause(e) instanceof TopicExistsException) {
            log.info("topic {} existed", entry.getKey());
        }
    }
}

KafkaFuture是“支持调用链和其他异步编程模式的灵活未来”，并且“最终将成为 Java 8 之上的薄垫片CompletebleFuture”。

于 2017-07-15T21:37:25.347 回答
14

要通过 java api 和 Kafka 0.8+ 创建主题，请尝试以下操作，

首先导入下面的语句

import kafka.utils.ZKStringSerializer$;

通过以下方式为 ZkClient 创建对象，

ZkClient zkClient = new ZkClient("localhost:2181", 10000, 10000, ZKStringSerializer$.MODULE$);
AdminUtils.createTopic(zkClient, myTopic, 10, 1, new Properties());
于 2014-11-20T12:22:53.790 回答
10

您可以尝试使用 kafka.admin.CreateTopicCommand scala 类从 Java 代码创建主题...提供必要的参数。

String [] arguments = new String[8];
arguments[0] = "--zookeeper";
arguments[1] = "10.***.***.***:2181";
arguments[2] = "--replica";
arguments[3] = "1";
arguments[4] = "--partition";
arguments[5] = "1";
arguments[6] = "--topic";
arguments[7] = "test-topic-Biks";

CreateTopicCommand.main(arguments);

注意：jopt-simple-4.5您应该为&添加 Maven 依赖项zkclient-0.1

于 2013-08-28T06:20:44.607 回答
6

基于最新的kafka-client api和Kafka 2.1.1，工作版本代码如下：

使用 sbt 导入最新的 kafka-clients。

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
libraryDependencies += Seq("org.apache.kafka" % "kafka-clients" % "2.1.1",
"org.apache.kafka" %% "kafka" % "2.1.1")

scala中创建主题的代码：

import java.util.Arrays
import java.util.Properties

import org.apache.kafka.clients.admin.NewTopic
import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig}

class CreateKafkaTopic {
  def create(): Unit = {
    val config = new Properties()
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.30.1.5:9092")

    val localKafkaAdmin = AdminClient.create(config)

    val partitions = 3
    val replication = 1.toShort
    val topic = new NewTopic("integration-02", partitions, replication)
    val topics = Arrays.asList(topic)

    val topicStatus = localKafkaAdmin.createTopics(topics).values()
    //topicStatus.values()
    println(topicStatus.keySet())
  }

}

使用以下方法验证新主题：

./kafka-topics.sh --zookeeper 192.30.1.5:2181 --list

希望它对某人有帮助。参考： http: //kafka.apache.org/21/javadoc/index.html ?org/apache/kafka/clients/admin/AdminClient.html

于 2019-02-21T09:32:04.680 回答
2

如果您使用的是 Kafka 0.10.0.0+，则从 Java 创建主题需要传递 RackAwareMode 类型的参数。这是一个 Scala 案例对象，从 Java 中获取它的实例很棘手（例如，证明：我如何从 Java 中“获取”一个 Scala 案例对象？但它不适用于我们的案例）。

幸运的是， rackAwareMode 是一个可选参数。然而 Java 不支持可选参数。我们如何解决这个问题？这是一个解决方案：

AdminUtils.createTopic(zkUtils, topic, 1, 1, 
    AdminUtils.createTopic$default$5(),
    AdminUtils.createTopic$default$6());

将它与 miguno 的答案一起使用，您就可以开始了。

于 2016-07-18T17:26:54.220 回答
1

从 Kafka 0.8生产者示例中，如果在 Kafka Broker 配置文件中将属性设置为（默认），下面的示例将创建一个名为的主题page_visits并开始生产auto.create.topics.enabletrue

import java.util.*;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class TestProducer {
    public static void main(String[] args) {
        long events = Long.parseLong(args[0]);
        Random rnd = new Random();

        Properties props = new Properties();
        props.put("metadata.broker.list", "broker1:9092,broker2:9092 ");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "example.producer.SimplePartitioner");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, String> producer = new Producer<String, String>(config);

        for (long nEvents = 0; nEvents < events; nEvents++) { 
            long runtime = new Date().getTime();  
            String ip = “192.168.2.” + rnd.nextInt(255); 
            String msg = runtime + “,www.example.com,” + ip; 
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
            producer.send(data);
        }
        producer.close();
   }
}
于 2013-08-23T22:49:19.587 回答
1

您的电话无法正常工作的几种方式。

  1. 如果您的 Kafka 集群没有足够的节点来支持复制值 3。

  2. 如果有 chroot 路径前缀，则必须在 zookeeper 端口之后附加它

  3. 运行时不在 Kafka 安装目录中（这是最有可能的）

于 2013-12-19T23:16:51.320 回答
1

我们AdminZkClient可以使用它来管理 Kafka 服务器中的主题。

String zookeeperHost = "127.0.0.1:2181";
Boolean isSucre = false;
int sessionTimeoutMs = 200000;
int connectionTimeoutMs = 15000;
int maxInFlightRequests = 10;
Time time = Time.SYSTEM;
String metricGroup = "myGroup";
String metricType = "myType";
KafkaZkClient zkClient = KafkaZkClient.apply(zookeeperHost,isSucre,sessionTimeoutMs,
                connectionTimeoutMs,maxInFlightRequests,time,metricGroup,metricType);

AdminZkClient adminZkClient = new AdminZkClient(zkClient);

String topicName1 = "myTopic";
int partitions = 3;
int replication = 1;
Properties topicConfig = new Properties();

adminZkClient.createTopic(topicName1,partitions,replication,
            topicConfig,RackAwareMode.Disabled$.MODULE$);

您可以参考此链接了解详情 https://www.analyticshut.com/streaming-services/kafka/create-and-list-kafka-topics-in-java/

于 2018-07-02T18:18:20.213 回答
0

您从哪个 IDE 尝试？

请提供完整路径，以下是来自终端的命令，它将创建一个主题

  1. cd kafka/bin
  2. ./kafka-create-topic.sh --topic test --zookeeper localhost:2181
于 2013-06-23T07:59:56.003 回答
-1

从 Kafka 0.10.1 开始，Michael 提到的 ZKStringSerializer 是私有的（对于 Scala）。您可以使用 ZkUtils 中的工厂方法 createZkClient 或 createZkClientAndConnection。

Kafka 0.10.1 的 Scala 示例：

import kafka.utils.ZkUtils

val sessionTimeoutMs = 10000
val connectionTimeoutMs = 10000
val (zkClient, zkConnection) = ZkUtils.createZkClientAndConnection(
  "localhost:2181", sessionTimeoutMs, connectionTimeoutMs)

然后按照 Michael 的建议创建主题：

import kafka.admin.AdminUtils

val zkUtils = new ZkUtils(zkClient, zkConnection, false)
val numPartitions = 4
val replicationFactor = 1
val topicConfig = new Properties
val topic = "my-topic"
AdminUtils.createTopic(zkUtils, topic, numPartitions, replicationFactor, topicConfig)
于 2017-06-20T08:58:23.923 回答