1.kafka介紹 · golang

Kafka是什么 ~~~ kafka使用scala開發，支持多語言客戶端（c++、java、python、go等） Kafka最先由LinkedIn公司開發，之后成為Apache的頂級項目。 Kafka是一個分布式的、分區化、可復制提交的日志服務 LinkedIn使用Kafka實現了公司不同應用程序之間的松耦和，那么作為一個可擴展、高可靠的消息系統支持高Throughput的應用 scale out：無需停機即可擴展機器持久化：通過將數據持久化到硬盤以及replication防止數據丟失支持online和offline的場景 ~~~ Kafka的特點 Kafka是分布式的，其所有的構件borker(服務端集群)、producer(消息生產)、consumer(消息消費者)都可以是分布式的。在消息的生產時可以使用一個標識topic來區分，且可以進行分區；每一個分區都是一個順序的、不可變的消息隊列，并且可以持續的添加。同時為發布和訂閱提供高吞吐量。據了解，Kafka每秒可以生產約25萬消息（50 MB），每秒處理55萬消息（110 MB）。消息被處理的狀態是在consumer端維護，而不是由server端維護。當失敗時能自動平衡常用的場景監控：主機通過Kafka發送與系統和應用程序健康相關的指標，然后這些信息會被收集和處理從而創建監控儀表盤并發送警告。消息隊列：應用程度使用Kafka作為傳統的消息系統實現標準的隊列和消息的發布—訂閱，例如搜索和內容提要（Content Feed）。比起大多數的消息系統來說，Kafka有更好的吞吐量，內置的分區，冗余及容錯性，這讓Kafka成為了一個很好的大規模消息處理應用的解決方案。消息系統一般吞吐量相對較低，但是需要更小的端到端延時，并嘗嘗依賴于Kafka提供的強大的持久性保障。在這個領域，Kafka足以媲美傳統消息系統，如ActiveMR或RabbitMQ 站點的用戶活動追蹤: 為了更好地理解用戶行為，改善用戶體驗，將用戶查看了哪個頁面、點擊了哪些內容等信息發送到每個數據中心的Kafka集群上，并通過Hadoop進行分析、生成日常報告。流處理：保存收集流數據，以提供之后對接的Storm或其他流式計算框架進行處理。很多用戶會將那些從原始topic來的數據進行階段性處理，匯總，擴充或者以其他的方式轉換到新的topic下再繼續后面的處理。例如一個文章推薦的處理流程，可能是先從RSS數據源中抓取文章的內容，然后將其丟入一個叫做“文章”的topic中；后續操作可能是需要對這個內容進行清理，比如回復正常數據或者刪除重復數據，最后再將內容匹配的結果返還給用戶。這就在一個獨立的topic之外，產生了一系列的實時數據處理的流程。日志聚合。使用Kafka代替日志聚合（log aggregation）。日志聚合一般來說是從服務器上收集日志文件，然后放到一個集中的位置（文件服務器或HDFS）進行處理。然而Kafka忽略掉文件的細節，將其更清晰地抽象成一個個日志或事件的消息流。這就讓Kafka處理過程延遲更低，更容易支持多數據源和分布式數據處理。比起以日志為中心的系統比如Scribe或者Flume來說，Kafka提供同樣高效的性能和因為復制導致的更高的耐用性保證，以及更低的端到端延遲持久性日志：Kafka可以為一種外部的持久性日志的分布式系統提供服務。這種日志可以在節點間備份數據，并為故障節點數據回復提供一種重新同步的機制。Kafka中日志壓縮功能為這種用法提供了條件。在這種用法中，Kafka類似于Apache BookKeeper項目。 Kafka中包含以下基礎概念 1. Topic(話題)：Kafka中用于區分不同類別信息的類別名稱。由producer指定 2. Producer(生產者)：將消息發布到Kafka特定的Topic的對象(過程) 3. Consumers(消費者)：訂閱并處理特定的Topic中的消息的對象(過程) 4. Broker(Kafka服務集群)：已發布的消息保存在一組服務器中，稱之為Kafka集群。集群中的每一個服務器都是一個代理(Broker). 消費者可以訂閱一個或多個話題，并從Broker拉數據，從而消費這些已發布的消息。 5. Partition(分區)：Topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset） 6. Message：消息，是通信的基本單位，每個producer可以向一個topic（主題）發布一些消息。消息消息由一個固定大小的報頭和可變長度但不透明的字節陣列負載。報頭包含格式版本和CRC32效驗和以檢測損壞或截斷消息格式 ~~~ 1. 4 byte CRC32 of the message 2. 1 byte "magic" identifier to allow format changes, value is 0 or 1 3. 1 byte "attributes" identifier to allow annotations on the message independent of the version bit 0 ~ 2 : Compression codec 0 : no compression 1 : gzip 2 : snappy 3 : lz4 bit 3 : Timestamp type 0 : create time 1 : log append time bit 4 ~ 7 : reserved 4. (可選) 8 byte timestamp only if "magic" identifier is greater than 0 5. 4 byte key length, containing length K 6. K byte key 7. 4 byte payload length, containing length V 8. V byte payload ~~~ 安裝kafka 1）打開config目錄下的server.properties，修改log.dirs為D:\kafka_logs,修改advertised.host.name=服務器ip 2）啟動kafka ./bin/windows/kafka-server-start.bat ./config/server.preperties // 報錯：（Error: missing `server' JVM at `C:\Program Files (x86)\Java\jre1.8.0_144\binver\jvm.dll'. // Please install or use the JRE or JDK that contains these missing components.）解決方法：將C:\Program Files (x86)\Java\jre1.8.0_144\bin\servertool.exe 改名為：server.exe Mac啟動：sh bin/kafka-server-start.sh config/server.properties kafka搭建 kafka環境基于zookeeper,zookeeper環境基于JAVA-JDK。安裝JAVA-JDK ~~~ 從oracle下載最新的SDK安裝。 ~~~ 安裝zookeeper 下載zookeeper 下載地址：http://apache.fayea.com/zookeeper/ 重命名conf/zoo_sample.cfg 為conf/zoo.cfg 編輯 conf/zoo.cfg，修改dataDir=/Users/liupengjie/Desktop/tool/zookeeper-3.4.10/data （Windows：修改dataDir=D:\zookeeper-3.3.6\data\）啟動zookeeper： ./bin/zkServer.sh start ZooKeeper JMX enabled by default Using config: /Users/liupengjie/Desktop/tool/zookeeper-3.4.10/bin/../conf/zoo.cfg Starting zookeeper ... STARTED 停止ZK服務: ./bin/zkServer.sh stop 重啟ZK服務: ./bin/zkServer.sh restart 查看ZK服務狀態: ./bin/zkServer.sh status （Windows 啟動：bin/zkServer.cmd）安裝kafka 下載kafka2.1.2 打開鏈接：http://kafka.apache.org/downloads.html （下載二進制包） kafka配置通常情況下需要在解壓縮kafka后，修改config/server.properties 配置文件中的以下項 ~~~ log.dirs = kafka-logs advertised.host.name=192.168.1.125 (本機ip) # root directory for all kafka znodes. zookeeper.connect = localhost:9092 # Timeout in ms for connecting to zookeeper zookeeper.connection.timeout.ms=6000 listeners = PLAINTEXT://ip:9092 (listeners = PLAINTEXT://your.host.name:9092) /* log.dirs=/Users/liupengjie/Desktop/tool/kafka_2.11-0.11.0.0/kafka_logs zookeeper.connect=localhost:2181 zookeeper.connection.timeout.ms=6000 #listeners=PLAINTEXT://:9092 advertised.host.name=192.168.1.125 */ ~~~ log.dirs 指的是kafka的log Data保存的目錄，默認為Null。如果不指定log Data會保存到log.dir設置的目錄中，log.dir默認為/tmp/kafka-logs。需要保證啟動KafKaServer的用戶對log.dirs或log.dir設置的目錄具有讀與寫的權限。 zookeeper.connect 指的是zookeeper集群的地址，可以是多個，多個之間用逗號分割hostname1:port1,hostname2:port2,hostname3:port3 listeners 監聽列表(以逗號分隔不同的協議(如plaintext,trace,ssl、不同的IP和端口)) 啟動kafka ~~~ cd bin ./kafka-server-start.sh ../config/server.properties (Windows啟動：./bin/windows/kafka-server-start.bat ./config/server.preperties) /* Windows啟動報錯： Error: missing `server' JVM at `C:\Program Files (x86)\Java\jre1.8.0_144\binver\jvm.dll'. Please install or use the JRE or JDK that contains these missing components. 解決方法：將C:\Program Files (x86)\Java\jre1.8.0_144\bin\servertool.exe 改名為：server.exe */ ~~~ kafka鏈接zookeeper ~~~ cd bin $ kafka-console-consumer.sh --topic nginx_log --zookeeper 127.0.0.1 2181 (Windows 執行：kafka-console-consumer.bat --topic nginx_log --zookeeper 127.0.0.1 2181) /* Windows啟動報錯： Error: missing `server' JVM at `C:\Program Files (x86)\Java\jre1.8.0_144\binver\jvm.dll'. Please install or use the JRE or JDK that contains these missing components. 解決方法：將C:\Program Files (x86)\Java\jre1.8.0_144\bin\servertool.exe 改名為：server.exe */ ~~~ 查看topic的詳細信息 ~~~ bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic nginx_log --describe ~~~ kafka消費者客戶端命令 ~~~ bin/kafka-console-consumer.sh --zookeeper 127.0.0.1:2181 --topic nginx_log --from-beginning ~~~ Kafka常用命令以下是kafka常用命令行總結： 1.查看topic的詳細信息 ./kafka-topics.sh -zookeeper 127.0.0.1:2181 -describe -topic testKJ1 2、為topic增加副本 ./kafka-reassign-partitions.sh -zookeeper 127.0.0.1:2181 -reassignment-json-file json/partitions-to-move.json -execute 3、創建topic ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testKJ1 4、為topic增加partition ./bin/kafka-topics.sh –zookeeper 127.0.0.1:2181 –alter –partitions 20 –topic testKJ1 5、kafka生產者客戶端命令 ./kafka-console-producer.sh --broker-list localhost:9092 --topic testKJ1 6、kafka消費者客戶端命令 ./kafka-console-consumer.sh -zookeeper localhost:2181 --from-beginning --topic testKJ1 7、kafka服務啟動 ./kafka-server-start.sh -daemon ../config/server.properties 8、下線broker ./kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 127.0.0.1:2181 --broker #brokerId# --num.retries 3 --retry.interval.ms 60 shutdown broker 9、刪除topic ./kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic testKJ1 --zookeeper 127.0.0.1:2181 ./kafka-topics.sh --zookeeper localhost:2181 --delete --topic testKJ1 10、查看consumer組內消費的offset ./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper localhost:2181 --group test --topic testKJ1