kafak学习

Posted 2020-11-20 seeasunnyday

发布与订阅消息系统。

数据(消息)的发送者不会直接把消息发送给接受者，这是发布与订阅消息系统的一个特点。发布者以某种方式对消息进行分类，接受者订阅他们，以便接受特定类型的消息。发布与订阅系统一般会有一个broker，也就是发布消息的中心点。

 

 

传统的交互方式： 用户>应用服务器>db 缺点：高并发服务器压力很大。

使用消息队列来替换db连接：用户>应用服务器>消息队列>db处理

 

生产者和消费者

 

kafak的客户端就是kafka的用户，他们被分为两种基本类型：生产者和消费者。除此之外还有其他高级客户端API,用于数据集成的kafka-connect API 和用于流式处理的kafka streams。这些高级客户端API使用生产者和消费者作为内部组件，提供了高级的功能。

 

生产者创建消息，也被称为发布者和写入者。一般情况下，一个消息会被发布到一个特定的主题上。生产者在默认的情况下把消息均衡的分布到主题的所有分区上,而并不关心特定消息会被写到那个分区。

 

消费者读取消息，也被称为订阅者或读者。消费者订阅一个或多个主题，并按照消息生成的顺序来读取他们。消费者通过检查消息的偏移量来判断是否读过消息。偏移量是另一种原数据，它是一个不断递增的整数值，在创建消息试，kafka会把它添加到消息里。消费者是消费群组的一部分，也就是说，会有一个或多个消费者共同读取一个主题。群组保证每个分区只能被一个消费者使用。

 

每一kafka服务称为broker，基于以下几种原因，最好使用多集群。

1.数据类型分离

2.安全需求隔离

3.多数据中心（灾难恢复）

 

注意：kafak的消息复制机制只能在单个集群里进行，不能再多个集群之间进行。但是提供了一个叫做mirrorMaker的工具复制集群之间的数据。

 

kafak的优点？

1.多个生产者，不管客户端在使用单个主题还是多个主题。都是以统一的格式对外提供数据。

2.多个消费者，kakfa支持多个消费者从一个单独的消息流上读取数据，而且消费者之间互不影响。多个消费者可以组成一个群组，他们共享一个消息流，并保证整个群组对每个给定的消息只处理一次。

3.基于磁盘的数据存储，对于单节点或者处理数据慢再或者高峰导致无法及时读取数据，因为kafka对数据进行了持久化。

4.伸缩性，在开发阶段可以使用一个broker，生产情况可以使用集群模式，而且这种扩展是水平扩展，根据业务需求。

5.高性能。

6.数据生态系统。

 

 

kafka使用场景

 

1.活动跟踪，kafak最初受用场景是跟踪用户的活动。网站用户与前端交互之后，前端程序生产用户相关的信息，这些信息就被发布到一个或多个主题上。就这样生产报告。

2.传递消息，将消息发送给应用程序或者其他组件(比如邮件)。

3.度量指标和日志记录，把这些发送到kafka的主题上。

4.提交日志

5.流处理

 

kafka依赖的环境

java,zookeeper（来保存broker元数据信息）

 

kafka的broker配置信息

1.broken.id 每个broken唯一标识，默认值是0，这个值在kafka集群是唯一的。

2.port kafka端口

3.zookeeper.connect 用于保存broker元数据的zookeeperd地址是通过zookeeper.connect来指定的。

4.log.dir kafka把所有消息都保存在磁盘上，存放这些日志片段是通过log.dir指定的。

5.num.recovery.threads.per.data.dir 每个日志片段使用线程的数量。 对于以下3种情况，kafka会只用可配置的线程池来处理日志片段：

　　(1).服务器正常启动，用于打开每个分区的日志片段。

　　(2).服务器崩溃后重启，用于检查和截短每个分区的日志片段。

　　(3).服务正常关闭,用于关闭日志片段。

6.auto.create.topics.enable 默认情况下，kafka会在如下几种情况创建主题：

　　（1）当一个生产者开始往主题写入消息时。

　　（2）当一个消费者开始从主题读取消息时。

　　（3）当任意一个客户端向主题发送元数据请求时。

7.num.partitions 指定主题包含多少个分区，如果自定义可以自己创建主题的时候指定。

8.log.retention.ms 决定数据保存多久，默认168小时也就是一周。

9.log.retention.bates 指定最多可以保存多大。

10.log.segment.bytes 这个作用于日志片段上（消息到达broker时，是被追加到日志片段上），表示日志片段上限多大，如果达到就新开一个分区日志片段。

11.log.segment.ms 指定日志片段多吃时间后被关闭。

12.message.max.bytes 限制消息的大小，如果超过，broker报错。