一、Kafka是一種高吞吐量的分布式發布訂閱消息系統，它可以處理消費者規模的網站中的所有動作流數據。
Kafka場景比喻
接下來我大概比喻下Kafka的使用場景
消息中間件：生產者和消費者
媽媽：生產者
你：消費者
饅頭：數據流、消息
正常情況下： 生產一個 消費一個
其他情況：
1）一直生產，你吃到某一個饅頭時，你卡主(機器故障)， 饅頭就丟失了
2）一直生產，做饅頭速度快，你吃來不及，饅頭也就丟失了
為了放着其他生產情況的出現，我們可以拿個碗/籃子，饅頭做好以后先放到籃子里，你要吃的時候去籃子里面取出來吃，而這籃子/框就可以為：Kafka。當籃子滿了，饅頭就裝不下了，咋辦？ 多准備幾個籃子 === Kafka的擴容
二、Kafka的架構和核心概念
這是張我在Kafka官網上截的圖，我大概可以把Kafka的主要結構分為以下幾點：
producer：生產者，就是生產饅頭(老媽)
consumer：消費者，就是吃饅頭的(你)
broker：籃子
topic：主題，給饅頭帶一個標簽，topica的饅頭是給你吃的，topicb的饅頭是給你弟弟吃
1.Kafka可以作為集群運行在一台或者多個服務器上面；
2.Kafka集群可以分類地存儲記錄流，以打標簽的方式，就是采用topics，每個broker可以打個topic，這樣能保證消費者可以根據topic選擇性消費；
3.每個記錄由Key、Value、timestamp構成。
Kafka四個核心的API
1.ProducerAPI：允許一個應用向一個或多個topic里發布記錄流；
2.ConsumerAPI：允許一個應用訂閱一個或多個topics，處理topic里的數據流，就相當於消費；
3.StreamAPI：允許應用扮演流處理的作用，從一個或多個topic里消費數據流，然后產生輸出流數據到其他一個或多個topic里，對輸入流數據有效傳輸到輸出口；
4.ConnectorAPI：允許運行和構建一個可重復利用的生產者和消費者，能將kafka的topic與其他存在的應用和數據庫設備相連接，比如鏈接一個實時數據庫，可以捕捉到每張表的變化。
這四個API，主要應用在IDEA上對應用程序的開發中，通過代碼的形式管理Kafka。在第四部分將會對前兩個API寫個簡單DEMO演示。
三、Kafka的快速使用
Kafka使用到了zookeeper，所以首先你得安裝zookeeper再安裝kafka。
1.單節點的broker部署
首先我們需要修改$KAFKA_HOME/config/server.properties這個配置文件，主要以下幾處需要修改：
broker.id=0，每個broker的ID需要唯一
listeners：監聽的端口（此處筆者設置的是默認端口9092）
host.name：當前機器
log.dirs：存儲日志的文件夾
num.partitions：分區的數量
zookeeper.connect：zookeeper的地址（默認為localhost：2181）
這幾處根據你自身需要進行配置，然后啟動步驟如下：
1）開啟zookeeper，此處需要注意的是zookeeper的conf目錄下的zoo.cfg配置文件，主要修改的也是日志存儲目錄那塊。
2）啟動Kafka，命令為：kafka-server-start.sh $KAFKA_HOME/config/server.properties
3）創建topic，需要指定zookeeper，命令為：kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic hello_topic。 注意指定zookeeper，后面幾個屬性可以根據你實際情況進行定義。另外查看所有topic的命令為：
kafka-topics.sh --list --zookeeper hadoop000:2181
4）發送消息，需要指定broker，命令為：kafka-console-producer.sh --broker-list hadoop000:9092 --topic hello_topic
5）消費消息，需要指定zookeeper，命令為：kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic hello_topic --from-beginning。意思就是指定zookeeper上的topic進行消費，from-beginning的設置，可以查看之前的消息。
2.單節點，多broker
主要是增加多個server.properties文件，一個配置文件就相當於一個broker，我就設置三個broker：
server-1.properties
log.dirs=/home/hadoop/app/tmp/kafka-logs-1
listeners=PLAINTEXT://:9093
broker.id=1
server-2.properties
log.dirs=/home/hadoop/app/tmp/kafka-logs-2
listeners=PLAINTEXT://:9094
broker.id=2
server-3.properties
log.dirs=/home/hadoop/app/tmp/kafka-logs-3
listeners=PLAINTEXT://:9095
broker.id=3
然后依次開啟，命令如下：
kafka-server-start.sh -daemon $KAFKA_HOME/config/server-1.properties &
接下來就跟上面的步驟一樣：
kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic my-replicated-topic
查看 topic的詳細信息：
kafka-topics.sh --describe --zookeeper hadoop000:2181 --topic my-replicated-topic
要注意的是，副本中會有個leader，而多副本也實現了kafka的容錯性，掛掉一個副本后，會自動在剩下副本里選出一個leader來同步操作。
根據上面步驟操作，我們在producer窗口輸入，在consumer消費窗口看到相應輸出。
四、Producer和Consumer API的使用
接下來展示一個簡單的Demo，在生產端簡單創建個線程進行循環輸出，然后用消費者端對輸出的內容進行展示，也就是消費。
配置文件
/**
public class KafkaProperties {
public static final String ZK = "192.168.199.111:2181";
public static final String TOPIC = "hello_topic";
public static final String BROKER_LIST = "192.168.199.111:9092";
public static final String GROUP_ID = "test_group1";
}
Producer API DEMO
import java.util.Properties;
/**
* Kafka生產者
*/
public class KafkaProducer extends Thread{
private String topic;
private Producer<Integer, String> producer;
public KafkaProducer(String topic) {
this.topic = topic;
Properties properties = new Properties();
properties.put("metadata.broker.list",KafkaProperties.BROKER_LIST);
properties.put("serializer.class","kafka.serializer.StringEncoder");
properties.put("request.required.acks","1");
producer = new Producer<Integer, String>(new ProducerConfig(properties));
}
public void run() {
int messageNo = 1;
while(true) {
String message = "message_" + messageNo;
producer.send(new KeyedMessage<Integer, String>(topic, message));
System.out.println("Sent: " + message);
messageNo ++ ;
try{
Thread.sleep(2000);
} catch (Exception e){
e.printStackTrace();
}
}
}
}
Consumer API DEMO
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
/**
* Kafka消費者
*/
public class KafkaConsumer extends Thread{
private String topic;
public KafkaConsumer(String topic) {
this.topic = topic;
}
private ConsumerConnector createConnector(){
Properties properties = new Properties();
properties.put("zookeeper.connect", KafkaProperties.ZK);
properties.put("group.id",KafkaProperties.GROUP_ID);
return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
}
public void run() {
ConsumerConnector consumer = createConnector();
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, 1);
// topicCountMap.put(topic2, 1);
// topicCountMap.put(topic3, 1);
// String: topic
// List<KafkaStream<byte[], byte[]>> 對應的數據流
Map<String, List<KafkaStream<byte[], byte[]>>> messageStream = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = messageStream.get(topic).get(0); //獲取我們每次接收到的暑假
ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
while (iterator.hasNext()) {
String message = new String(iterator.next().message());
System.out.println("rec: " + message);
}
}
}
最后在main函數對這兩個類調用即可，結果如下：
五、搭建kafka集群
1)下載kafka0.8(http://kafka.apache.org/downloads.html),保存到服務器/home/wwb目錄下kafka-0.8.0-beta1-src.tgz(kafka_2.8.0-0.8.0-beta1.tgz)
2)解壓 tar -zxvf kafka-0.8.0-beta1-src.tgz,產生文件夾kafka-0.8.0-beta1-src更改為kafka01
3)配置
修改kafka01/config/server.properties,其中broker.id,log.dirs,zookeeper.connect必須根據實際情況進行修改，其他項根據需要自行斟酌。大致如下：
broker.id=1
port=9091
num.network.threads=2
num.io.threads=2
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dir=./logs
num.partitions=2
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=536870912
num.replica.fetchers=2
log.cleanup.interval.mins=10
zookeeper.connect=192.168.0.1:2181,192.168.0.2:2182,192.168.0.3:2183
zookeeper.connection.timeout.ms=1000000
kafka.metrics.polling.interval.secs=5
kafka.metrics.reporters=kafka.metrics.KafkaCSVMetricsReporter
kafka.csv.metrics.dir=/tmp/kafka_metrics
kafka.csv.metrics.reporter.enabled=false
4）初始化因為kafka用scala語言編寫，因此運行kafka需要首先准備scala相關環境。
> cd kafka01
> ./sbt update
> ./sbt package
> ./sbt assembly-package-dependency
在第二個命令時可能需要一定時間，由於要下載更新一些依賴包。所以請大家 耐心點。
5) 啟動kafka01
>JMX_PORT=9997 bin/kafka-server-start.sh config/server.properties &
a)kafka02操作步驟與kafka01雷同，不同的地方如下
修改kafka02/config/server.properties
broker.id=2
port=9092
##其他配置和kafka-0保持一致
啟動kafka02
JMX_PORT=9998 bin/kafka-server-start.shconfig/server.properties &
b)kafka03操作步驟與kafka01雷同，不同的地方如下
修改kafka03/config/server.properties
broker.id=3
port=9093
##其他配置和kafka-0保持一致
啟動kafka02
JMX_PORT=9999 bin/kafka-server-start.shconfig/server.properties &
6)創建Topic(包含一個分區，三個副本)
>bin/kafka-create-topic.sh--zookeeper 192.168.0.1:2181 --replica 3 --partition 1 --topicmy-replicated-topic
7)查看topic情況
>bin/kafka-list-top.sh --zookeeper 192.168.0.1:2181
topic: my-replicated-topic partition: 0 leader: 1 replicas: 1,2,0 isr: 1,2,0
8)創建發送者
>bin/kafka-console-producer.sh--broker-list 192.168.0.1:9091 --topic my-replicated-topic
my test message1
my test message2
^C
9)創建消費者
>bin/kafka-console-consumer.sh --zookeeper127.0.0.1:2181 --from-beginning --topic my-replicated-topic
...
my test message1
my test message2
^C
10)殺掉server1上的broker
>pkill -9 -f config/server.properties
11)查看topic
>bin/kafka-list-top.sh --zookeeper192.168.0.1:2181
topic: my-replicated-topic partition: 0 leader: 1 replicas: 1,2,0 isr: 1,2,0
發現topic還正常的存在
11）創建消費者，看是否能查詢到消息
>bin/kafka-console-consumer.sh --zookeeper192.168.0.1:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C
說明一切都是正常的。
OK,以上就是對Kafka個人的理解，不對之處請大家及時指出。
補充說明：
1、public Map<String, List<KafkaStream<byte[], byte[]>>> createMessageStreams(Map<String, Integer> topicCountMap)，其中該方法的參數Map的key為topic名稱，value為topic對應的分區數，譬如說如果在kafka中不存在相應的topic時，則會創建一個topic，分區數為value，如果存在的話，該處的value則不起什么作用
2、關於生產者向指定的分區發送數據，通過設置partitioner.class的屬性來指定向那個分區發送數據，如果自己指定必須編寫相應的程序，默認是kafka.producer.DefaultPartitioner,分區程序是基於散列的鍵。
3、在多個消費者讀取同一個topic的數據，為了保證每個消費者讀取數據的唯一性，必須將這些消費者group_id定義為同一個值，這樣就構建了一個類似隊列的數據結構，如果定義不同，則類似一種廣播結構的。
4、在consumerapi中，參數設計到數字部分，類似Map<String,Integer>,
numStream,指的都是在topic不存在的時，會創建一個topic，並且分區個數為Integer,numStream,注意如果數字大於broker的配置中num.partitions屬性，會以num.partitions為依據創建分區個數的。
5、producerapi，調用send時，如果不存在topic，也會創建topic，在該方法中沒有提供分區個數的參數，在這里分區個數是由服務端broker的配置中num.partitions屬性決定的
關於kafka說明可以參考：http://kafka.apache.org/documentation.html