kafka單機搭建及操作做個記錄

最近一直在搞springboot整合kafka，於是自己搭建了一套單機的kafka環境，以便用於測試

wget

tar -xzvf kafka_2.11-1.1.0.tgz

2.kafka需要安裝zookeeper使用，但kafka整合zookeeper，在單機搭建時可直接使用，只需配置kafka_2.11-1.1.0/config 下的「zookeeper.properties」

以下是修改的zookeeper.properties配置引數：

#建立zookeeper目錄 mkdir /usr/local/kafka/zookeeper #建立zookeeper日誌目錄 mkdir -p /usr/local/kafka/log/zookeeper #進入配置目錄 cd /usr/local/kafka/config vi zookeeper.properties #zookeeper資料目錄 datadir=/usr/local/kafka/zookeeper #zookeeper日誌目錄 datalogdir=/usr/local/kafka/log/zookeeper clientport=2181 maxclientcnxns=100 ticktime=2000 initlimit=10 synclimit=5

3.配置kafka_2.11-1.1.0/config下的「server.properties」，修改log.dirs和zookeeper.connect。前者是日誌存放資料夾，後者是zookeeper連線位址（埠和clientport保持一致），目錄建立及引數配置如下：

#建立kafka日誌目錄 mkdir /usr/local/kafka/log/kafka #進入配置目錄 cd /usr/local/kafka/config #編輯修改kafka相應的引數 vi server.properties broker.id=0 #topic可以刪,預設是false delete.topic.enable=true #埠號，可不配置 port=9092 #伺服器ip位址，也可修改為自己的伺服器ip host.name=192.254.64.128 #日誌存放路徑，上面建立的目錄 log.dirs=/usr/local/kafka/log/kafka #zookeeper位址和埠，單機配置部署，localhost:2181 zookeeper.connect=192.254.64.128:2181 listeners=plaintext:

到此，kafka的單機環境就搭建成功了。

啟動zookeeper

啟動啟動kakfa

/usr/local/kafka/bin/kafka-server-start.sh-daemon/usr/local/kafka/config/server.properties

-daemon：這個比較關鍵需要注意，守護執行緒，要不然xshell一關程序就丟了

jps程序顯示如下：

./kafka-topics.sh --zookeeper 192.168.226.129:2182 --describe --topic ordertopic ；檢視名字為ordertopic的topic

./kafka-topics.sh --zookeeper 192.168.226.129:2182 --list ；檢視topic 列表

記一段問題：（kafka丟失最開始日誌訊息的問題）

props = new properties();
props.put(consumerconfig.bootstrap_servers_config, address);
props.put(consumerconfig.enable_auto_commit_config, false);
props.put(consumerconfig.auto_commit_interval_ms_config, 1000);
props.put(consumerconfig.session_timeout_ms_config, 100000);
props.put(consumerconfig.request_timeout_ms_config,110000);
props.put(consumerconfig.key_deserializer_class_config, stringdeserializer.class);
props.put(consumerconfig.value_deserializer_class_config, stringdeserializer.class);
props.put(consumerconfig.group_id_config, "testconsumer"+system.currenttimemillis());
props.put(consumerconfig.auto_offset_reset_config, "earliest");
props.put(consumerconfig.max_poll_records_config, 50);
public static mapconsume(string topics, maprtnmap) 
consumer.subscribe(collections.singletonlist(topics));
stringbuffer sb = new stringbuffer();
consumerrecordsrecords = null;
records = consumer.poll(duration.ofmillis(1000));
for (consumerrecordrecord : records) 
if(stringutils.isnotempty(sb.tostring()))
rtnmap.put("msg", sb.tostring());
return rtnmap;
}

records = consumer.poll(duration.ofmillis(1000));留意一下就好，遇到具體問題具體對待

kafka單機搭建及操作做個記錄

kafka 單機環境kafka搭建

Kafka單機模式搭建

Kafka單機Windows環境搭建

kafka單機搭建及操作 做個記錄

kafka 單機環境kafka搭建

Kafka單機模式搭建

Kafka單機Windows環境搭建

相關推薦

kafka單機搭建及操作做個記錄