我想通过Java在Kafka(kafka_2.8.0-0.8.1.1)中创建一个主题。如果我在命令提示符下创建一个主题,并且如果我通过java api推送消息,它也可以正常工作。但是我想通过java api创建一个主题。经过长时间的搜索,我发现了以下代码,
ZkClient zkClient = new ZkClient("localhost:2181", 10000, 10000); AdminUtils.createTopic(zkClient, myTopic, 10, 1, new Properties());
我尝试了上面的代码,它表明创建了主题,但是无法在该主题中推送消息。我的代码有什么问题吗?还是通过其他方式实现以上目标?
编辑 -较新版本的Kafka不需要Zookeeper。请参阅@Neeleshkumar Srinivasan Mannur的API版本1.01.0以上的答案
API 0.11.0+中的过程似乎已大大简化。使用它,可以完成以下操作
import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.CreateTopicsResult; import org.apache.kafka.clients.admin.NewTopic; Properties properties = new Properties(); properties.load(new FileReader(new File("kafka.properties"))); AdminClient adminClient = AdminClient.create(properties); NewTopic newTopic = new NewTopic("topicName", 1, (short)1); //new NewTopic(topicName, numPartitions, replicationFactor) List<NewTopic> newTopics = new ArrayList<NewTopic>(); newTopics.add(newTopic); adminClient.createTopics(newTopics); adminClient.close();
kafka.properties文件的内容如下
bootstrap.servers=localhost:9092 group.id=test enable.auto.commit=true auto.commit.interval.ms=1000 key.deserializer=org.apache.kafka.common.serialization.StringDeserializer value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
请注意,必须关闭AdminClient的实例才能反映新创建的主题。
我修好了..经过长时间的研究..
通过上面的代码,ZkClient将创建一个主题,但是该主题信息将不了解kafka。所以我们要做的是,我们需要通过以下方式为ZkClient创建对象,
首先导入以下语句,
import kafka.utils.ZKStringSerializer$;
并通过以下方式为ZkClient创建对象,
ZkClient zkClient = new ZkClient("localhost:2181", 10000, 10000, ZKStringSerializer$.MODULE$); AdminUtils.createTopic(zkClient, myTopic, 10, 1, new Properties());
由于api已更改,因此上述代码不适用于kafka> 0.9,请使用以下代码适用于kafka> 0.9
import java.util.Properties; import kafka.admin.AdminUtils; import kafka.utils.ZKStringSerializer$; import kafka.utils.ZkUtils; import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.ZkConnection; public class KafkaTopicCreationInJava { public static void main(String[] args) throws Exception { ZkClient zkClient = null; ZkUtils zkUtils = null; try { String zookeeperHosts = "192.168.20.1:2181"; // If multiple zookeeper then -> String zookeeperHosts = "192.168.20.1:2181,192.168.20.2:2181"; int sessionTimeOutInMs = 15 * 1000; // 15 secs int connectionTimeOutInMs = 10 * 1000; // 10 secs zkClient = new ZkClient(zookeeperHosts, sessionTimeOutInMs, connectionTimeOutInMs, ZKStringSerializer$.MODULE$); zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperHosts), false); String topicName = "testTopic"; int noOfPartitions = 2; int noOfReplication = 3; Properties topicConfiguration = new Properties(); AdminUtils.createTopic(zkUtils, topicName, noOfPartitions, noOfReplication, topicConfiguration); } catch (Exception ex) { ex.printStackTrace(); } finally { if (zkClient != null) { zkClient.close(); } } } }