Kafka系列一之架构介绍和安装

Kafka架构介绍和安装

写在前面
还是那句话，当你学习一个新的东西之前，你总得知道这个东西是什么？这个东西可以用来做什么？然后你才会去学习它，使用它。简单来说，kafka既是一个消息队列，如今，它也演变为一个分布式的流处理平台，这一点就是很厉害的了。所以，学习kafka对于大数据的小伙伴们来说，是非常有益的。
（一）Kafka概述
1.Kafka是一个分布式的流处理平台，有消息的订阅系统（订阅之后，一旦有新的消息产生，你就会收到消息），可以以多副本的方式进行存储，可以构建实时数据管道，以及实时的流处理，能够横向水平扩展，容错等特点。大数据的实时流处理场景很多时候都会使用到kafka

上面是官网中的图片
Kafka和消息系统类似，消息中间件的一种：有生产者和消费者的概念。

起源：

Kafka 是 Apache 基金会开源的一个分布式发布 - 订阅消息中间件，流处理平台。 它起源于 LinkedIn，由 Scala 和 Java两种语言编写而成。于 2011 年成为 Apache 项目，2012 成为 Apache 基金会下顶级项目。

Kafka 专为分布式高吞吐系统而设计。相比较其他消息中间件，如 RabbitMq 等，Kafka 具有更好的吞吐量，内置分区，复制和固有的容错能力，使得它非常适合应用在大数据领域。另外，Kafka 还支持离线、在线消费消息。

特性：

• 低延迟 - Kafka 支持低延迟消息传递，速度极快，能达到 200w 写/秒；
• 高性能 - Kafka对于消息的发布、订阅都具有高吞吐量。即使存储了 TB 级的消息，依然能够保证稳定的性能；
• 可靠性 - Kafka 是分布式，分区，复制和容错的，保证零停机和零数据丢失。
• 可拓展性 - Kafka 支持集群水平拓展。
• 耐用性 - Kafka 使用"分布式提交日志"，消息能够快速的持久化的磁盘上。

（二）Kafka架构

从上面官网给出的图片中，我们知道，kafka是有以下几个主要的该丽娜
1）producer：生产者，

2）consumer：消费者，

3）broker：一个broker就是一个kafka,

4）topic：主题，给生产的消息打上一个标签，代表是谁生产的.

First a few concepts:[几个概念]
Kafka is run as a cluster on one or more servers.

The Kafka cluster stores streams of records in categories called topics.

Each record consists of a key, a value, and a timestamp.

（三）Kafka的安装部署
我们首先来到官网上去，http://kafka.apache.org/quickstart
从这个快速开始中，知道我们的kafka是依赖于zookeeper的，所以我们第一步来安装zookpeer,然后安装Kafka

zookeeper的安装

1. 下载zookeeper，地址在cdh5的页面http://archive.cloudera.com/cdh5/cdh/5/
2. 解压到自己指定的目录中去，我一般喜欢~/app，这个看自己的习惯就好，这里我把解压后的文件修改名称为zk,
3. 把解压的目录导出，

[hadoop@hadoop000 zk]$ vim ~/.bash_profile 
# .bash_profile

# Get the aliases and functions
if [ -f ~/.bashrc ]; then
        . ~/.bashrc
fi

# User specific environment and startup programs

export JAVA_HOME=/home/hadoop/app/jdk1.8.0_144
export PATH=$JAVA_HOME/bin:$PATH

export FLUME_HOME=/home/hadoop/app/flume
export PATH=$FLUME_HOME/bin:$PATH


export ZK_HOME=/home/hadoop/app/zk
export PATH=$ZK_HOME/bin:$PATH

#:wq 保存退出

4.再source一个系统环境

[hadoop@hadoop000 zk]$ source ~/.bash_profile 
[hadoop@hadoop000 zk]$ 

5.配置zookeeper，来到解压目录下的conf文件
拷贝一份

[hadoop@hadoop000 conf]$ cp zoo_sample.cfg zoo.cfg
[hadoop@hadoop000 conf]$ vim zoo.cfg 
#修改数据的临时存放目录，默认是在tmp目录下的
dataDir=/home/hadoop/app/tmp/zk
#:wq 保存退出

6.启动zookeeper

[hadoop@hadoop000 conf]$ cd ../bin/
[hadoop@hadoop000 bin]$ zkServer.sh start
JMX enabled by default
Using config: /home/hadoop/app/zk/bin/../conf/zoo.cfg
Starting zookeeper ... already running as process 10943.
[hadoop@hadoop000 bin]$ 

7.查看进程

[hadoop@hadoop000 bin]$ jps
27179 Jps
10943 QuorumPeerMain
[hadoop@hadoop000 bin]$ 
#看见QuorumPeerMain，说明就启动成功了

Kafka的安装部署

按照官网上的指南，我们先下载Kafka,点击官网上面的downloadhttp://kafka.apache.org/downloads 在选择版本的时候，我们推荐使用0.9.0.0版本的，目前这个版本在上产上比较常用，

