分布式消息通信 kafka, kafka与zookeeper进行集成

小灰灰 2022-05-26 13:28 219阅读 0赞

Kafka是一款分布式消息发布和订阅的系统，具有高性能，高吞吐量的特点而被广泛应用于大数据传输场景，它是由LinkedIn

公司开发，使用 Scale语言编写的，之后成为Apache基金会的一个顶级项目。

kafka介绍网站

[http://kafka.apache.org/documentation/\#gettingStarted][http_kafka.apache.org_documentation_gettingStarted]

kafka 是分布式的消息和订阅系统，高性能，高吞吐量，scale 语言进行编写。

内置分区，实现集群，运用在行为跟踪和日志收集这些功能上。

kafka可以使用在系统的流量监控上面，配合elk进行使用。

![70][]

![70 1][]

kafka和activemq的最大不同之处在于：

activemq主动推送消息给到 constomer 消费端，而 kafka 是去主动获得消息的

安装kafka，单机安装：

https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.0/kafka\_2.12-1.1.0.tgz

tar -zxvf xxxx.tgz 文件

先安装并且启动 zookeeper的集群，集群的启动操作如下博客：

https://blog.csdn.net/baidu\_24545901/article/details/79924289

对kafka进行集群操作

对 kafka文件进行配置

编辑kafka当中的config目录下的  server.properties文件

编辑另外两台kafka当中的 config目录下的  server.properties,将它们都设置成不同的  broker.id

本篇中设置的三台broker.id与虚拟机ip地址对应关系如下：

<table style="width:200px;"> 
 <tbody> 
  <tr> 
   <td>虚拟机ip</td> 
   <td>broker.id</td> 
  </tr> 
  <tr> 
   <td>192.168.159.136</td> 
   <td>0</td> 
  </tr> 
  <tr> 
   <td>192.168.159.135</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>192.168.159.137</td> 
   <td>2</td> 
  </tr> 
 </tbody> 
</table>

然后把三台server.properties文件后面找到   listeners=P=PLAINTEXT://192.168.159.136:9092,三台都设置与其ip对应的listeners

然后再把zookeeper.connect 这个属性都更改一下，三台都改成改成下面的

zookeeper.connect=192.168.159.136:2181,192.168.159.135:2181,192.168.159.137:2181

启动 三台机器上的kafka

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

启动之后可以在 kafka的 logs目录下查看  logs文件，查看kakfa在启动过程中是否有什么错误信息。

没有错误信息的话，那么可以登录zookeeper

./zkCli.sh -server localhost:2181

然后 ls / 可以查看 zookeeper当中又增加了哪些个节点信息，可以看到增加下下面的几个节点。

cluster、controller、controller\_epoch、brokers、isr\_change\_notification、consumers、latest\_producer\_id\_block、config

节点的作用：

controller:控制节点

brokers-kafka集群的 broker信息  topic

consumer  ids/owners/offsets

可以查看 ls /brokers/ids  可以查看到 kafka当中集群配置的节点数目

get /controller   使用这个命令可以查看到当前是哪个节点是集群的主节点(主机器)

![70 2][]

查看上面的brokerid 即可显示主节点.

基本操作：

接下来实现 kafka消息通信，实现  192.168.159.136与192.168.159.135之间的消息通信

首先第一步，创建一个topic

sh kafka-topics.sh --create --zookeeper 192.168.159.136:2181 --replication-factor 1 --partitions 1 --topic MyTopic

查看 kafka集群当中有多少个 topic

sh kafka-topics.sh --list --zookeeper localhost:2181 还可以登录zookeeper 查看  ls /brokers/topics  使用这个命令来查看有多少个topic

使用kafka的producer端向kafka集群发送一条消息(kafka的默认端口是 9092)：

sh kafka-console-producer.sh --broker-list 192.168.159.136:9092 --topic MyTopic

![70 3][]

然后再另外一台机器上执行  kafka的消息接收命令:

sh kafka-console-consumer.sh --bootstrap-server 192.168.159.136:9092 --topic MyTopic --from-beginning

![70 4][]

可以看到kafka consumer接收到provider端发送过来的消息了。

kafka的实现细节:

消息

消息有key 和 value；消息批量发送

topic 和 partition

topic是用来存储消息的逻辑概念，

![70 5][]

如果说把topic看成是一个数据库的话，那么每个partition则是数据库中的每张表

1000w的表做分表操作，分成10个表 每个表 100w

每个表partition当中存储的数据是以消息的key的hash值去做路由

kafka的高吞吐因素:

1.它是以顺序写的方式进行数据的存储；频繁的io操作，网络io和磁盘io

2.批量发送概念。

3.零拷贝

为什么是零拷贝呢，对比传统的消息传送模型 磁盘-->内存-->用户空间-->内存网卡，这些过程会经过4次上下文切换和数据的复制，频繁io操作浪费性能

![70 6][]

而kafka的模型是这样的

![70 7][]

日志策略：

日志保留策略

时间和大小

日志压缩策略

topic里面partition的物理存储可以通过下面的命令来观看:

sh kafka-topics.sh --create --zookeeper 192.168.159.136:2181 --replication-factor 1 --partitions 3 --topic aaatopic

这里的 --partitions 3表示的是创建的 三个  partition的节点  --topic表示的是 创建的topic名称是 aaatopic

replication-factor 1这个属性代表的含义是  交叉备份的副本数，支持多台 kafka集群进行相互之间的备份操作，可以使用副本备份增加可靠性，比如说当某台服务器上面的消息丢失了，我们可以有另外一台服务器上面的备份消息存在，保证消息不至于丢失

执行完毕这条命令之后，可以看到  kafka的日志目录  可以在server.properties当中配置   其/tmp/kafka-logs这个目录当中都各有一个节点，节点的名称后面加上了-num，比如 192.168.159.136这台服务器

![70 8][]

kafka的可靠性机制：

生产者发送消息到 broker，有三种确认方式 (request.required.acks)

acks=0:producer不会等待broker(leader)发送ack，因为发送消息网络超时或者  broker crash

Leader还没有commit消息  2.  Leader与Follower数据不同步，既有可能丢失也有可能会重发

acks=1：当leader接收到消息之后会发送ack，丢失会重发，丢失的概率很小。

acks=-1：当所有的follower都同步消息成功后发送 ack，丢失消息的可能性会比较低。

zookeeper当中的副本机制：

sh kafka-topics.sh --create --zookeeper 192.168.159.136:2181 --replication-factor 3 --partitions 3 --topic aaatopic

使用这句代码创建 kafka  topic的时候 副本的时候，利用replication-factor这个属性指定副本数量为3 partitions这个属性表示的是节点的数量是 3，创建出来的副本模型可以参考下面的图：

![70 9][]

leader选举：

ISR副本同步队列，维护的是所有的follow节点。

1.副本中的所有的节点必须要和zookeeper中保持连接状态

2.副本的最后一条消息的offset和leader副本的最后一条消息的offset之间的差值不能超过指定的阀值，这个阀值可以设置。

![70 10][]

在zookeeper当中 isr的副本消息同步可以在这个节点下看到：

![70 11][]

文件存储机制：

在执行了sh kafka-topics.sh --create --zookeeper 192.168.159.136:2181 --replication-factor 3 --partitions 3 --topic aaatopic  这条命令之后，kafka的partition节点会被存储在 设置的日志目录当中，默认是在 /tmp/kafka-logs/   这个目录下的，我们先创建一个 kafka的topic

sh kafka-topics.sh --create --zookeeper 192.168.159.136:2181 --replication-factor 2 --partitions 3 --topic zzzz

kafka会在一台机器上创建下面的两个节点，

执行  ls /tmp/kafka-logs/zzzz-0   查看其中一个节点的信息

![70 12][]

我们可以看到该partition中会存在  一个 .index的文件和另外一个 .log的文件

然后在使用kafka的一个节点去进行消息的发送

sh kafka-console-producer.sh --broker-list 192.168.159.136:9092 --topic zzzz

![70 13][]

然后另外一端接收消息：

sh kafka-console-consumer.sh --bootstrap-server 192.168.159.136:9092 --topic MyTopic --from-beginning

![70 14][]

执行下面的语句查看  kafka中发送的消息解释：

sh kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/zzzz-0/00000000000000000000.log --print-data-log

看到下面的消息:

![70 15][]

.index 和  .log之间的关系就像下面的关系:

![70 16][]

[http_kafka.apache.org_documentation_gettingStarted]: http://kafka.apache.org/documentation/#gettingStarted
[70]: /images/20220526/19b11674da0242cc9363730304e8bd3d.png
[70 1]: /images/20220526/11aef448dd1e405ebbee859b38e8bd3e.png
[70 2]: /images/20220526/fafa144ab0a9463d906ad79bb4d48b03.png
[70 3]: /images/20220526/0004a5c998264c5c8de8292782620a22.png
[70 4]: /images/20220526/3c76ea24af934ed2a5b028daf3ba2ef8.png
[70 5]: /images/20220526/451aef96cd14485eab36df9fc7dc665d.png
[70 6]: /images/20220526/3ab14fb7248c43e29208eb534fa8f8aa.png
[70 7]: /images/20220526/559ae6b9c32540c7b5a1e37ee866b506.png
[70 8]: /images/20220526/4dfd209b4f80462187089f8b55bb0e82.png
[70 9]: /images/20220526/ee9f956dfe8a44efaf710129f1cb5dfc.png
[70 10]: /images/20220526/6de8f51b64c14a879c3a8e0e1ccde9fc.png
[70 11]: /images/20220526/ec47339dc33049c2ad704352aa0b7545.png
[70 12]: /images/20220526/6d93f7bd0f7b4749bc48d1a259285c44.png
[70 13]: /images/20220526/a73a630c89944608ab8ad3f20eb34df1.png
[70 14]: /images/20220526/c2d44c7a287e4ae990a598af2d6ff7c4.png
[70 15]: /images/20220526/dfae8f8325924e5c9be66776c1e4910f.png
[70 16]: /images/20220526/0b958a66d1444a7897dcf5aa11a439ca.png