碼上快樂

相關內容    簡體    繁體

kafka安裝及配置

本文轉載自   查看原文   2020-01-17 17:59   12451    kafka/ kafka安裝及配置

一、下載安裝

1、下載地址

https://www.apache.org/dyn/closer.cgi?path=/kafka/2.4.0/kafka_2.13-2.4.0.tgz

下載命令：wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.13-2.4.0.tgz

2、解壓

解壓命令：tar –zxvf  kafka_2.13-2.4.0.tgz

移動： cp -R kafka_2.13-2.4.0 /zjl/program/

創建軟鏈接： ln -s kafka_2.13-2.4.0/  /zjl/program/kafka

解壓后即可使用

配置zookeeper集群參見：https://www.cnblogs.com/happydreamzjl/articles/12186171.html

二、配置kafka

1、配置server.properties

位置:  /kafka/config

默認配置 advertised.listeners=PLAINTEXT://your.host.name:9092

修改為 advertised.listeners=PLAINTEXT://ip:9092

例：advertised.listeners=PLAINTEXT://192.168.244.128:9092

  1. ip為服務器ip
  2. hostname和端口是用來建議給生產者和消費者使用的，如果沒有設置，將會使用listeners的配置，如果listeners也沒有配置，將使用java.net.InetAddress.getCanonicalHostName()來獲取這個hostname和port，對於ipv4，基本就是localhost了。
  3. "PLAINTEXT"表示協議，可選的值有PLAINTEXT和SSL，hostname可以指定IP地址，也可以用"0.0.0.0"表示對所有的網絡接口有效，如果hostname為空表示只對默認的網絡接口有效。也就是說如果你沒有配置advertised.listeners，就使用listeners的配置通告給消息的生產者和消費者，這個過程是在生產者和消費者獲取源數據(metadata)。

主要注意三個配置：

  1. broker.id標識本機
  2. log.dirs是kafka接收消息存放路徑
  3. zookeeper.connect指定連接的zookeeper集群地址

2、kafka集群計划

三台機器：192.168.244.128    192.168.244.130   192.168.244.131

修改server.properties文件，只修改三項，其他根據需要配置

192.168.244.128

broker.id=1
log.dirs=/zjl/program/kafka/datalogs/kafka-logs
zookeeper.connect=192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181

192.168.244.130

broker.id=2
log.dirs=/zjl/program/kafka/datalogs/kafka-logs
zookeeper.connect=192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181

192.168.244.131

broker.id=3
log.dirs=/zjl/program/kafka/datalogs/kafka-logs
zookeeper.connect=192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181

3、啟動

三個節點分別執行：nohup ./kafka-server-start.sh -daemon ../config/server.properties &

啟動時若報錯java.net.UnknownHostException  XXX

命令查看:hostname  打印出本機hostname 我的機器是  zjlLinux1

修改文件 ：vim /etc/hosts

添加內容  192.168.244.128  zjlLinux1  #ip是本機IP 后邊是hostname

4、較驗啟動是否成功

  1. jps
  2. ps –ef|grep kafka

5、kafka-manager

下載地址：https://github.com/yahoo/kafka-manager/releases

https://github.com/yahoo/CMAK/releases

編譯

tar -zxvf 1.3.3.17.tar.gz

cd kafka-manager-1.3.3.17/

./sbt clean dist

配置

在解壓后的conf目錄中打開 application.conf文件，修改其中的配置信息，最主要的內容為：kafka-manager.zkhosts="192.168.1.22:2181,192.168.1.23:2181,192.168.1.24:2181"  配置為Kafka的 zookeeper 服務器。你還可以通過環境變量ZK_HOSTS配置這個參數值。

啟動

在解壓的目錄中，使用下面的命令啟動Kafka-manager ：    bin/kafka-manager

默認情況下端口為9000，你還可以通過下面的命令指定配置文件和端口： bin/kafka-manager -Dconfig.file=/path/to/application.conf -Dhttp.port=9000

啟動后，從瀏覽器打開：http://[服務器IP]:9000/  就可以看到配置頁面了

三、kafka基本命令操作

1、創建TOPIC

./kafka-topics.sh --create --zookeeper 192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181  --replication-factor 3 --partitions 6 --topic kfk_test

2、列出創建的topic

./kafka-topics.sh --list --zookeeper 192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181 

3、生成數據

注意這里的端口號是kafka的端口  9092

./kafka-console-producer.sh -broker-list 192.168.244.128:9092,192.168.244.130:9092,192.168.244.131:9092   --topic kfk_test

image

4、消費生產數據

kafka 0.9版本之前用zookeeper
./kafka-console-consumer.sh --zookeeper 192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181  --from-beginning --topic kfk_test
 
kafka 0.9版本之后不推薦zookeeper方式,仍然支持，但逐漸會被取消，推薦bootstrap-server方式

注意這里的端口號是kafka的端口  9092
./kafka-console-consumer.sh --bootstrap-server  192.168.244.128:9092,192.168.244.130:9092,192.168.244.131:9092  --from-beginning --topic kfk_test

image

5、查看指定topic信息

