Kafaka介绍
Kafka是一个分布式的消息队列系统(Message Queue)
一、Kafka介绍
kafka是一个高吞吐的分布式消息队列系统。特点是生产者消费者模式,先进先出(FIFO)保证顺序,自己不丢数据,默认每隔7天清理数据。消息列队常见场景:系统之间解耦合、峰值压力缓冲、异步通信。
二、Kafka生产消息、存储消息、消费消息
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