Kafka架构详解及安装部署

一、前述

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

 

• kafka集群有多个Broker服务器组成，每个类型的消息被定义为topic。
• 同一topic内部的消息按照一定的key和算法被分区(partition)存储在不同的Broker上。
• 消息生产者producer和消费者consumer可以在多个Broker上生产/消费topic。

二、概念理解

2.1、Topics and Logs：

• Topic即为每条发布到Kafka集群的消息都有一个类别，topic在Kafka中可以由多个消费者订阅、消费。
• 每个topic包含一个或多个partition（分区），partition数量可以在创建topic时指定，每个分区日志中记录了该分区的数据以及索引信息。如下图：

 

• Kafka只保证一个分区内的消息有序，不能保证一个主题的不同分区之间的消息有序。如果你想要保证所有的消息都绝对有序可以只为一个主题分配一个分区。（分区内有序，一个主题topic不一定是有序的）
• 分区会给每个消息记录分配一个顺序ID号（偏移量）， 能够唯一地标识该分区中的每个记录。Kafka集群保留所有发布的记录，不管这个记录有没有被消费过，Kafka提供相应策略通过配置从而对旧数据处理。

•  实际上，每个消费者唯一保存的元数据信息就是消费者当前消费日志的位移位置。位移位置是由消费者控制，即：消费者可以通过修改偏移量读取任何位置的数据。

2.2、Producers -- 生产者

消息生产者，自己决定往哪个partition中写入数据

• 1、基于轮询的负载均衡
• 2、基于hash的partition策略

指定topic来发送消息到Kafka Broker

2.3、Consumers -- 消费者

• 根据topic消费相应的消息
• 消息消费者，自己在zookeeper中维护offset
• 每个消费者都有自己的消费者组，同一个topic中的数据只能在相同的消费组内消费一次，topic的每个partition只能同时被一个消费者组内的消费者消费
• 不同的消费者组之间消费相同的topic会不影响

2.4、broker

组成kafka集群的节点，没有主从关系，依靠zookeeper协调，broker负责消息的读写，存储。一个broker可以管理多个partition。

2.5、topic

消息队列，一类消息。topic由partition组成，一个topic有多少个partition？创建可以指定

2.6、partition

• 组成topic的单元，相当于一个文件，一个partition归一个broker管理，每个partition有副本，有多少个？创建指定
• 一个partition对应一个broker，一个broker可以管多个partition，比如说，topic有6个partition，有两个broker，那每个broker就管3个partition。
• 消息直接写入文件（partition可以很简单想象为一个文件，当数据发过来的时候它就往这个partition上面append，追加就行）,并不是存储在内存中；根据时间策略(默认一周)删除，而不是消费完就删除

2.7、zookeeper的作用：

  1、存储原数据

         topic

         partition

         broker

  2、存储consumer的offsets

三、特点

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

kafka与其他消息队列对比 :

• RabbitMQ:分布式，支持多种MQ协议，重量级
• ActiveMQ：与RabbitMQ类似
• ZeroMQ：以库的形式提供，使用复杂，无持久化
• redis：单机、纯内存性好，持久化较差
• kafka：分布式，较长时间持久化，高性能，轻量灵活

消息中间件如何实现每秒几十万的高并发写入？

1、页缓存技术 + 磁盘顺序写：操作系统本身有一层缓存，叫做page cache，是在内存里的缓存，我们也可以称之为os cache，意思就是操作系统自己管理的缓存。

2、零拷贝技术：不需要把os cache里的数据拷贝到应用缓存，再从应用缓存拷贝到Socket缓存了，两次拷贝都省略了，所以叫做零拷贝。

 

四、集群安装

4.1、集群规划：

Zookeeper集群共三台服务器，分别为：node03、node04、node05。

Kafka集群共三台服务器，分别为：node03、node04、node05。

4.2、安装Kafka:  tar -zxvf kafka_2.10-0.9.0.1.tgz -C /opt/

4.3、修改配置文件：config/server.properties

• 节点编号：（不同节点按0,1,2,3整数来配置）

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0

• 真实数据存储位置：

# A comma seperated list of directories under which to store log files
log.dirs=/opt/huawei/kafka-logs

• zookeeper的节点：

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=node03:2181,node04:2181,node05:2181

核心配置参数说明：broker.id: broker集群中唯一标识id，0、1、2、3依次增长（broker即Kafka集群中的一台服务器）

注：当前Kafka集群共三台节点，分别为：node03、node04、node05。对应的broker.id分别为0、1、2。

4.4、节点分发

[root@node03 huawei]# scp -r kafka_2.10-0.8.2.2/ node04:`pwd`
[root@node03 huawei]# scp -r kafka_2.10-0.8.2.2/ node05:`pwd`

4.5、 启动kakka集群

注：先启动Zookeeper集群，再启动Kafka集群。

bin/kafka-server-start.sh config/server.properties

优化：分别在三台服务器上执行以下命令启动：

后台启动：nohup bin/kafka-server-start.sh   config/server.properties > kafka.log 2>&1 &
可以创建个脚本：（放在与bin同一级别下，注意创建后要修改权限：chmod 755 startkafka.sh）

 测试： 

五、集群安装

（kafka-topics.sh --help查看帮助手册）

5.1、创建topic话题：

./bin/kafka-topics.sh --zookeeper node03:2181,node04:2181,node05:2181 --create --replication-factor 2 --partitions 3 --topic lxk

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

5.2、查看topic列表：

bin/kafka-topics.sh --zookeeper node03:2181,node04:2181,node05:2181 --list

5.3、查看“test”topic描述：

bin/kafka-topics.sh --zookeeper node03:2181,node04:2181,node05:2181 --describe --topic lxk

5.4、创建生产者：

bin/kafka-console-producer.sh --broker-list node03:9092,node04:9092,node05:9092 --topic lxk

5.5、创建消费者： 

bin/kafka-console-consumer.sh --zookeeper node03:2181,node04:2181,node05:2181 --from-beginning --topic lxk

注：查看帮助手册  bin/kafka-console-consumer.sh help  （默认更加key的hash值分区，只有value默认key为null,默认10min换一个分区）

5.6、删除kafka中的数据。

① ：在kafka集群中删除topic，当前topic被标记成删除。

 ./kafka-topics.sh --zookeeper node03:2181,node04:2181,node05:2181 --delete --topic t0425

在每台broker节点上删除当前这个topic对应的真实数据。

② ：进入zookeeper客户端，删除topic信息

rm -rf /brokers/topics/t0425

③ ：删除zookeeper中被标记为删除的topic信息

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

转载：https://blog.csdn.net/lixinkuan328/article/details/102052687