【ELFK】之消息队列kafka

一、什么是kafka？

• kafka是由Linkedin公司开发，是一个分布式，支持分区（partition）、多副本的（repilca），基于zookeeper协调的分布式消息系统。

• 它的最大的特性就是可以实时的处理大量数据以满足各种需求场景 ，比如：基于hadoop（分布式）的批处理系统，低延迟的实时系统，storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。

• kafka时一个分布式流式计算平台，但常用于消息系统使用。他是一个分布式消息队列

kafka的特性

• 高吞吐量、低延迟 ： kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒。
• 可扩展性(分布式)： kafka集群支持热扩展
• 持久性、可靠性： 消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
• 容错性： 允许集群中节点失败
• 高并发： 支持数千个客户端同时读写。

1、使用kafka消息队列的好处

解耦

• 允许你独立的扩展或修改两边的处理过程，只要确保他们遵守同样的接口约束。

可恢复性

• 系统的一部分组件失效时，不会影响到整个系统，消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

缓冲

• 有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

灵活性和峰值处理能力

• 访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

异步通信

• 很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

二、kafka的场景应用

• 日志收集： 一个公司可以用kafka手机各种服务的日志，通过kafka以统一接口服务的方式开放给各种consumer（消费者）。

• 消息系统： 解耦和生产者和消费者、缓存消息等。

• 用户活动跟踪： kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

• 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

• 流式处理

• 事件源

二、kafka的模式

kafka的模式不遵守JMS规范(消息队列的规范)。

1、点对点消息传递模式

（一对一，消费者主动拉取数据，消息收到后消息清除）

• 在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。

• 但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。

• 该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序。

2、发布订阅消息传递模式

• 在发布 – 订阅消息系统中，消息被持久化到一个 topic 中。
• 与点对点消息系统不同的是，消费者可以订阅一个或多个 topic，消费者可以消费该 topic 中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。

三、kafka系统基础架构

1、Producer(生产者)

• 消息的生产者，是消息的入口

2、Broker(实例)

• 一台kafka服务器就是一个broker，一个集群由多个broker组成。一个broker可以容纳多个topic（主题）

3、Topic(主题)

• 消息的主题，可以理解成消息的分类，kafka获取到的数据就是按照不同的类型存储在不同的topic主题中。
• topic主题中有很多的分区。

4、Partition(分区)

• Topic的分区，每个Topic可以有很多个分区，分区作用是用来做负载均衡，提高kafka的吞吐量。
• 同一个Topic在不同的分区的数据是不重复的，分区的表现形式就是一个个文件夹。

分区的原因： ①方便在集群中扩展，②实现负载均衡的效果。③提高kafka的吞吐量。④实现高并发的效果。

5、Repica(副本）

• 每一个分区都有多个副本，副本的作用就是备份数据。一个Topic的每个分区都有若干个副本，一个leader和若干个follower。
• 当主分区故障后，副本进行顶替它的位置。

6、leader(领导者)

• 每个分区有多个副本，其中有且仅有一个作为leader，leader是当前负责数据读写的分区。

7、follower(追随者)

• follower跟随leader，所有请求都通过leader路由，数据变更会广播给所有follower，follower和leader保持数据同步，follower只负责备份，不负责数据的读写。
• 如果leader故障，则从follower中选举出一个新的leader。
• 当follower挂掉，卡主或者同步太慢，leader会把这个follower从集群列表中删除，重新创建一个follower。

#实例和主题和分区和副本的理解。
1、实例(broker)就是一台服务器，装好kafka的服务器。多个实例组成一个kafka集群
2、主题(topic)是存储的类型，将不同生产者生成的数据按照类型存储。理解成一个逻辑概念，下面有很多的分区(partition)，分区才是用来存储具体数据的，分区也会称为leader。每一个分区都会进行生成一个或多个副本(Repica)，用来备份分区的数据，也会被称为follower。

一个Topic会产生多个分区Partition，分区中分为Leader和Follower，消息一般发送到Leader，Follower通过数据的同步与Leader保持同步，消费的话也是在Leader中发生消费，如果多个消费者，则分别消费Leader和各个Follower中的消息，当Leader发生故障的时候，某个Follower会成为主节点，此时会对齐消息的偏移量。
1
2
3
4
5

8、Message(消息)

• 消息的实体

9、Consumer

• 消费者，消息的出口

10、Consumer Group

• 多个消费者组成一个消费者组，在kafka的设计中，同一个分区的数据只能被消息者组中的某一个消费者消费，同一个的消费组的消费者可以消费同一个主题中的不同分区的数据。

11、offset偏移量

• 可以唯一的标识一条消息
• 偏移量决定读取数据的位置，不会有线程安全的问题，消费者通过偏移量来决定下次读取的消息（即消费位置）。
• 消费被消费之后，并不会被删除，这样多个业务就可以重复使用kafka的消息。
• 某一个业务也可以通过修改偏移量达到重新读取消息的目的，偏移量由用户控制。
• 消息最终还是会被删除，默认生命周期为1周（168小时）。

12、zookeeper

• kafka集群依赖zookeeper来存储meta(变化)信息。

#offset和zookeeper理解

由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置继续消费，所以consumer需要时实记录自己消费到了哪个offset，以便恢复后继续消费。

kafka 在0.9 版本之前，consumer默认将offset保存在zookeeper中。
从0.9版本后，consumer默认将offset保存在kafka一个内置的topic中，该topic为_consumer_offset.

也就是说，zookeeper的作用就是，生产者输出数据到kafka集群，就必须要找到kafka集群的节点在哪里，这些都是通过zookeeper去寻找的，消费者消费到哪一条数据，也需要zookeeper的支持，从zookeeper获得offset，offset记录上一次消费的数据消费到哪里，这样就可以接着下一跳数据进行消费。
1
2
3
4
5
6
7
8

四、数据存储到kafka中的流程

1、生产者先获取分区中的leader。

2、Producter将消息发送给leader

3、Leader将消息写入本机文件

4、Follower从leader同步消息（follower主动去leader进行同步）

5、Follower将消息写入本地后向leader发送ACK确认消息。

6、leader收到所有副本的ACK后，向生产者发送ACK确认消息。

1、分区的原因

• 便于在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了。
• 可以提高并发，因为可以以分区为单位读写。

2、分区的目的

生产者采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据时有序的。

数据会写入到不同的分区。为什么进行分区？

**1、方便扩展：**因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据。

2、提高并发：以分区为单位进行读写数据，提高消费的处理效率。

五、Kafka原则

类似于负载均衡，当我们向某个服务器发送请求的时候，服务端可能会对请求做一个负载，将流量分发到不同的服务器，那在kafka中，如果某个topic有多个partition，producer又怎么知道该将数据发往哪个partition呢？kafka中有几个原则：

• prtition在写入的时候可以指定需要写入的partition，如果有指定，则写入对应的partition。
• 如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。
• 如果既没指定partition，又没有设置key，则会轮询选出一个partition。

保证消息不丢失是一个消息队列中间件的基本保证，那producer在向kafka写入消息的时候，怎么保证消息不丢失呢？其实上面的写入流程图中有描述出来，那就是通过ACK应答机制！在生产者向队列写入数据的时候可以设置参数来确定是否确认kafka接收到数据，这个参数可设置的值为0、1、all。

1、0代表producer往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效率最高。

2、 1代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。

3、all代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低。

效率最高。

2、 1代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。

3、all代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低。