單機配置kafka和zookeeper
1:環境準備
jdk 推薦oracle，不建議open sdk
在/etc/profile加入下列環境變量
在PATH中將jdk和jre的bin加入path里面
$JAVA_HOME/bin:$JRE_HOME/bin
2：安裝zookeeper
下載zookeeper tar
url: https://archive.apache.org/dist/zookeeper/zookeeper-3.4.5/zookeeper-3.4.5.tar.gz
將壓縮包移動到/usr/local下面
tar -zxvf ***
更改配置文件
（1）將conf/zoo_sample.cfg更改為zoo.cfg
（2）更改配置如下
注意：
*其中默認port為2181。
*datadir需手動創建 mkdir -p datadir
*注釋掉的參數在單機中無用
（3）加入環境變量 /etc/profile
測試：
可以cd 到bin目錄下通過執行 zkServer.sh shell腳本啟動、停止或者查看狀態
eg:
./zkServer.sh start/stop/status
3：安裝kafka
先下載tar包、解壓、mv到/usr/local下面
wget https://www.apache.org/dyn/closer.cgi?path=/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz sudo mv *** /usr/local cd /usr/local sudo tar -zxvf ****
sudo rm **.tar.gz
修改config目錄下配置文件
vim server.properties
修改如下參數
#broker.id需改成正整數，單機為1就好 broker.id=1 #指定端口號 port=9092 #localhost這一項還有其他要修改，詳細見下面說明 host.name=localhost #指定kafka的日志目錄 log.dirs=/usr/local/kafka_2.11-0.11.0.0/kafka-logs #連接zookeeper配置項，這里指定的是單機，所以只需要配置localhost，若是實際生產環境，需要在這里添加其他ip地址和端口號 zookeeper.connect=localhost:2181
vim zookeeper.properties
修改如下參數
#數據目錄 dataDir=/usr/local/kafka_2.11-0.11.0.0/zookeeper/data #客戶端端口 clientPort=2181 host.name=localhost
producer.properties and consumer.properties
zookeeper.connect=localhost:2181
4：啟動kafka
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties
終端都是info log
注意：如果是非root用戶，可能會使用到sudo去啟動zookeeper和kafka，但是那樣會失敗必須將kafka整個目錄下的文件的owner和group都改為自己的用戶。
可以通過jps命令查看兩者是否啟動成功
2576 QuorumPeerMain表示zookeeper
如果發現沒啟動成功，可以在zookeeper/config/zookeeper.out里debug。
成功啟動后，可以簡單通過demo進行測試
github有各種語言的kafka支持
url：https://github.com/edenhill/librdkafka/tree/v0.9.5
可以簡單通過go的demo進行測試
consumer
import ( "fmt" "github.com/confluentinc/confluent-kafka-go/kafka" ) func main() { c, err := kafka.NewConsumer(&kafka.ConfigMap{ "bootstrap.servers": "localhost", "group.id": "myGroup", "auto.offset.reset": "earliest", }) if err != nil { panic(err) } c.SubscribeTopics([]string{"myTopic", "^aRegex.*[Tt]opic"}, nil) for { msg, err := c.ReadMessage(-1) if err == nil { fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value)) } else { fmt.Printf("Consumer error: %v (%v)\n", err, msg) break } } c.Close() }
producer
import ( "fmt" "github.com/confluentinc/confluent-kafka-go/kafka" ) func main() { p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost"}) if err != nil { panic(err) } go func() { for e := range p.Events() { switch ev := e.(type) { case *kafka.Message: if ev.TopicPartition.Error != nil { fmt.Printf("Delivery failed: %v\n", ev.TopicPartition) } else { fmt.Printf("Delivered message to %v\n", ev.TopicPartition) } } } }() topic := "myTopic" for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} { p.Produce(&kafka.Message{ TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, Value: []byte(word), }, nil) } p.Flush(15 * 1000) }
浙公網安備 33010602011771號