./kafka-topics.sh --describe --zookeeper 192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181 --topic kfk_test

6、刪除kafka中的topic

./kafka-topics.sh --delete --zookeeper 192.168.244.128:2181,192.168.244.130:2181,192.168.244.131:2181 --topic kfk_test

刪除zookeeper中的topic信息

  1. rmr /brokers/topics/kfk_test

  2. rmr /config/topics/kfk_test

  3. rmr /admin/delete_topics/kfk_test

  4. rmr /consumers/kfk_test-group

 刪除topic數據相關的目錄
rm -rf /var/local/kafka/data/kfk_test*

四、配置文件詳解

  1. broker.id = 0 每一個broker在集群中的唯一表示，要求是正數。當該服務器的IP地址發生改變時，broker.id沒有變化，則不會影響consumers的消息情況
  2. log.dirs=/data/kafka-logs   kafka數據的存放地址，多個地址的話用逗號分割 /data/kafka-logs-1，/data/kafka-logs-2
  3. port =9092   broker server服務端口
  4. message.max.bytes =6525000  表示消息體的最大大小，單位是字節
  5. num.network.threads =4  broker處理消息的最大線程數，一般情況下不需要去修改
  6. num.io.threads =8  broker處理磁盤IO的線程數，數值應該大於你的硬盤數
  7. background.threads =4  一些后台任務處理的線程數，例如過期消息文件的刪除等，一般情況下不需要去做修改
  8. queued.max.requests =500  等待IO線程處理的請求隊列最大數，若是等待IO的請求超過這個數值，那么會停止接受外部消息，應該是一種自我保護機制。
  9. host.name  broker的主機地址，若是設置了，那么會綁定到這個地址上，若是沒有，會綁定到所有的接口上，並將其中之一發送到ZK，一般不設置
  10. socket.send.buffer.bytes=100*1024  socket的發送緩沖區，socket的調優參數SO_SNDBUFF
  11. socket.receive.buffer.bytes =100*1024  socket的接受緩沖區，socket的調優參數SO_RCVBUFF
  12. socket.request.max.bytes =100*1024*1024  socket請求的最大數值，防止serverOOM，message.max.bytes必然要小於socket.request.max.bytes，會被topic創建時的指定參數覆蓋
  13. log.segment.bytes =1024*1024*1024  topic的分區是以一堆segment文件存儲的，這個控制每個segment的大小，會被topic創建時的指定參數覆蓋
  14. log.roll.hours =24*7  這個參數會在日志segment沒有達到log.segment.bytes設置的大小，也會強制新建一個segment會被 topic創建時的指定參數覆蓋
  15. log.cleanup.policy = delete  日志清理策略選擇有：delete和compact主要針對過期數據的處理，或是日志文件達到限制的額度，會被 topic創建時的指定參數覆蓋
  16. log.retention.minutes=60*24 # 一天后刪除  數據存儲的最大時間超過這個時間會根據log.cleanup.policy設置的策略處理數據，也就是消費端能夠多久去消費數據; log.retention.bytes和log.retention.minutes任意一個達到要求，都會執行刪除，會被topic創建時的指定參數覆蓋
  17. log.retention.bytes=-1   topic每個分區的最大文件大小，一個topic的大小限制 = 分區數*log.retention.bytes= -1沒有大小限制；log.retention.bytes和log.retention.minutes任意一個達到要求，都會執行刪除，會被topic創建時的指定參數覆蓋
  18. log.retention.check.interval.ms=5minutes  文件大小檢查的周期時間，是否處罰 log.cleanup.policy中設置的策略
  19. log.cleaner.enable=false  是否開啟日志壓縮
  20. log.cleaner.threads = 2  日志壓縮運行的線程數
  21. log.cleaner.io.max.bytes.per.second=None   日志壓縮時候處理的最大大小
  22. log.cleaner.dedupe.buffer.size=500*1024*1024  日志壓縮去重時候的緩存空間，在空間允許的情況下，越大越好
  23. log.cleaner.io.buffer.size=512*1024  日志清理時候用到的IO塊大小一般不需要修改
  24. log.cleaner.io.buffer.load.factor =0.9  日志清理中hash表的擴大因子一般不需要修改
  25. log.cleaner.backoff.ms =15000  檢查是否處罰日志清理的間隔
  26. log.cleaner.min.cleanable.ratio=0.5  日志清理的頻率控制，越大意味着更高效的清理，同時會存在一些空間上的浪費，會被topic創建時的指定參數覆蓋
  27. log.cleaner.delete.retention.ms =1day   對於壓縮的日志保留的最長時間，也是客戶端消費消息的最長時間，同log.retention.minutes的區別在於一個控制未壓縮數據，一個控制壓縮后的數據。會被topic創建時的指定參數覆蓋
  28. log.index.size.max.bytes =10*1024*1024  對於segment日志的索引文件大小限制，會被topic創建時的指定參數覆蓋
  29. log.index.interval.bytes =4096  當執行一個fetch操作后，需要一定的空間來掃描最近的offset大小，設置越大，代表掃描速度越快，但是也更好內存，一般情況下不需要搭理這個參數
  30. log.flush.interval.messages=None  log文件”sync”到磁盤之前累積的消息條數,因為磁盤IO操作是一個慢操作,但又是一個”數據可靠性"的必要手段,所以此參數的設置,需要在"數據可靠性"與"性能"之間做必要的權衡.如果此值過大,將會導致每次"fsync"的時間較長(IO阻塞),如果此值過小,將會導致"fsync"的次數較多,這也意味着整體的client請求有一定的延遲.物理server故障,將會導致沒有fsync的消息丟失.
  31. log.flush.scheduler.interval.ms =3000  檢查是否需要固化到硬盤的時間間隔
  32. log.flush.interval.ms = None  僅僅通過interval來控制消息的磁盤寫入時機,是不足的.此參數用於控制"fsync"的時間間隔,如果消息量始終沒有達到閥值,但是離上一次磁盤同步的時間間隔達到閥值,也將觸發.
  33. log.delete.delay.ms =60000  文件在索引中清除后保留的時間一般不需要去修改
  34. log.flush.offset.checkpoint.interval.ms =60000  控制上次固化硬盤的時間點，以便於數據恢復一般不需要去修改
  35. auto.create.topics.enable =true 是否允許自動創建topic，若是false，就需要通過命令創建topic
  36. default.replication.factor =1  創建replica副本因子，即設置該partition有幾個副本。
  37. num.partitions =1  每個topic的分區個數，若是在topic創建時候沒有指定的話會被topic創建時的指定參數覆蓋

 

