您好,登錄后才能下訂單哦!
1、簡介
Kafka is a distributed,partitioned,replicated commit logservice。它提供了類似于JMS的特性,但是在設(shè)計實(shí)現(xiàn)上完全不同,此外它并不是JMS規(guī)范的實(shí)現(xiàn)。kafka對消息保存時根據(jù)Topic進(jìn)行歸類,發(fā)送消息者成為Producer,消息接受者成為Consumer,此外kafka集群有多個kafka實(shí)例組成,每個實(shí)例(server)成為broker。無論是kafka集群,還是producer和consumer都依賴于zookeeper來保證系統(tǒng)可用性集群保存一些meta信息。
Kafka是一種分布式的,基于發(fā)布/訂閱的消息系統(tǒng) , 它的架構(gòu)包括以下組件:
i. 消息的發(fā)布(publish)稱作producer,消息的訂閱(subscribe)稱作consumer,中間的存儲陣列稱作broker。
ii. 多個broker協(xié)同合作,producer、consumer和broker三者之間通過zookeeper來協(xié)調(diào)請求和轉(zhuǎn)發(fā)。
iii. producer產(chǎn)生和推送(push)數(shù)據(jù)到broker,consumer從broker拉取(pull)數(shù)據(jù)并進(jìn)行處理。
iv. broker端不維護(hù)數(shù)據(jù)的消費(fèi)狀態(tài),提升了性能。 已發(fā)布的消息保存在一組服務(wù)器中,稱之為Kafka集群。集群中的每一個服務(wù)器都是一個代理(Broker). 消費(fèi)者可以訂閱一個或多個話題,并從Broker拉數(shù)據(jù),從而消費(fèi)這些已發(fā)布的消息。
v. 直接使用磁盤進(jìn)行存儲,線性讀寫,速度快:避免了數(shù)據(jù)在JVM內(nèi)存和系統(tǒng)內(nèi)存之間的復(fù)制,減少耗性能的創(chuàng)建對象和垃圾回收。
vi. Kafka使用scala編寫,可以運(yùn)行在JVM上。
如上圖所示,一個典型的Kafka集群中包含:
若干Producer(可以是web前端產(chǎn)生的Page View,或者是服務(wù)器日志,系統(tǒng)CPU、Memory等),若干broker(Kafka支持水平擴(kuò)展,一般broker數(shù)量越多,集群吞吐率越高)。
若干Consumer Group,以及一個Zookeeper集群。Kafka通過Zookeeper管理集群配置,選舉leader,以及在Consumer Group發(fā)生變化時進(jìn)行rebalance。Producer使用push模式將消息發(fā)布到broker,Consumer使用pull模式從broker訂閱并消費(fèi)消息。
Topic在邏輯上可以被認(rèn)為是一個queue,每條消費(fèi)都必須指定它的Topic,可以簡單理解為必須指明把這條消息放進(jìn)哪個queue里。為了使得Kafka的吞吐率可以線性提高,物理上把Topic分成一個或多個Partition,每個Partition在物理上對應(yīng)一個文件夾,該文件夾下存儲這個Partition的所有消息和索引文件。若創(chuàng)建topic1和topic2兩個topic,且分別有13個和19個分區(qū),則整個集群上會相應(yīng)會生成共32個文件夾
開始安裝kafka集群:
1,創(chuàng)建用戶
在所有主機(jī)上添加用戶:
groupadd kafka
useradd kafka -g kafka
2,主機(jī)分配 分別為Hadoop1 ,Hadoop2,Hadoop3
3,綁定hosts
172.16.1.250 hadoop1
172.16.1.252 hadoop2
172.16.1.253 hadoop3
4,下載,解壓
https://kafka.apache.org/
tar -xzf kafka_2.9.2-0.8.1.1.tgz
cd kafka_2.9.2-0.8.1.1
ln -s /usr/local/hadoop/kafka_2.10-0.8.1.1 /usr/local/hadoop/kafka
chown -R kafka:kafka /usr/local/hadoop
在Hadoop3機(jī)器先安裝
5,修改配置文件
cd /usr/local/hadoop/kafka/config
vim /kafka/server.properties
broker.id=3 三臺機(jī)的id不能一樣
port=9092
num.network.threads=2
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=2
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=60000
log.cleaner.enable=false
zookeeper.connect=hadoop1:2181,hadoop2:2181,hadoop3:2181/kafka (zookpeer集群)
zookeeper.connection.timeout.ms=1000000
啟動
bin/kafka-server-start.sh /usr/local/hadoop/kafka/config/server.properties &
6,配置Java環(huán)境
#java
export JAVA_HOME=/soft/jdk1.7.0_79
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export PATH=$PATH:/$JAVA_HOME/bin:$HADOOP_HOME/bin
7,部署kafka集群
由于kafka集群需要依賴zookeeper,所以安裝下zookeeper
見:
https://taoistwar.gitbooks.io/spark-operationand-maintenance-management/content/spark_relate_software/kafka_install.html
7,同步三臺機(jī)器的配置文件,并修改相應(yīng)的broker.id=1,broker.id=2,broker.id=3
cd /usr/local/hadoop/
在Hadoop3機(jī)器上先安裝好一臺了
scp -r kafka/ hadoop1:/usr/local/hadoop/
scp -r kafka/ hadoop2:/usr/local/hadoop/
在Hadoop1機(jī)器上,修改配置文件,并啟動
vim conf/server.properties
broker.id=1 三臺機(jī)的id不能一樣
port=9092
num.network.threads=2
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=2
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=60000
log.cleaner.enable=false
zookeeper.connect=hadoop1:2181,hadoop2:2181,hadoop3:2181/kafka (zookpeer集群)
zookeeper.connection.timeout.ms=1000000
啟動
bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
在Hadoop2機(jī)器上,修改配置文件,并啟動
vim conf/server.properties
broker.id=2 三臺機(jī)的id不能一樣
port=9092
num.network.threads=2
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=2
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=60000
log.cleaner.enable=false
zookeeper.connect=hadoop1:2181,hadoop2:2181,hadoop3:2181/kafka (zookpeer集群)
zookeeper.connection.timeout.ms=1000000
啟動
bin/kafka-server-start.sh /usr/local/hadoop/kafka/config/server.properties &
8驗證
使用Kafka自帶的腳本,啟動基于Console的producer和consumer。
9,錯誤總結(jié):
ttp://wenda.chinahadoop.cn/question/4079?notification_id=290954&rf=false&item_id=10382#!answer_10382
http://blog.csdn.net/wenxuechaozhe/article/details/52664774
http://472053211.blog.51cto.com/3692116/1655844
10,實(shí)際操作見:
https://taoistwar.gitbooks.io/spark-operationand-maintenance-management/content/spark_relate_software/kafka_install.html
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。