1. 下载https://archive.apache.org/dist/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz 因为kafka是scala语言写的，后面我们也可能会使用到scala，因为我安装的scala是2.11版本。这个下载的版本主要看自己的情况，在官网上选择合适的就好
2. 解压到指定目录，再导出到系统环境中‘’

[hadoop@hadoop000 ~]$ vim ~/.bash_profile 
# .bash_profile

# Get the aliases and functions
if [ -f ~/.bashrc ]; then
        . ~/.bashrc
fi

# User specific environment and startup programs

export JAVA_HOME=/home/hadoop/app/jdk1.8.0_144
export PATH=$JAVA_HOME/bin:$PATH

export FLUME_HOME=/home/hadoop/app/flume
export PATH=$FLUME_HOME/bin:$PATH


export ZK_HOME=/home/hadoop/app/zk
export PATH=$ZK_HOME/bin:$PATH


export KAFKA_HOME=/home/hadoop/app/kafka
export PATH=$KAFKA_HOME/bin:$PATH

#:wq保存退出

3.再source一下

4.配置，修改解压目录下config配置文件

配置server.properties[需要注意的地方]
broker.id=0  解释：是kafka，一个broker
listeners      解释：监听的端口
host.name    解释：当前机器
log.dirs       解释：kafka的日志存放目录
zookeeper.connect  zk的地址

修改好之后，保存退出

5.现在，就可以来启动了
如何启动呢？？看官网是如何说的，http://kafka.apache.org/quickstart

bin/zookeeper-server-start.sh config/zookeeper.properties
2）.再启动我们的kafka
bin/kafka-server-start.sh config/server.properties

#启动kafka的命令
kafka-server-start.sh $KAFKA_HOME/config/server.properties

6.启动之后

[hadoop@hadoop000 ~]$ jps
27251 Kafka
27334 Jps
10943 QuorumPeerMain
[hadoop@hadoop000 ~]$ 
#可以看见Kafka的进程，就对啦

好的，到这里我们的kafka就安装完成了，接下来，我们就开始做broker的部署…
推荐一下：Kafka系列二之部署与使用
http://blog.csdn.net/liuge36/article/details/78593404

更多Kafka 相关好文 ：https://blog.csdn.net/liuge36/article/category/9392861

----------------拓展----- 2019-12-09 （更新）-----------

下载地址:
	Zookeeper:
	http://mirror.bit.edu.cn/apache/zookeeper/current/

	Scala:
	http://www.scala-lang.org/download/2.11.8.html

	Kafka:
	http://kafka.apache.org/downloads


一.Zookeeper部署
	1.下载解压zookeeper-3.4.6.tar.gz
	[root@hadoop001 software]# tar -xvf zookeeper-3.4.6.tar.gz
	[root@hadoop001 software]# mv zookeeper-3.4.6 zookeeper
	[root@hadoop001 software]#
	[root@hadoop001 software]# chown -R root:root zookeeper
	2.修改配置
	[root@hadoop001 software]# cd zookeeper/conf
	[root@hadoop001 conf]# ll
	total 12
	-rw-rw-r--. 1 root root  535 Feb 20  2014 configuration.xsl
	-rw-rw-r--. 1 root root 2161 Feb 20  2014 log4j.properties
	-rw-rw-r--. 1 root root  922 Feb 20  2014 zoo_sample.cfg
	[root@hadoop001 conf]# cp zoo_sample.cfg zoo.cfg
	[root@hadoop001 conf]# vi zoo.cfg 
	# The number of milliseconds of each tick
	tickTime=2000
	# The number of ticks that the initial
	# synchronization phase can take
	initLimit=10
	# The number of ticks that can pass between
	# sending a request and getting an acknowledgement
	syncLimit=5
	# the directory where the snapshot is stored.
	# do not use /tmp for storage, /tmp here is just
	# example sakes.

	dataDir=/opt/software/zookeeper/data

	# the port at which the clients will connect
	clientPort=2181
	# the maximum number of client connections.
	# increase this if you need to handle more clients
	#maxClientCnxns=60
	#
	# Be sure to read the maintenance section of the
	# administrator guide before turning on autopurge.
	#
	# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
	#
	# The number of snapshots to retain in dataDir
	#autopurge.snapRetainCount=3
	# Purge task interval in hours
	# Set to "0" to disable auto purge feature
	#autopurge.purgeInterval=1

	server.1=hadoop001:2888:3888
	server.2=hadoop002:2888:3888
	server.3=hadoop003:2888:3888

	~
	"zoo.cfg" 36L, 1028C written
	[root@hadoop001 conf]# cd ../
	[root@hadoop001 zookeeper]#  mkdir data
	[root@hadoop001 zookeeper]# touch data/myid
	[root@hadoop001 zookeeper]# echo 1 > data/myid
	[root@hadoop001 zookeeper]#

	3.hadoop002/003,也修改配置,就如下不同
	[root@hadoop001 software]# scp -r  zookeeper 192.168.137.141:/opt/software/
	[root@hadoop001 software]# scp -r  zookeeper 192.168.137.142:/opt/software/

	[root@hadoop002 zookeeper]# echo 2 > data/myid
	[root@hadoop003 zookeeper]# echo 3 > data/myid

	###切记不可echo 3>data/myid,将>前后空格保留,否则无法将 3 写入myid文件

	4.启动Zookeeper集群
	[root@hadoop001 bin]# ./zkServer.sh start
	[root@hadoop002 bin]# ./zkServer.sh start
	[root@hadoop003 bin]# ./zkServer.sh start

	5.查看Zookeeper状态
	[root@hadoop001 bin]# ./zkServer.sh status
	JMX enabled by default
	Using config: /opt/software/zookeeper/bin/../conf/zoo.cfg
	Mode: follower
	[root@hadoop002 bin]#  ./zkServer.sh status
	JMX enabled by default
	Using config: /opt/software/zookeeper/bin/../conf/zoo.cfg
	Mode: leader
	[root@hadoop002 bin]# 
	[root@hadoop003 bin]#  ./zkServer.sh status
	JMX enabled by default
	Using config: /opt/software/zookeeper/bin/../conf/zoo.cfg
	Mode: follower
	[root@hadoop003 bin]# 


	6.进入客户端
	[root@hadoop001 bin]# ./zkCli.sh 
	[zk: localhost:2181(CONNECTED) 0] ls /
	[zookeeper, yarn-leader-election, hadoop-ha, rmstore]
	[zk: localhost:2181(CONNECTED) 1] 
	[zk: localhost:2181(CONNECTED) 1] help
	ZooKeeper -server host:port cmd args
		stat path [watch]
		set path data [version]
		ls path [watch]
		delquota [-n|-b] path
		ls2 path [watch]
		setAcl path acl
		setquota -n|-b val path
		history 
		redo cmdno
		printwatches on|off
		delete path [version]
		sync path
		listquota path
		rmr path
		get path [watch]
		create [-s] [-e] path data acl
		addauth scheme auth
		quit 
		getAcl path
		close 
		connect host:port
	[zk: localhost:2181(CONNECTED) 2] 