以下是kafka中Leader,replicas配置參數

  1. controller.socket.timeout.ms =30000  partition leader與replicas之間通訊時,socket的超時時間
  2. controller.message.queue.size=10  partition leader與replicas數據同步時,消息的隊列尺寸
  3. replica.lag.time.max.ms =10000  replicas響應partition leader的最長等待時間，若是超過這個時間，就將replicas列入ISR(in-sync replicas)，並認為它是死的，不會再加入管理中
  4. replica.lag.max.messages =4000  如果follower落后與leader太多,將會認為此follower[或者說partition relicas]已經失效（##通常,在follower與leader通訊時,因為網絡延遲或者鏈接斷開,總會導致replicas中消息同步滯后，如果消息之后太多,leader將認為此follower網絡延遲較大或者消息吞吐能力有限,將會把此replicas遷移到其他follower中。在broker數量較少,或者網絡不足的環境中,建議提高此值.）
  5. replica.socket.timeout.ms=30*1000  follower與leader之間的socket超時時間
  6. replica.socket.receive.buffer.bytes=64*1024  leader復制時候的socket緩存大小
  7. replica.fetch.max.bytes =1024*1024  replicas每次獲取數據的最大大小
  8. replica.fetch.wait.max.ms =500  replicas同leader之間通信的最大等待時間，失敗了會重試
  9. replica.fetch.min.bytes =1  fetch的最小數據尺寸,如果leader中尚未同步的數據不足此值,將會阻塞,直到滿足條件
  10. num.replica.fetchers=1  leader進行復制的線程數，增大這個數值會增加follower的IO
  11. replica.high.watermark.checkpoint.interval.ms =5000  每個replica檢查是否將最高水位進行固化的頻率
  12. controlled.shutdown.enable =false  是否允許控制器關閉broker ,若是設置為true,會關閉所有在這個broker上的leader，並轉移到其他broker
  13. controlled.shutdown.max.retries =3  控制器關閉的嘗試次數
  14. controlled.shutdown.retry.backoff.ms =5000  每次關閉嘗試的時間間隔
  15. leader.imbalance.per.broker.percentage =10  leader的不平衡比例，若是超過這個數值，會對分區進行重新的平衡
  16. leader.imbalance.check.interval.seconds =300  檢查leader是否不平衡的時間間隔
  17. offset.metadata.max.bytes   客戶端保留offset信息的最大空間大小

 

kafka中zookeeper參數配置

  1. zookeeper.connect = localhost:2181  zookeeper集群的地址，可以是多個，多個之間用逗號分割 hostname1:port1,hostname2:port2,hostname3:port3
  2. zookeeper.session.timeout.ms=6000  ZooKeeper的最大超時時間，就是心跳的間隔，若是沒有反映，那么認為已經死了，不易過大
  3. zookeeper.connection.timeout.ms =6000  ZooKeeper的連接超時時間
  4. zookeeper.sync.time.ms =2000  zookeeper的follower同leader的同步時間

整理自：
https://www.cnblogs.com/toutou/p/linux_install_kafka.html

https://baijiahao.baidu.com/s?id=1598139489983645370&wfr=spider&for=pc

https://blog.csdn.net/gangsijay888/article/details/94630763


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 kafka的認識、安裝與配置 KAFKA集群安裝和配置 kafka window安裝與配置 kafka安裝以及配置詳解 Linux——安裝並配置Kafka Kafka SSL安裝與配置 Kafka安裝配置 kafka安裝及配置過程 kafka 安裝與配置 kafka集群的安裝和配置
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM