Kafaka介绍

Kafka是一个分布式的消息队列系统(Message Queue)

一、Kafka介绍

kafka是一个高吞吐的分布式消息队列系统。特点是生产者消费者模式，先进先出（FIFO）保证顺序，自己不丢数据，默认每隔7天清理数据。消息列队常见场景：系统之间解耦合、峰值压力缓冲、异步通信。

二、Kafka生产消息、存储消息、消费消息

图片.png

Kafka架构是由producer（消息生产者）、consumer（消息消费者）、borker(kafka集群的server，负责处理消息读、写请求，存储消息，在kafka cluster这一层这里，其实里面是有很多个broker)、topic（消息队列/分类相当于队列，里面有生产者和消费者模型）、zookeeper(元数据信息存在zookeeper中，包括：存储消费偏移量，topic话题信息，partition信息) 这些部分组成。
kafka里面的消息是有topic来组织的，简单的我们可以想象为一个队列，一个队列就是一个topic，然后它把每个topic又分为很多个partition，这个是为了做并行的，在每个partition内部消息强有序，相当于有序的队列，其中每个消息都有个序号offset，比如0到12，从前面读往后面写。一个partition对应一个broker，一个broker可以管多个partition，比如说，topic有6个partition，有两个broker，那每个broker就管3个partition。这个partition可以很简单想象为一个文件，当数据发过来的时候它就往这个partition上面append，追加就行，消息不经过内存缓冲，直接写入文件，kafka和很多消息系统不一样，很多消息系统是消费完了我就把它删掉，而kafka是根据时间策略删除，而不是消费完就删除，在kafka里面没有一个消费完这么个概念，只有过期这样一个概念。
producer自己决定往哪个partition里面去写，这里有一些的策略，譬如hash。consumer自己维护消费到哪个offset，每个consumer都有对应的group，group内是queue消费模型（各个consumer消费不同的partition，因此一个消息在group内只消费一次），group间是publish-subscribe消费模型，各个group各自独立消费，互不影响，因此一个消息在被每个group消费一次。

三、Kafka的特点

 系统的特点：生产者消费者模型，FIFO
Partition内部是FIFO的，partition之间呢不是FIFO的，当然我们可以把topic设为一个partition，这样就是严格的FIFO。
 高性能：单节点支持上千个客户端，百MB/s吞吐，接近网卡的极限
 持久性：消息直接持久化在普通磁盘上且性能好
直接写到磁盘中去，就是直接append到磁盘里去，这样的好处是直接持久化，数据不会丢失，第二个好处是顺序写，然后消费数据也是顺序的读，所以持久化的同时还能保证顺序，比较好，因为磁盘顺序读比较好。
 分布式：数据副本冗余、流量负载均衡、可扩展
分布式，数据副本，也就是同一份数据可以到不同的broker上面去，也就是当一份数据，磁盘坏掉的时候，数据不会丢失，比如3个副本，就是在3个机器磁盘都坏掉的情况下数据才会丢，在大量使用情况下看这样是非常好的，负载均衡，可扩展，在线扩展，不需要停服务。
 很灵活：消息长时间持久化+Client维护消费状态
消费方式非常灵活，第一原因是消息持久化时间跨度比较长，一天或者一星期等，第二消费状态自己维护消费到哪个地方了可以自定义消费偏移量。

四、集群搭建

1.服务器准备

Zookeeper集群共三台服务器，分别为：node05、node06、node07
Kafka集群共三台服务器，分别为：node05、node06、node07

2.Zookeeper集群准备

Kafka是一个分布式消息队列，需要依赖ZooKeeper，请先安装好zk集群。
Zookeeper集群安装步骤略，可参照【Storm学习笔记三：全分布式搭建】。

3.Kafka安装解压

（1）解压

[root@node05 software] tar xf kafka_2.10-0.9.0.1.tgz -C /opt/sxt
[root@node05 sxt] mv kafka_2.10-0.9.0.1/ kafka

（2）修改配置文件 server.properties

[root@node05 sxt] cd /opt/sxt/kafka/config

############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
#broker集群中唯一标识id，0、1、2、3依次增长（broker即Kafka集群中的一台服务器）

############################# Zookeeper #############################
# root directory for all kafka znodes.
#zookeeper.connect=localhost:2181
zookeeper.connect=node05:2181,node06:2181,node07:2181

（3）安装包分发到node06/node07

scp -r kafka/ node06:`pwd`
scp -r kafka/ node07:`pwd`

分别修改node06、node07上Kafka配置文件中的broker.id为1、2

4.启动Kafka集群

启动Zookeeper集群

#分别在三台服务器上执行以下命令启动
zkServer.sh start

启动Kafka集群

#分别在三台服务器上执行以下命令启动
cd /opt/sxt/kafka
bin/kafka-server-start.sh config/server.properties

5.测试Kafka

XShell新启动一个标签页

[c:\~]$ ssh node05

创建Topic

#查看帮助手册
bin/kafka-topics.sh --help 
#创建Topic
bin/kafka-topics.sh --zookeeper node05:2181,node06:2181,node07:2181 --create --replication-factor 2 --partitions 3 --topic test

参数说明：
--replication-factor：指定每个分区的复制因子个数，默认1个
--partitions：指定当前创建的kafka分区数量，默认为1个
--topic：指定新建topic的名称

查看Topic列表

bin/kafka-topics.sh --zookeeper node05:2181,node06:2181,node07:2181 --list

查看"test"Topic描述

bin/kafka-topics.sh --zookeeper node05:2181,node06:2181,node07:2181 --describe --topic test

创建生产者（注意端口9092）

bin/kafka-console-producer.sh --broker-list node05:9092,node06:9092,node07:9092 --topic test

参数说明：
--topic test 表示给test主题创建生产者
注：此时shell会阻塞住，等待生产数据，不断输入数据按回车即可

创建消费者
此时从新开一个标签页执行以下命令

bin/kafka-console-consumer.sh --zookeeper node05:2181,node06:2181,node07:2181 --from-beginning --topic test

参数说明：
--from-beginning 消费时从头消费数据
--topic test 表示给test主题创建生产者
注：此时就会看到生产者发送的数据，且生产者新生产一个，消费者就接收到一个

五、Kafka的leader的均衡机制

当一个broker停止或者crashes时，所有本来将它作为leader的分区将会把leader转移到其他broker上去，极端情况下，会导致同一个leader管理多个分区，导致负载不均衡，同时当这个broker重启时，如果这个broker不再是任何分区的leader,kafka的client也不会从这个broker来读取消息，从而导致资源的浪费。
kafka中有一个被称为优先副本（preferred replicas）的概念。如果一个分区有3个副本，且这3个副本的优先级别分别为0,1,2，根据优先副本的概念，0会作为leader 。当0节点的broker挂掉时，会启动1这个节点broker当做leader。当0节点的broker再次启动后，会自动恢复为此partition的leader。不会导致负载不均衡和资源浪费，这就是leader的均衡机制。
在配置文件conf/ server.properties中配置开启（默认就是开启）：

auto.leader.rebalance.enable true

六、Kafka 0.11版本改变

kafka 0.8.2版本消费者offset存储在zookeeper中，对于zookeeper而言每次写操作代价是很昂贵的，而且zookeeper集群是不能扩展写能力。kafka 0.11版本默认使用新的消费者api ,消费者offset会更新到一个kafka自带的topic【__consumer_offsets】中。以消费者组groupid 为单位，可以查询每个组的消费topic情况：

#查看所有消费者组
./kafka-consumer-groups.sh --bootstrap-server c7node1:9092,c7node2:9092,c7node3:9092 --list

#查看消费者消费的offset位置信息
./kafka-consumer-groups.sh --bootstrap-server c7node1:9092,c7node2:9092,c7node3:9092  --describe --group MyGroupId

#重置消费者组的消费offset信息 ，--reset-offsets –all-topics 所有offset。--to-earliest 最小位置。
# --execute 执行
./kafka-consumer-groups.sh --bootstrap-server c7node1:9092,c7node2:9092,c7node3:9092 --group MyGroupId --reset-offsets --all-topics --to-earliest --execute