二.Kafka部署
1.解压并配置Scala
[root@hadoop001 software]# tar -xzvf scala-2.11.8.tgz
[root@hadoop001 software]# chown -R root:root scala-2.11.8
[root@hadoop001 software]# ln -s scala-2.11.8 scala


#环境变量
[root@hadoop001 software]# vi /etc/profile
export SCALA_HOME=/opt/software/scala
export PATH=$SCALA_HOME/bin:$PATH

[root@hadoop001 software]# source /etc/profile
[root@hadoop001 software]# scala
Welcome to Scala 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_45).
Type in expressions for evaluation. Or try :help.

 
2.下载基于Scala 2.11的kafka版本为0.10.0.1 
[root@hadoop001 software]# tar -xzvf kafka_2.11-0.10.0.1.tgz
[root@hadoop001 software]# ln -s kafka_2.11-0.10.0.1 kafka
[root@hadoop001 software]# 

3.创建logs目录和修改server.properties
[root@hadoop001 software]# cd kafka
[root@hadoop001 kafka]# mkdir logs
[root@hadoop001 kafka]# cd config/
[root@hadoop001 config]# vi server.properties
broker.id=1
port=9092
host.name=192.168.137.141
log.dirs=/opt/software/kafka/logs
zookeeper.connect=192.168.137.141:2181,192.168.137.142:2181,192.168.137.143:2181/kafka

4.环境变量
[root@hadoop001 config]# vi /etc/profile
export KAFKA_HOME=/opt/software/kafka
export PATH=$KAFKA_HOME/bin:$PATH
[root@hadoop001 config]# source /etc/profile

5.另外两台机器如上操作
 
6.启动/停止
[root@sht-sgmhadoopdn-01 kafka]# nohup kafka-server-start.sh config/server.properties &
[root@sht-sgmhadoopdn-02 kafka]# nohup kafka-server-start.sh config/server.properties &
[root@sht-sgmhadoopdn-03 kafka]# nohup kafka-server-start.sh config/server.properties &
###停止
bin/kafka-server-stop.sh

---------------------------------------------------------------------------------------------------------------------------------------------
7.模拟实验1
创建test topic
bin/kafka-topics.sh --create 
--zookeeper 192.168.137.141:2181,192.168.137.142:2181,192.168.137.143:2181/kafka 
--replication-factor 3 --partitions 3 --topic test


在一个终端，启动Producer，并向我们上面创建的名称为my-replicated-topic5的Topic中生产消息，执行如下脚本：
bin/kafka-console-producer.sh 
--broker-list 192.168.137.141:9092,192.168.137.142:9092,192.168.137.143:9092 --topic test

在另一个终端，启动Consumer，并订阅我们上面创建的名称为my-replicated-topic5的Topic中生产的消息，执行如下脚本：
bin/kafka-console-consumer.sh 
--zookeeper 192.168.137.141:2181,192.168.137.142:2181,192.168.137.143:2181/kafka 
--from-beginning --topic test
可以在Producer终端上输入字符串消息行，就可以在Consumer终端上看到消费者消费的消息内容。