kafka原理解析

Posted 2020-11-01

什么是消息系统：
用来处理消息队列的系统；
什么是消息队列：
是用于进程间通信或同一进程内的线程间通信的软件工程组件；
他们使用一个队列来传播消息----这里传播的消息就是-->传递控制或者内容；

这里面有个问题：
消息队列是用来提高性能，加速消息传输的吗？
答案，显然不是，消息队列虽然提供了数据上的冗余，但它不是一种缓存。
如果你想加速，直接把消费者和生产者结合在一起写，中间自己加一个全内存的
queue，没有了持久化，没有了网络传输，这样岂不是更快；
消息队列，最好的诠释就是，“fire and forget”，英译就是“解耦”，它实现了生产者与
消费者的有效解耦，降低了系统复杂性；
作为一个生产者，它主要关心的应该就是自己的生产工作，它不应该关心自己生产的东西，到底
被谁消费，如何消费。它应该就是简单的把生产好的东西，往一个仓库一放（即fire），然后就可
以不管了（forget）,毫无心理负担。至于后面的事，消息如何交付给消费者，这种交付方式是不
是会丢失消息之类的可靠性问题一概不管（这也就是为什么消息队列不仅是一个中间结果存放区
的原因）。这个作为中间仓库，负责与消费者打交道，同时保证后续交付可靠性的角色，就是消
息队列来担当的。

kafka将消息以topic为单位进行归纳；
将向kafka topic发布消息的程序称为：producers；
将预订topics并消费消息的程序称为consumer；
kafka以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个broker；
producers通过网络将消息发送到kafka集群，集群向消费者提供消息；

消息队列的分类：

点对点的消息队列queue：
一个消息只能被一个消费者消费，但是queue支持有多个消费者。
消息被消费后，queue中不再存储，就是一个消息被消费后，其他消费者不能再消费。

发布与订阅的消息队列topic：kafka
一个消息可以被多个消费者消费，实现消息共享；

为什么要搭建kafka：
活动数据的采集：
网站用户相关行为数据，如：PV,UV等
运营数据
监控核心系统性能指标

这些数据的特点：
数据不可变
数据量庞大
需实时处理

kafka是一个分布式的消息系统（集群）：
负载均衡+失败迁移

消息系统的基本结构：

broker是接收消息的机制；每台消息服务器可以有多个broker，多台服务器组成消息系统的分布式集群；
zookeeper作为协调者，broker注册到zookeeper上，zookeeper同步信息给其他broker；
kafka使用zookeeper存储消息的元信息；

基本概念：
1.topic：特制kafka处理的消息源，代表kafka中的消息；
2.partition（分区）：topic消息物理上的分组；一个topic可以有多个partition，每个partition是一个有序的队列，partition每一个消息都会被分配给有序的id；
3.message：指的是具体的消息，是通信的基本单位，每个生产者可以向topic发送主题；
4.producer：生产者，向kafka的一个topic发布消息的过程叫做生产；
5.consumer：消费者，订阅topic并处理其发布的消息的过程叫做消费；
6.broker：缓存代理，kafka集群中的一台或者多台服务器，负责真正的接收消息和处理消息；

kafka的应用场景：
message（消息系统）
websit activity tracking（网站活性追踪）
log aggregation（日志收集中心）

首先, Partition对上层应用不透明. 用户可以指定产生的消息pull到那个paritition.
topic之下分partition的原因负载均衡. 不考虑消息的顺序, 单个Topic可以用多个Partition, leader均匀分散在全体broker上, 缓解单个broker过热问题.
还有一个原因是, 每个.log文件(segment)对应的index文件中, offset为32bit, 如果单个Partition的record数目超过4GB, 应该怎么呢? 使用多Partition.
如果hdd大小是8TB, ssd大小为1TB, 一个topic下产生的消息超过了磁盘大小,怎么办呢?
如果一个服务器配置24/36块盘, 把topic分成partition, 粒度小一下, 单机磁盘的负载也会根据均衡.
如果一个topic数据, 太大, 必然有partitioning的需要.

