我们如何使用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
我按照原样关注了开发人员设置。
在Kafka 0.8.1+ - Kafka的最新版本 - 您可以通过AdminCommand以编程方式创建一个新主题。 CreateTopicCommand（旧Kafka 0.8.0的一部分）的功能在此问题的先前答案之一中被提及移至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();
}
}
您尝试从哪个IDE？
请提供完整的路径，下面是来自终端的命令，它将创建一个主题
cd kafka/bin./kafka-create-topic.sh --topic test --zookeeper localhost:2181从Kafka 0.10.1开始，Michael提到的ZKStringSerializer是私有的（对于Scala）。您可以在ZkUtils中使用工厂方法createZkClient或createZkClientAndConnection。
Scaf示例为Kafka 0.10.1：
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)
从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)));
此命令的输出是CreateTopicsResult，您可以使用它来获取整个操作或每个主题创建的Future：
CreateTopicsResult#all()。Futures，请使用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上的一个薄的垫片。”
要通过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());
您可以尝试使用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和zkclient-0.1添加maven依赖项
基于最新的kafka-client api和Kafka 1.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" % "1.0.0")
在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
如果您使用的是Kafka 0.10.0.0+，则从Java创建主题需要传递RackAwareMode类型的参数。它是一个Scala案例对象，从Java获取它的实例很棘手（例如：How do I "get" a Scala case object from Java?。但它不适用于我们的情况）。
幸运的是，rackAwareMode是一个可选参数。然而，Java不支持可选参数。我们如何解决这个问题？这是一个解决方案：
AdminUtils.createTopic(zkUtils, topic, 1, 1,
AdminUtils.createTopic$default$5(),
AdminUtils.createTopic$default$6());
将它与miguno的答案一起使用，你很高兴。
一些方法你的电话不起作用。
从Kafka 0.8 Producer Example下面的示例将创建一个名为page_visits的主题，并且如果在Kafka auto.create.topics.enable文件中将true属性设置为Broker config（默认），也会开始生成
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();
}
}
我们可以使用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/