使用Kafka的同学都知道,我们每次创建Kafka主题(Topic)的时候可以指定分区数和副本数等信息,如果将这些属性配置到server.properties文件中,以后调用Java API生成的主题将使用默认值,先改变需要使用命令bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my-topic --config max.message.bytes=128000显示的修改,我们也希望将此过程在Producer调用之前通过API的方式进行设定,无需在之前或之后使用脚本进行操作,所以才了这篇文章。查看源码发现,其实内部所有的实现都是通过TopicCommand的main方法,在此记录两种方式:
1、创建主题(Topic)
【命令方式】:bin/kafka-topics.sh --zookeeper zk_host:port/chroot --create --topic my_topic_name --partitions 20 --replication-factor 3 --config x=y
【JAVA API方式】:
- String[] options = new String[]{
- "--create",
- "--zookeeper",
- "zk_host:port/chroot",
- "--partitions",
- "20",
- "--topic",
- "my_topic_name",
- "--replication-factor",
- "3",
- "--config",
- "x=y"
- };
- TopicCommand.main(options);
2、查看所有主题
【命令方式】:bin/kafka-topics.sh --list --zookeeper localhost:2181
【JAVA API方式】:
- String[] options = new String[]{
- "--list",
- "--zookeeper",
- "localhost:2181"
- };
- TopicCommand.main(options);
3、查看指定主题:
【命令方式】:bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
【JAVA API方式】:
- String[] options = new String[]{
- "--describe",
- "--zookeeper",
- "localhost:2181",
- "--topic",
- "my-replicated-topic",
- };
- TopicCommand.main(options);
4、修改主题:
【命令方式】:bin/kafka-topics.sh --zookeeper zk_host:port/chroot --alter --topic my_topic_name --deleteConfig x
【JAVA API方式】:
- String[] options = new String[]{
- "--alter",
- "--zookeeper",
- "zk_host:port/chroot",
- "--topic",
- "my_topic_name",
- "--deleteConfig",
- "x"
- };
- TopicCommand.main(options);
5、删除出题:
【命令方式】:无
【JAVA API方式】:
- String[] options = new String[]{
- "--zookeeper",
- "zk_host:port/chroot",
- "--topic",
- "my_topic_name"
- };
- DeleteTopicCommand.main(options);