【Kafka专栏】-Kafka从初始到搭建到应用

一、前述

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

kafka集群有多个Broker服务器组成，每个类型的消息被定义为topic。

同一topic内部的消息按照一定的key和算法被分区(partition)存储在不同的Broker上。

消息生产者producer和消费者consumer可以在多个Broker上生产/消费topic。

二、概念理解

Topics and Logs：

Topic即为每条发布到Kafka集群的消息都有一个类别，topic在Kafka中可以由多个消费者订阅、消费。

每个topic包含一个或多个partition（分区），partition数量可以在创建topic时指定，每个分区日志中记录了该分区的数据以及索引信息。如下图：

Kafka只保证一个分区内的消息有序，不能保证一个主题的不同分区之间的消息有序。如果你想要保证所有的消息都绝对有序可以只为一个主题分配一个分区。（分区内有序，一个主题topic不一定是有序的）

分区会给每个消息记录分配一个顺序ID号（偏移量）， 能够唯一地标识该分区中的每个记录。Kafka集群保留所有发布的记录，不管这个记录有没有被消费过，Kafka提供相应策略通过配置从而对旧数据处理。

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

Producers -- 生产者

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

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

• 1.hash
• 2.轮循

指定topic来发送消息到Kafka Broker

Consumers -- 消费者

根据topic消费相应的消息

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->/2/2.2/www -->

消息消费者，自己在zookeeper中维护offset

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

每个消费者都有自己的消费者组，同一个topic中的数据只能在相同的消费组内消费一次，topic的每个partition只能同时被一个消费者组内的消费者消费

不同的消费者组之间消费相同的topic会不影响

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

broker

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

topic

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

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

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

partition

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

组成topic的单元，相当于一个文件，一个partition归一个broker管理，每个partition有副本，有多少个？创建指定

zookeeper的作用：

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

  1、存储原数据

         topic

         partition

         broker

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

  2、存储consumer的offsets

<!-- li { list-style: none; margin: 0; } p { margin: 0; } span.l { color: red; font-weight: bold; } a.mapnode:link {text-decoration: none; color: black; } a.mapnode:visited {text-decoration: none; color: black; } a.mapnode:active {text-decoration: none; color: black; } a.mapnode:hover {text-decoration: none; color: black; background: #eeeee0; } -->

三、集群安装

集群规划：

Zookeeper集群共三台服务器，分别为：node1、node2、node3。

Kafka集群共三台服务器，分别为：node1、node2、node3。

安装Kafka:

tar zxvf kafka_2.10-0.9.0.1.tgz -C /opt/

mv kafka_2.10-0.9.0.1/ kafka

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

核心配置参数说明：

broker.id: broker集群中唯一标识id，0、1、2、3依次增长（broker即Kafka集群中的一台服务器）

注：

当前Kafka集群共三台节点，分别为：node1、node2、node3。对应的broker.id分别为0、1、2。

zookeeper.connect: zk集群地址列表

当前node1服务器上的Kafka目录同步到其他node2、node3服务器上：

scp -r /opt/kafka/ node2:/opt

scp -r /opt/kafka/ node3:/opt

修改node2、node3上Kafka配置文件中的broker.id（分别在node2、3服务器上执行以下命令修改broker.id）(当集群大时，使用此安装命令)

sed -i -e 's/broker.id=.*/broker.id=1/' /opt/kafka/config/server.properties

sed -i -e 's/broker.id=.*/broker.id=2/' /opt/kafka/config/server.properties

 启动kakka集群

A、启动Zookeeper集群。

B、启动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查看帮助手册）

创建topic话题：

bin/kafka-topics.sh --zookeeper node1:2181,node2:2181,node3:2181 --create --replication-factor 2 --partitions 3 --topic test

（参数说明：

--replication-factor：指定每个分区的复制因子个数，默认1个

--partitions：指定当前创建的kafka分区数量，默认为1个

--topic：指定新建topic的名称）

查看topic列表：

bin/kafka-topics.sh --zookeeper node1:2181,node2:2181,node3:2181 --list

查看“test”topic描述：

bin/kafka-topics.sh --zookeeper node1:2181,node2:2181,node3:2181 --describe --topic test

创建生产者：

bin/kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic test

创建消费者：

bin/kafka-console-consumer.sh --zookeeper node1:2181,node2:2181,node3:2181 --from-beginning --topic test

注：

查看帮助手册：

bin/kafka-console-consumer.sh help

删除kafka中的数据。

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

 ./kafka-topics.sh --zookeeper node3:2181,node4:2181,node5:2181 --delete --topic t1205

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

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

rmr /brokers/topics/t1205

③ ：删除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