可用的内存和分区数：Brokers会为每个分区分配replica.fetch.max.bytes参数指定的内存空间，假设replica.fetch.max.bytes=1M，且有1000个分区，则需要差不多1G的内存，确保 分区数最大的消息不会超过服务器的内存，否则会报OOM错误。同样地，消费端的fetch.message.max.bytes指定了最大消息需要的内存空间，同样，分区数最大需要内存空间 不能超过服务器的内存。所以，如果你有大的消息要传送，则在内存一定的情况下，只能使用较少的分区数或者使用更大内存的服务器。

1 什么是消费者组
consumer group是kafka提供的可扩展且具有容错性的消费者机制。既然是一个组，那么组内必然可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。当然，每个分区只能由同一个消费组内的一个consumer来消费。
consumer group组的三个特性：
● consumer group下可以有一个或多个consumer instance，consumer instance可以是一个进程，也可以是一个线程
● group.id是一个字符串，唯一标识一个consumer group
● consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)
消费者位置(consumer position)
消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在Kafka中这个位置信息有个专门的术语：位移(offset)。很多消息引擎都把这部分信息保存在服务器端(broker端)。这样做的好处当然是实现简单，但会有三个主要的问题：1. broker从此变成有状态的，会影响伸缩性；2. 需要引入应答机制(acknowledgement)来确认消费成功。3. 由于要保存很多consumer的offset信息，必然引入复杂的数据结构，造成资源浪费。而Kafka选择了不同的方式：每个consumer group保存自己的位移信息，那么只需要简单的一个整数表示位置就够了；同时可以引入checkpoint机制定期持久化，简化了应答机制的实现。
3 位移管理(offset management)
3.1 自动VS手动
Kafka默认是定期帮你自动提交位移的(enable.auto.commit = true)，你当然可以选择手动提交位移实现自己控制。另外kafka会定期把group消费情况保存起来，做成一个offset map，如下图所示：

上图中表明了test-group这个组当前的消费情况。

3.2 位移提交
老版本的位移是提交到zookeeper中的，图就不画了，总之目录结构是：/consumers/<group.id>/offsets/<topic>/<partitionId>，但是zookeeper其实并不适合进行大批量的读写操作，尤其是写操作。因此kafka提供了另一种解决方案：增加consumeroffsets topic，将offset信息写入这个topic，摆脱对zookeeper的依赖(指保存offset这件事情)。consumer_offsets中的消息保存了每个consumer group某一时刻提交的offset信息。依然以上图中的consumer group为例，格式大概如下：

consumers_offsets topic配置了compact策略，使得它总是能够保存最新的位移信息，既控制了该topic总体的日志容量，也能实现保存最新offset的目的。compact的具体原理请参见：Log Compaction
至于每个group保存到__consumers_offsets的哪个分区，如何查看的问题请参见这篇文章：Kafka 如何读取offset topic内容 (consumer_offsets)

kafka的原理：
kafka的设计理念：
持久化
约束是吞吐量而不是功能
已经被使用了的状态信息保存为数据使用者的一部分，而不是保存在服务器上
分布式系统

kafka在zookeeper的存储结构：
kafka会在zookeeper存储元信息，会有多个目录，目录结构如下：
consumer获取消息是通过zookeeper处理发送到broker，kafka集群随机分配请求道broker服务器上；

kafka的部署方式：
1.单Broker部署：一台message server上部署一个broker服务；
2.单机多broker部署（伪分布式）：一台message server上部署多个broker服务；
3.多机多broker部署（真正的分布式）：
多台物理message server，每台message server上部署一个或多个broker服务；
每个broker有个独立的id，这个id在创建broker时进行定义；
broker id按顺序排列；

单broker部署：

单机多broker部署：

jps命令查看kafka的broker信息；
kafka启动报错：UseCompressedOps，修改bin/kafka-run-class.sh，去掉此选项；

kafka的配置文件和demo演示：