58 lines
5.9 KiB
Markdown
58 lines
5.9 KiB
Markdown
# Kafka
|
||
## 简介
|
||
### 消息队列应用场景
|
||
缓存/削峰、解耦、异步通信
|
||
### 缓存/削峰
|
||
当生产端生产数据的速率大于消费端消费数据的速率时，消息队列可用于对消费不完的数据进行缓存
|
||
### 解耦
|
||
当数据生产方来源存在多个（例如数据库、网络接口等），且消费方也存在多种（Hadoop大数据平台、spring boot服务实例等）时，可以将消息队列中存储的消息作为数据从生产端到消费端的中间格式。
|
||
|
||
生产端负责将产生的数据转化成特定格式的消息发送到消息队列，而消费端负责消费消息队列中的消息。不同种类的生产端和消费端只用针对消息的队列中的消息各自进行适配即可。
|
||
|
||
### 异步通信
|
||
通过消息队列，可以实现多个服务实例之间的异步调用，服务调用方在将调用信息封装到消息并发送消息到mq后，即可返回，并不需要同步等待。被调用方可以异步的从mq中获取消息并进行消费。
|
||
|
||
### 消息队列模式
|
||
#### 点对点
|
||
一个消息队列可以对应多个生产者和多个消费者，**但消息只能被一个消费者进行消费**，消费者将数据拉取并消费后，向mq发送确认，mq中将被消费的消息删除
|
||
#### 发布订阅模式
|
||
**生产者产生的数据可以被多个消费者消费**，生产者将消息发送到topic，订阅该topic的消费者会拉取消息进行消费，消费完成后并不会删除该消息，该消息仍可被其他订阅该topic的消费者进行消费。（消息并不会永久保存在消息队列中，通常会设置超期时间，消息保存超过该时间后自动删除）
|
||
|
||
### kafka架构
|
||
#### 分区存储
|
||
kafka将一个topic分为了多个分区，用于分布式存储数据。而每个分区都跨多台服务器实例进行存储，从而增加容错性。对特定分区来说，其存在于多台服务器实例，其中一台为主机，其他服务器为从机，**主机会处理对该分区数据所有的读写请求**（由于读写操作全都在主机上发生，而每台服务器都担当了某些分区的主机和其他分区的从机，故而读写请求在不同的服务器之间进行了负载均衡），而从机只会被动的复制主机修改。
|
||
#### group消费
|
||
每个消费者实例都属于一个consume group，一个consume group则是可以保存多个消费者实例。对于发送到kafka mq topic的每一条消息，都会被广播给订阅了该topic的每个consume group。而consume group接收到消息后，只会将消息发送给group中的一个消费者实例来进行处理，故而在consume group中，实现了消息在不同消费实例之间的负载均衡。
|
||
> 在kafka mq中，topic的实际订阅者是consume group，kafka mq会将topic中的消息广播给所有订阅了该topic的consume group，而每条消息都只会被consume group中的一个消费者实例进行消费
|
||
>
|
||
> 在一个consume group中，每个消费者实例都负责某一topic中相应的分区，每个特定分区只有一个消费者实例负责。
|
||
|
||
#### zookeeper
|
||
zookeeper作为注册中心，用于记录存在多个kafka实例时，当前已上线且状态正常的kafka实例，以及各个分区leader实例的信息
|
||
|
||
#### 生产者
|
||
##### 分区
|
||
生产者客户端将会决定将消息发送到某个分区，可以通过负载均衡随机决定将消息发送到哪个分区，也可以提供一个key并通过算法决定将消息发送到哪个分区。
|
||
|
||
决定将消息发送到哪个分区后，生产者会直接将消息发送到该分区对应的leader broker中。
|
||
|
||
##### 批量发送
|
||
生产者并不会每次产生消息后都立即将消息发送到broker，而是会累积消息，**直到消息数据积累到特定大小（默认为16K）后**，才会将消息发送给broker。
|
||
> **消息发送条件**
|
||
>
|
||
> - batch.size: 当消息累计到特定大小（默认为16K）后，发送给broker。如果消息的大小大于`batch.size`,那么并不会对消息做累积操作。**发送到broker的请求将会包含多个batch，每个batch对应一个分区，不同分区的消息通过不同的batch进行发送**
|
||
> - linger.ms: 生产者将会把发送给broker的两个请求之间的消息都累积到一个batch里，该batch将会在下次发送请求给broker时发送。但是，对特定分区累积大小如果没有达到`batch.size`的限制，哪个通过linger.ms来控制该消息延迟发送的最长时间。`linger.ms`单位为ms，**其默认值为0，代表即使消息累积没有达到`batch.size`，也会立马发送给broker**。若`linger.ms`设置为1000，则代表没有累积到`batch.size`大小的消息也会在延迟1s后发送给broker
|
||
|
||
##### ack级别
|
||
生产者将消息发送到broker后，broker对生产者有三种应答级别：
|
||
- 0：生产者发送数据后无需等待broker返回ack应答
|
||
- 1：生产者发送消息到broker后，leader broker将消息持久化后向生产者返回ack
|
||
- -1：生产者发送过来的数据，leader broker和isr队列中所有的broker都持久化完数据后，返回ack
|
||
|
||
##### 异步发送
|
||
kafka生产者异步发送是指生产者调用kafka客户端接口将发送的消息传递给kafka的客户端，此时消息并未发送到broker，而异步调用接口即返回。调用异步接口可以通过指定回调来获取消息传递到的topic、分区等信息。
|
||
|
||
##### 同步发送
|
||
除了异步调用接口外，还可以调用同步调用的接口来发送消息。在调用异步接口发送消息时，仅仅将消息放到缓冲区后调用就返回，可能放到缓冲区的消息并没有发送到broker。可以通过调用同步发送消息的接口来发送消息，其会阻塞并等待broker将消息持久化后返回的异常或者ack应答。
|
||
|
||
想要同步发送消息，只需要对异步接口返回的future对象调用`.get`即可，其会等待future对象完成。 |