kafka详细介绍，安装，配置-创新互联

1、简介

  Kafka is a distributed,partitioned,replicated commit logservice。它提供了类似于JMS的特性，但是在server)成为broker。无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。

Kafka是一种分布式的，基于发布/订阅的消息系统 ， 它的架构包括以下组件：

i. 消息的发布（publish）称作producer，消息的订阅（subscribe）称作consumer，中间的存储阵列称作broker。

ii. 多个broker协同合作，producer、consumer和broker三者之间通过zookeeper来协调请求和转发。

iii. producer产生和推送(push)数据到broker，consumer从broker拉取(pull)数据并进行处理。

iv. broker端不维护数据的消费状态，提升了性能。 已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理(Broker). 消费者可以订阅一个或多个话题，并从Broker拉数据，从而消费这些已发布的消息。

v. 直接使用磁盘进行存储，线性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。

vi. Kafka使用scala编写，可以运行在JVM上。

如上图所示，一个典型的Kafka集群中包含：

若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高）。

若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

Topic & Partition

Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息和索引文件。若创建topic1和topic2两个topic，且分别有13个和19个分区，则整个集群上会相应会生成共32个文件夹

开始安装kafka集群：

1，创建用户

在所有主机上添加用户：

groupadd kafka

useradd kafka -g kafka

2,主机分配 分别为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机器先安装

5,修改配置文件

cd  /usr/local/hadoop/kafka/config

vim  /kafka/server.properties

broker.id=3  三台机的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环境

#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，同步三台机器的配置文件，并修改相应的broker.id=1,broker.id=2,broker.id=3

cd /usr/local/hadoop/

在Hadoop3机器上先安装好一台了

scp -r kafka/  hadoop1:/usr/local/hadoop/

scp -r kafka/  hadoop2:/usr/local/hadoop/

在Hadoop1机器上，修改配置文件，并启动

vim conf/server.properties

broker.id=1  三台机的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机器上，修改配置文件，并启动

vim conf/server.properties

broker.id=2  三台机的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,错误总结:

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,实际操作见：

https://taoistwar.gitbooks.io/spark-operationand-maintenance-management/content/spark_relate_software/kafka_install.html

另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。