我們都知道柑潦，有時候我們創(chuàng)建topic的時候一些屬性沒有設置好埂软，導致后面出現(xiàn)很多狀況锈遥，下面介紹如何創(chuàng)建，修改所灸，刪除kafka主題的簡單操作诉字。如下介紹2種方法陵霉。
方法1：
首先，創(chuàng)建一個pojo類：
package com.ky.common;
/**
@Author: xwj
@Date: 2019/1/7 0007 13:53
-
@Version 1.0
*/
public class KafkaTopicBean {private String topicName;
private Integer partition;
private Integer replication;
private String descrbe;String getTopicName() {
return topicName;
}public void setTopicName(String topicName) {
this.topicName = topicName;
}Integer getPartition() {
return partition;
}public void setPartition(Integer partition) {
this.partition = partition;
}Integer getReplication() {
return replication;
}public void setReplication(Integer replication) {
this.replication = replication;
}public String getDescrbe() {
return descrbe;
}public void setDescrbe(String descrbe) {
this.descrbe = descrbe;
}@Override
public String toString() {
return "KafkaTopicBean [topicName=" + topicName + ", partition=" + partition
+ ", replication=" + replication + ", descrbe=" + descrbe +"]";
}
}
創(chuàng)建kafkaUtil工具類
package com.ky.common;
import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.server.ConfigType;
import kafka.utils.ZkUtils;
import org.apache.kafka.common.security.JaasUtils;
import scala.collection.JavaConversions;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;
/**
@Author: xwj
@Date: 2019/1/7 0007 13:54
-
@Version 1.0
*/
public class KafkaUtil {public static void createKafaTopic(String zkStr, KafkaTopicBean topic) {
ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
try {
AdminUtils.createTopic(zkUtils, topic.getTopicName(), topic.getPartition(),
topic.getReplication(), new Properties(), RackAwareMode.Enforced);
System.out.printf("create topic:%s success..", topic.getTopicName());
} catch (Exception e) {
System.out.printf("create topic error..the root cause:", e.getMessage());
}
}public static void deleteKafaTopic(String zkStr, KafkaTopicBean topic) {
ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
try {
AdminUtils.deleteTopic(zkUtils, topic.getTopicName());
System.out.printf("delete topic:%s success..", topic.getTopicName());
} catch (Exception e) {
System.out.printf("create topic error..the root cause:", e.getMessage());
}
zkUtils.close();
}public static void updateTopic(String zkStr, KafkaTopicBean topic) {
ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
try {
Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topic.getTopicName());
props.put("min.cleanable.dirty.ratio", "0.3");
props.remove("max.message.bytes");
AdminUtils.changeTopicConfig(zkUtils, topic.getTopicName(), props);
System.out.printf("update topic:%s success..", topic.getTopicName());
} catch (Exception e) {
System.out.printf("create topic error..the root cause:", e.getMessage());
}
zkUtils.close();
}public static List<String> getAllTopics(String zkStr) {
ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
List<String> topicList = null;
try {
List<String> allTopicList = JavaConversions.seqAsJavaList(zkUtils.getAllTopics());
topicList = allTopicList.stream()
.filter(topic -> !topic.equalsIgnoreCase("rawMessage") && !topic.equalsIgnoreCase("blmessage"))
.collect(Collectors.toList());
} catch (Exception e) {
System.out.printf("create topic error..the root cause:", e.getMessage());
}
return topicList;
}
}
測試：
package com.ky.service;
import com.ky.common.KafkaTopicBean;
import com.ky.common.KafkaUtil;
/**
@Author: xwj
@Date: 2019/1/7 0007 13:51
-
@Version 1.0
*/
public class KafkaService {public static void main(String[] args) {
String zkStr = "192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
String topic = "rawMessage";
final KafkaTopicBean topicBean = new KafkaTopicBean();
topicBean.setPartition(1);
topicBean.setReplication(1);
topicBean.setTopicName(topic);
KafkaUtil.createKafaTopic(zkStr, topicBean);
}
}
然后去linux上去查看运嗜，發(fā)現(xiàn)topic已經(jīng)創(chuàng)建了。
但是刪除kafka的topic岭参，只是標記被刪除了，并沒有真正的刪除，要做真正的刪除程癌，需要做如下動作。
打開kafka的配置文件，并修改如下屬性：
auto.create.topics.enable 設置為false，不讓程序自動創(chuàng)建topic，就是程序再往kafka生產(chǎn)數(shù)據(jù)的時候蚕涤，如果發(fā)現(xiàn)topic不存在就會自動創(chuàng)建。
delete.topic.enable設置為true较沪，刪除topic的同時们何，會過一段時間刪除目錄和對應數(shù)據(jù)文件目錄。
方法2：
基于方法1上冒签，代碼如下
import org.I0Itec.zkclient.ZkClient;
import java.util.List;
/**
@Author: xwj
@Date: 2019/1/7 0007 16:52
-
@Version 1.0
*/
public class ZkService {public static void main(String[] args) {
String zkStr = "192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
String path = "/brokers/topics/";
final List<String> allTopics = KafkaUtil.getAllTopics(zkStr);
ZkClient zkClient;
zkClient = new ZkClient(zkStr, 30000, 30000);
for (String topic : allTopics) {
try {
zkClient.deleteRecursive(path + topic);
System.out.println("delete topic sucess..");
} catch (Exception e) {
e.printStackTrace();
}
}}
}
