KafkaKafkaScalaJMS（JavaMessageService）規范的同學來說，消息系統已經不是什么新概念了(例如ActiveMQ，RabbitMQ等)。Kafka這樣來說，Kafka借鑒了JMS規范的思想，但是確并沒有完全遵循JMS規kafka是一個分布式的，分區的消息(稱之為commitlog)服務。它提供一Topic:Kafka按照Topic分類來消Producer：發布(publish)消息到Topic的進程稱之為生產Consumer：訂閱(subscribe)Topic并且處理Topic中消息的進程BrokerKafka因此，從一個較高的層面上來看，producersKafkaconsumers服務端(brokers)和客戶端(producer、consumer)TCP成。我們為Kafka提供了一個Java客戶端，但是也可以使用其他語言編寫的Topic讓我們首先深入理解Kafka提出一個次的抽象概念-TopicTopicmessageTopic于每一個Topic，kafka集群按照如下方式一個分區(Partition,可以將消息就理解為一個隊列Queue)日志文件:partitionmessagemessagecommitlogpartition為offset，用來唯一標示某個分區中的message。partition，commit-logpartitionmessage的offset都是唯一的，但是不同的partitionmessage的offsetafka集，在配置的時間范圍內，所有的由produer生成的消息，而不管這些消息有沒有被消費。例如日志保留(geteon2天。afka會最近2天生產的所有消息，而2天前的消息會被丟棄。的性能與保留的數據量的大小沒有關系，因此保存大量的數據(日志信息)不會有什么影響。每個consumer是基于自己在commitlog中的消費進度(offset)來進行工作的。在kafka中，offset由consumer來：一般情況下我們按照順序逐條消費commitlog中的消息，當然我可以通過指定offset來重復消費某些消kafkaconsumerconsumer，consumer個consumer各自的offset。log（partitioned），log統文件系統的限制時，可以自動拆分。每個partition對應的log都受到所在機器的文件系統大小的限制，但是一個Topic中是可以有很多分區的，因此可logpartitionskafkabrokerbrokerbrokerpartitionkafkapartitionpartitionbrokerleader”的作用，0的broker作為“follwers”的作用。leader處理所有的針對這個partition的讀寫請求，而followersleader的結果。如果這個leader失效followerleader。broker所管理的partition的leader，同時又是其他broker所管理partitions的followers，kafka通過這種方式來達到負載均衡。topicmessagetopic一個partition中。通過round-robin做簡單的負載均衡。也可以根據消息中傳統的消息傳遞模式有2種：隊列(queuing)和（publish-subscribe）。在queuing模式中，多個consumer從服務器中數據，消息只會到達一個consumer。在publish-subscribe模型中，消息會被廣播給所有的consumer。Kafka基于這2種模式提供了一種consumer的抽象概念：consumergroup。consumerconsumergrouptopicmessagemessage會被傳遞到consumergroup中的一個consumer實例。consumer實例可以運行在不同的進程上，也可以在不同的物理機器consumerconsumergroup的queue模式，并在眾多的consumerinstance之間進行負載均衡。consumerconsumergroup，這就類似于傳統的publish-subscribe模型。更一般的情況是，通常一個topic會有幾個consumergroup，每個consumergroup（logicalsubscriber）。每個consumergroup由多個consumerinstance組成，從而達到可擴展和容災的功能。這并沒有什么特殊的地方，僅僅是將publish-subscribe模型中的運行在單個進程上的consumers中的consumer替換成一個consumer2brokerkafka4Parition(P0-P3)。這個2ConsumerGroup，A2consumerinstancesBKafka比傳統的消息系統有著更強的順序保證。在傳統的情況下，服務器按照consumerconsumerconsumer取到消息時間可能比后消費的consumer獲取到消息的時間長，導致不能保證順序性。這表明，當進行并行的消費的時候，消息在多個consumer之間可能會失去順序性。消息系統通常會采取一種“exclusiveconsumer”的概念，來確保同一時間內只有一個consumer能夠從隊列中進行消費，但是這實際上Kafka在這方面做的更好。通過Topic中并行度的概念，即partition，Kafka可以同時提供順序性保證和多個consumer同時消費時的負載均衡。實現的原理是通過將一個topic中的partition分配給一個consumergroup中的不同consumerinstancepartitionconsumerinstancetopic多個partition，但是一個consumergroup中同時也有多個consumerconsumergroup中的consumerinstance的數量不能比一個Topic中的partition的數量多。Kafka只在partition的范圍內保證消息消費的局部順序性，不能在同一個topicpartitiontopicpartition1consumerconsumerinstance從較高的層面上來說的話，Kafka發送到一個Topic中的message會按照發送的順序添加到commitlog中。M1，M2producerM1M2話，那么在commitlog中，M1的offset就會commit2的offset小。consumercommitlogtopicreplicationfactorN，Kafka以N-1一個服務器，而在commitlog中的消息不會丟失。kafkaKafkaScalaJVMKafka需要先安裝JDK。#yuminstalljava-1.8.0-openjdk*-kafka依賴zookeeper#wgethttp://m #tar-zxvfzookeeper-3.4.12.tar.gz#cdzookeeper-3.4.12#cpconf/zoo_sample.cfg#bin/zkServer.shstartconf/zoo.cfg&#bin/zkCli.sh#ls #查看zk的 相關節第一步：安裝1.1.0release##tar-xzfkafka_2.11-1.1.0.tgz#cdkafka_2.11-1.1.0

kafka啟動語法：kafka-server-start.sh[-daemon]perties以進程運行，否則ssh客戶端退出后，就會停止服務。(注意，在啟動kafka時會使用linux主機名關聯的ip地址，所以需要把主機名配置到本地host里，用vim/etc/hosts)#bin/kafka-server-start.sh-daemon我們進入 通過zookeeper客戶端查看下zookeeper ##ls #查看zk的 kafka相關節ls/brokers/idskafka現在我們來創建一個名字為“testTopic，topicpartition#bin/kafka-topics.sh--create--zookeeperlocalhost:2181--replication-factor1--partitions1--topictestkafkatopic#bin/kafka-topics.sh--list--zookeeperlocalhost:2181Topicbroker，producer布一個消息某個指定的Topic，但是這個Topic并不存在時，就自動創建。kafka自帶了一個producer命令客戶端，可以從本地文件中內容，或者kafka集群中。在默認情況下，每一個行會被當做成一個獨立的消息。首先我們要運行發布消息的，然后在命令中輸入要發送的消息的內容#bin/kafka-console-producer.sh--broker-listlocalhost:9092--topic>thisisa>thisisaanotherconsumer，kafka#bin/kafka-console-consumer.sh--zookeeperlocalhost:2181--topictest--from-beginning如果你是通過不同的終端窗口來運行以上令，你將會看到在producer終端輸入的內容，很快就會在consumer的終端窗口上顯示出來。第六步：kafkabroker，這并沒有什么意思。對于kafkabrokerkafkakafkabroker理解，現在我們在一臺機器上同時啟動三個broker實例。2brokercpconfig/pertiesconfig/pertiescpconfig/pertiesconfig/perties broker.id屬性在kafka集群中必須要是唯一的。我們需要重新指定port和 consumer只能獲取到一個instance實例的信息，或者是相互之間的數據會被zookeeperbroker只需要在啟動2個broker實例即可：bin/kafka-server-start.shconfig/perties&bin/kafka-server-start.shconfig/perties&現在我們創建一個新的topic，備份因子設置為3：#bin/kafka-topics.sh--create--zookeeperlocalhost:2181--replication-factor3--partitions1--topicmy-replicated-topic3topic，但是到底是brokertopic同時只有一個broker在處理這個topic)？#bin/kafka-topics.sh--describe--zookeeperlocalhost:2181--topicmy-partitionpartitionpartitionleaderpartitiontopicpartitionspartitionleaderreplicaspartitionbrokerisrreplicaspartition現在我們的案例中，0perties我們可以運行相同令查看之前創建的名稱為”test“的#bin/kafka-topics.sh--describe--zookeeperlocalhost:2181--topictopicpartition1，備份因子為1，因此顯示就如上所示了。topic#bin/kafka-console-producer.sh--broker-listlocalhost:9092--topic>mytestmsg>mytestmsg#bin/kafka-console-consumer.sh--zookeeperlocalhost:2181--from-beginning--topicmy-replicated-topicmytestmsg1mytestmsg現在我們來測試我們容錯性broker0leader