kafka基础知识

ゝ一纸荒年。 2021-09-20 11:30 400阅读 0赞

分布式集群消息队列

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70]

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

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

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

概念理解：

**Topics and Logs：**

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70 1]

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70 2]

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

Ø Distribution – 分布式

Ø Producers – 生产者

指定topic来发送消息到Kafka Broker

Ø Consumers – 消费者

根据topic消费相应的消息

*  高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。
 *  可扩展性：kafka集群支持热扩展
 *  持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
 *  容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）
 *  高并发：支持数千个客户端同时读写

--------------------

### 11.2 集群部署 ###

**集群规划**：

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

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

kafka是一个分布式消息队列，需要依赖ZooKeeper，请先安装好zk集群。

Zookeeper集群安装步骤略。

下载压缩包（官网地址：[http://kafka.apache.org/downloads.html）][http_kafka.apache.org_downloads.html]

解压：

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][node3_Kafka_broker.id]（分别在node2、[3服务器上执行以下命令修改broker.id][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

启动Kafka集群

A、启动Zookeeper集群。

B、启动Kafka集群。

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

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

\#启动  
./kafka-server-start.sh …/config/server.properties &  
\#关闭  
pkill -9 -f server.properties

创建话题

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

创建topic：

bin/kafka-topics.sh --zookeeper storm01:2181,storm02:2181,storm03:2181 --create --replication-factor 2 --partitions 3 --topic test

*（参数说明：*

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

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

*–topic*：指定新建topic的名称）

查看topic列表：

bin/kafka-topics.sh --zookeeper storm01:2181,storm02:2181,storm03:2181 --list

查看“test”topic描述：

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

创建生产者：

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

创建消费者：

bin/kafka-console-consumer.sh --zookeeper storm01:2181,storm02:2181,storm03:2181 --from-beginning --topic test

注：

查看帮助手册：

bin/kafka-console-consumer.sh help

### 1.3 架构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70 3]

1）Producer ：消息生产者，就是向kafka broker发消息的客户端。

2）Consumer ：消息消费者，向kafka broker取消息的客户端

3）Topic ：可以理解为一个队列。

4） Consumer Group （CG）：这是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制-给consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。

5）Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。

6）Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。

7）Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70]: /images/20210827/cf16aca380fc40c9be4de2cc8baec981.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70 1]: /images/20210827/a35fdce4bfd94f53a89a780efb1b61d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70 2]: /images/20210827/771847f661cc4be3910fd2ed7de7d075.png
[http_kafka.apache.org_downloads.html]: http://kafka.apache.org/downloads.html%EF%BC%89
[broker.id]: http://broker.id
[node3_Kafka_broker.id]: http://xn--node3Kafkabroker-uq9yfsk3jez0nqt8ck8wbo5zf.id
[3_broker.id]: http://xn--3broker-gw3kja48sja692asqmrwgl2h6z7c39joiin44o.id
[kafka-topics.sh]: http://kafka-topics.sh
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0FybmVf_size_16_color_FFFFFF_t_70 3]: /images/20210827/46fbc501f8224d28b3eac40992f58fd1.png