1
0
mirror of https://github.com/Snailclimb/JavaGuide synced 2025-08-01 16:28:03 +08:00

Compare commits

...

9 Commits

Author SHA1 Message Date
dlf
5a3d7cb7f9
Merge 5e4c28f10a5cee7405e033fd965aff0a82eed9d6 into 3131d095c2385738d728d34849d4750f92054652 2024-02-23 07:40:46 -08:00
Guide
3131d095c2 [docs update]zookeeper内容完善&部分markdown格式规范 2024-02-23 16:44:05 +08:00
Guide
a8ec6e8d7d
Merge pull request #2276 from Snailclimb/dependabot/npm_and_yarn/ip-2.0.1
Bump ip from 2.0.0 to 2.0.1
2024-02-21 21:55:22 +08:00
Guide
35432c4209
Merge pull request #2277 from zouwei/main
feat:Amazon Document与MongoDB 的差异
2024-02-21 21:54:49 +08:00
wei.zou
504827e05d feat:Amazon Document与MongoDB 的差异 2024-02-21 18:05:24 +08:00
dependabot[bot]
8fc2ca84a8
Bump ip from 2.0.0 to 2.0.1
Bumps [ip](https://github.com/indutny/node-ip) from 2.0.0 to 2.0.1.
- [Commits](https://github.com/indutny/node-ip/compare/v2.0.0...v2.0.1)

---
updated-dependencies:
- dependency-name: ip
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
2024-02-21 03:16:39 +00:00
dlf
5e4c28f10a
Merge branch 'Snailclimb:main' into main 2023-09-30 12:10:31 +08:00
戴林峰
5d2169f076 update kafka 2023-09-30 12:07:43 +08:00
戴林峰
d9e554a4a6 add chapter message queue & ch1 kafka 2023-09-29 18:07:09 +08:00
9 changed files with 316 additions and 9 deletions

View File

@ -274,6 +274,63 @@ MongoDB 单文档原生支持原子性,也具备事务的特性。当谈论 Mo
WiredTiger 日志也会被压缩,默认使用的也是 Snappy 压缩算法。如果日志记录小于或等于 128 字节WiredTiger 不会压缩该记录。
## Amazon Document 与 MongoDB 的差异
Amazon DocumentDB与 MongoDB 兼容) 是一种快速、可靠、完全托管的数据库服务。Amazon DocumentDB 可在云中轻松设置、操作和扩展与 MongoDB 兼容的数据库。
### `$vectorSearch` 运算符
Amazon DocumentDB 不支持`$vectorSearch`作为独立运营商。相反,我们在`$search`运营商`vectorSearch`内部支持。有关更多信息,请参阅 [向量搜索 Amazon DocumentDB](https://docs.aws.amazon.com/zh_cn/documentdb/latest/developerguide/vector-search.html)。
### `OpCountersCommand`
Amazon DocumentDB 的`OpCountersCommand`行为偏离于 MongoDB 的`opcounters.command` 如下:
- MongoDB 的`opcounters.command` 计入除插入、更新和删除之外的所有命令,而 Amazon DocumentDB 的 `OpCountersCommand` 也排除 `find` 命令。
- Amazon DocumentDB 将内部命令(例如`getCloudWatchMetricsV2`)对 `OpCountersCommand` 计入。
### 管理数据库和集合
Amazon DocumentDB 不支持管理或本地数据库MongoDB `system.*``startup_log` 集合也不支持。
### `cursormaxTimeMS`
在 Amazon DocumentDB 中,`cursor.maxTimeMS` 重置每个请求的计数器。`getMore`因此,如果指定了 3000MS `maxTimeMS`,则该查询耗时 2800MS而每个后续`getMore`请求耗时 300MS则游标不会超时。游标仅在单个操作无论是查询还是单个`getMore`请求)耗时超过指定值时才将超时`maxTimeMS`。此外,检查游标执行时间的扫描器以五 (5) 分钟间隔尺寸运行。
### explain()
Amazon DocumentDB 在利用分布式、容错、自修复的存储系统的专用数据库引擎上模拟 MongoDB 4.0 API。因此查询计划和`explain()` 的输出在 Amazon DocumentDB 和 MongoDB 之间可能有所不同。希望控制其查询计划的客户可以使用 `$hint` 运算符强制选择首选索引。
### 字段名称限制
Amazon DocumentDB 不支持点“。” 例如,文档字段名称中 `db.foo.insert({x.1:1})`
Amazon DocumentDB 也不支持字段名称中的 $ 前缀。
例如,在 Amazon DocumentDB 或 MongoDB 中尝试以下命令:
```shell
rs0:PRIMARY< db.foo.insert({"a":{"$a":1}})
```
MongoDB 将返回以下内容:
```shell
WriteResult({ "nInserted" : 1 })
```
Amazon DocumentDB 将返回一个错误:
```shell
WriteResult({
"nInserted" : 0,
"writeError" : {
"code" : 2,
"errmsg" : "Document can't have $ prefix field names: $a"
}
})
```
## 参考
- MongoDB 官方文档(主要参考资料,以官方文档为准):<https://www.mongodb.com/docs/manual/>

View File

@ -57,7 +57,9 @@ ZooKeeper 将数据保存在内存中,性能是不错的。 在“读”多于
- **原子性:** 所有事务请求的处理结果在整个集群中所有机器上的应用情况是一致的,也就是说,要么整个集群中所有的机器都成功应用了某一个事务,要么都没有应用。
- **单一系统映像:** 无论客户端连到哪一个 ZooKeeper 服务器上,其看到的服务端数据模型都是一致的。
- **可靠性:** 一旦一次更改请求被应用,更改的结果就会被持久化,直到被下一次更改覆盖。
- **实时性:** 每个客户端的系统视图都是最新的。
- **实时性:** 一旦数据发生变更,其他节点会实时感知到。每个客户端的系统视图都是最新的。
- **集群部署**3~5 台(最好奇数台)机器就可以组成一个集群,每台机器都在内存保存了 ZooKeeper 的全部数据,机器之间互相通信同步数据,客户端连接任何一台机器都可以。
- **高可用:**如果某台机器宕机,会保证数据不丢失。集群中挂掉不超过一半的机器,都能保证集群可用。比如 3 台机器可以挂 1 台5 台机器可以挂 2 台。
### ZooKeeper 应用场景
@ -253,7 +255,7 @@ Paxos 算法应该可以说是 ZooKeeper 的灵魂了。但是ZooKeeper 并
### ZAB 协议介绍
ZABZooKeeper Atomic Broadcast 原子广播) 协议是为分布式协调服务 ZooKeeper 专门设计的一种支持崩溃恢复的原子广播协议。 在 ZooKeeper 中,主要依赖 ZAB 协议来实现分布式数据一致性基于该协议ZooKeeper 实现了一种主备模式的系统架构来保持集群中各个副本之间的数据一致性。
ZABZooKeeper Atomic Broadcast原子广播) 协议是为分布式协调服务 ZooKeeper 专门设计的一种支持崩溃恢复的原子广播协议。 在 ZooKeeper 中,主要依赖 ZAB 协议来实现分布式数据一致性基于该协议ZooKeeper 实现了一种主备模式的系统架构来保持集群中各个副本之间的数据一致性。
### ZAB 协议两种基本的模式:崩溃恢复和消息广播
@ -262,12 +264,38 @@ ZAB 协议包括两种基本的模式,分别是
- **崩溃恢复**:当整个服务框架在启动过程中,或是当 Leader 服务器出现网络中断、崩溃退出与重启等异常情况时ZAB 协议就会进入恢复模式并选举产生新的 Leader 服务器。当选举产生了新的 Leader 服务器,同时集群中已经有过半的机器与该 Leader 服务器完成了状态同步之后ZAB 协议就会退出恢复模式。其中,**所谓的状态同步是指数据同步,用来保证集群中存在过半的机器能够和 Leader 服务器的数据状态保持一致**。
- **消息广播****当集群中已经有过半的 Follower 服务器完成了和 Leader 服务器的状态同步,那么整个服务框架就可以进入消息广播模式了。** 当一台同样遵守 ZAB 协议的服务器启动后加入到集群中时,如果此时集群中已经存在一个 Leader 服务器在负责进行消息广播,那么新加入的服务器就会自觉地进入数据恢复模式:找到 Leader 所在的服务器,并与其进行数据同步,然后一起参与到消息广播流程中去。
### ZAB 协议&Paxos 算法文章推荐
关于 **ZAB 协议&Paxos 算法** 需要讲和理解的东西太多了,具体可以看下面这几篇文章:
- [Paxos 算法详解](https://javaguide.cn/distributed-system/protocol/paxos-algorithm.html)
- [ZooKeeper 与 Zab 协议 · Analyze](https://wingsxdu.com/posts/database/zookeeper/)
- [Raft 算法详解](https://javaguide.cn/distributed-system/protocol/raft-algorithm.html)
## ZooKeeper VS Etcd
[Etcd](https://etcd.io/) 是一种强一致性的分布式键值存储它提供了一种可靠的方式来存储需要由分布式系统或机器集群访问的数据。Etcd 内部采用 [Raft 算法](https://javaguide.cn/distributed-system/protocol/raft-algorithm.html)作为一致性算法,基于 Go 语言实现。
与 ZooKeeper 类似Etcd 也可用于数据发布/订阅、负载均衡、命名服务、分布式协调/通知、分布式锁等场景。那二者如何选择呢?
得物技术的[浅析如何基于 ZooKeeper 实现高可用架构](https://mp.weixin.qq.com/s/pBI3rjv5NdS1124Z7HQ-JA)这篇文章给出了如下的对比表格,可以作为参考:
| | ZooKeeper | Etcd |
| -------------- | --------------------------------------------------------------------- | ------------------------------------------------------ |
| **语言** | Java | Go |
| **协议** | TCP | Grpc |
| **接口调用** | 必须要使用自己的 client 进行调用 | 可通过 HTTP 传输,即可通过 CURL 等命令实现调用 |
| **一致性算法** | Zab 协议 | Raft 算法 |
| **Watch 功能** | 较局限,一次性触发器 | 一次 Watch 可以监听所有的事件 |
| **数据模型** | 基于目录的层次模式 | 参考了 zk 的数据模型,是个扁平的 kv 模型 |
| **存储** | kv 存储,使用的是 ConcurrentHashMap内存存储一般不建议存储较多数据 | kv 存储,使用 bbolt 存储引擎,可以处理几个 GB 的数据。 |
| **支持 MVCC** | 不支持 | 支持,通过两个 B+ Tree 进行版本控制 |
| **权限校验** | 实现的 ACL | 实现了 RBAC |
| **事务能力** | 提供了简易的事务能力 | 只提供了版本号的检查能力 |
| 部署维护 | 复杂 | 简单 |
实际选用哪个要根据实际业务场景和需求来定Etcd 相对来说更适合云原生领域,并且提供了更稳定的高负载稳定读写能力以及更高的可用性。
## 总结
1. ZooKeeper 本身就是一个分布式程序只要半数以上节点存活ZooKeeper 就能正常服务)。

View File

@ -62,8 +62,7 @@ SELECT * FROM t_order WHERE id >= (SELECT id FROM t_order limit 1000000, 1) LIMI
```sql
SELECT t1.* FROM t_order t1
INNER JOIN (SELECT id FROM t_order limit 1000000, 10) t2
ON t1.id = t2.id
LIMIT 10;
ON t1.id = t2.id;
```
除了使用 INNER JOIN 之外,还可以使用逗号连接子查询。

View File

@ -313,6 +313,16 @@ JVM 这部分内容主要参考 [JVM 虚拟机规范-Java8](https://docs.oracle.
[Web 实时消息推送详解](./system-design/web-real-time-message-push.md)
## 消息队列
### Kafka
[Kafka基础](./mq/kafka/kafka-basis.md)
## 分布式
### 理论&算法&协议

Binary file not shown.

After

Width:  |  Height:  |  Size: 499 KiB

View File

@ -0,0 +1,213 @@
---
title: Kafka基础
category: 消息队列
tag:
- 消息队列
head:
- - meta
- name: keywords
content: Kafka基础
- - meta
- name: description
content: Kafka是一个分布式系统由服务器和客户端组成通过高性能的TCP网络协议进行通信。它可以部署在本地和云环境中的裸机硬件、虚拟机和容器上。
---
## 什么是Kafka
**KafKa**是一个**分布式**的基于**发布/订阅模式**的消息队列主要应用于大数据实时处理领域。Kafka由**服务端**和**客户端**组成通过高性能的TCP网络协议进行通信。它可以部署在本地和云环境中的裸机硬件、虚拟机和容器上。
+ 服务端Kafka作为一个或多个服务器集群运行其中部分服务器构成了存储层Brokers。其他服务器作为[Kafka Connect](https://kafka.apache.org/documentation/#connect)以事件流的形式持续导入和导出数据。同时Kafka集群具有高度的**可扩展性**和**容错性**:如果其中任何一台服务器出现故障,其他服务器将接管其工作,以确保其保持持续运行状态。
+ 客户端:提供接口编写分布式和微服务程序,以并行、大规模和容错的方式读取、写入和处理事件流。
## Kafka的使用背景为什么要使用Kafka
这道题可以理解为为什么要使用消息队列? (消息队列的作用?优点?)
+ **缓冲和削峰**:消息队列在应对类似双十一这样的突发高流量场景中发挥着关键的作用,它可以被视为一个非常有效的**缓冲**和**削峰**机制。考虑以下情形:当突然涌入大量订单请求时,下游的处理服务器可能不具备足够的计算资源来立即处理这些请求。直接将这些请求传递给下游服务器可能导致其超负荷运行,甚至崩溃。消息队列通过将这些突发的订单流量缓存到消息队列中,允许订单处理端按照其自身的处理能力逐一从消息队列中提取订单并进行处理。这种方式有效地平滑了流量高峰,确保了系统的稳定性。因此,消息队列在这里扮演了一个关键的角色,既能够充当缓冲,将请求暂时保存在队列中,又能够削减流量高峰,防止直接冲击到下游服务器,从而实现了系统的平稳运行。
+ **解耦和扩展性**:在项目开发中,由于需求的不确定性,消息队列充当了一个关键的接口层,通过将关键的业务流程解耦。这种解耦使得在后续业务需要扩展时,只需遵循约定并进行数据编程,就能轻松实现所需的扩展能力。
+ **异步通信**:消息队列提供了一种强大的机制,允许用户将消息放入队列中,而无需立即处理它们。这种异步处理方式可以显著提高业务处理速度,例如在需要发送短信验证码的用户注册等场景中,业务主线程可以将发送短信验证码的任务放入消息队列,然后继续处理其他业务,而无需等待短信发送完成。这种机制极大地提高了系统的效率和响应性。
+ **可恢复性**:即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。
## Kafka的使用场景
+ **日志收集**:一个公司可以用 Kafka 可以收集各种服务的 log通过 Kafka 以统一接口服务的方式开放给各种 consumer。也就是在系统各个运行的位置将日志输送到一个统一的地方进行保存和处理。
+ **消息系统**:将业务进行解耦合,分成消息的生产者和消费者,实现异步通信、可恢复、解耦和缓冲与削峰。
+ **用户行为跟踪**Kafka 经常被用来记录web用户或者 app 用户的各种活动,如浏览网页、搜索、点击等活动,这些活动信息被各个服务器发布到 kafka 的 Topic 中然后订阅者通过订阅这些topic来做实时的监控分析或者装载到 Hadoop、数据仓库中做离线分析和挖掘。
## Kafka架构
<img src="kafka-basis.assets/image-20230929154749134.png" alt="image-20230929154749134" style="zoom:50%;" />
+ **Producer**(生产者):生产者可以将数据发布到所选择的topic主题中。生产者负责将记录分配到topic的指定分区partition这里可以使用多个partition循环发送来实现多个server负载均衡。
+ **Consumer**(消费者): 消息消费者,从 Kafka Broker 取消息的客户端。
+ **Broker**Broker是kafka的服务节点一个Broker就是一个服务节点即Kafka服务器。一个broker可以容纳多个topic。broker可以看作事消息的代理Producers往Brokers里面指定的Topic写消息Consumers从Brokers里面拉取指定的消息然后进行业务处理broker在中间起到一个代理保存消息的中转站。
+ **Topic**(主题):可以理解为一个队列,一个 Topic 又分为一个或多个分区。
+ **Partition**分区Topic 是一个逻辑的概念,它可以细分为多个分区,每个分区只属于单个主题。 同一个主题下不同分区包含的消息是不同的分区在存储层面可以看作一个可追加的日志Log文件消息在被追加到分区日志文件的时候都会分配一个特定的偏移量Offset
+ **Offset**偏移量是消息在分区中的唯一标识Kafka 通过它来保证消息在分区内的顺序性,不过 Offset 并不跨越分区也就是说Kafka 保证的是分区有序性而不是主题有序性,即局部有序。
+ **Replication** (副本):是 Kafka 保证数据高可用的方式Kafka 同一 Partition 的数据可以在多 Broker 上存在多个副本,通常只有主副本对外提供读写服务,当主副本所在 Broker 崩溃或发生网络一场Kafka 会在 Controller 的管理下会重新选择新的 Leader 副本对外提供读写服务。
+ **Record** :实际写入 Kafka 中并可以被读取的消息记录。每个 Record 包含了 key、value 和 timestamp。
+ **Consumer Group** 消费者组CG消费者组内每个消费者负责消费不同分区的数据提高消费能力。一个分区只能由组内一个消费者消费消费者组之间互不影响。所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者。
## Kafka为什么要分区Kafka分区的目的
+ **提供并行处理能力**通过将消息分散到多个分区Kafka可以实现消息的并行处理。消费者可以独立地从不同的分区中读取消息从而提高整体的处理能力。
+ **提高可靠性和可伸缩性**Kafka通过复制机制实现数据的可靠性和冗余存储每个分区可以配置多个副本这些副本分布在不同的Broker节点上当一个副本不可用时可以使用其他副本来继续提供服务。同时通过增加分区的数量可以增加整个系统的处理能力、存储容量和实现负载均衡提高并发度提高效率。
## Kafka如何实现消息有序性
kafka中每一个partition中的消息在写入的时候都是有序的而且单独一个partition只能由一个消费者去消费可以在里面保证消息的顺序性但是分区之间的额消息是不保证有序的。总结就是kafka只保证了单个partition的有序性并没有保证多个partition的有序性因为如果需要保证多个partition的有序性那么整个kafka就退化成了单一队列毫无并发性可言了。
**那如果需要保证全局的有序性怎么办呢?**
1. 创建一个Topic只创建一个Partition这样就不会存在多个partition也自然是全局有序的了。
2. 生产者发送消息的时候发送到指定的partition。
## Kafka为什么这么快
kafka会把接收到的信息都写入硬盘中来保证消息的不丢失。为了优化写入速度Kafka采用了顺序写入和MMFile两个技术。
**写入数据**
1. 顺序写入:因为硬盘是机械结构,每次读写都会寻址->写入其中寻址是一个“机械动作”它是最耗时的。所以硬盘最讨厌随机I/O最喜欢顺序I/O。为了提高读写硬盘的速度Kafka就是使用顺序I/O。
2. MMFile即使是顺序写入硬盘的访问速度还是与内存速度有较大的差距。因此Kafka并不是实时写入硬盘的它还利用了操作系统的分页存储来利用内存提高I/O效率。
> Memory Mapped Files(后面简称mmap)也被翻译成 内存映射文件 在64位操作系统中一般可以表示20G的数据文件它的工作原理是直接利用操作系统的Page来实现文件到物理内存的直接映射。完成映射之后你对物理内存的操作会被同步到硬盘上操作系统在适当的时候
**读取数据**
1. 基于sendfile实现**零拷贝**,减少拷贝次数。
> 零拷贝是指计算机执行IO操作时CPU不需要将数据从一个存储区域复制到另一个存储区域从而可以减少上下文切换以及CPU的拷贝时间。它是一种`I/O`操作优化技术。(减少用户态与内核态之间的数据复制次数)
>sendfile具体流程如下
>
>1. sendfile系统调用文件数据被copy至内核缓冲区
>2. 从内核缓冲区copy至内核中socket相关的缓冲区
>3. socket相关的缓冲区copy到协议引擎
2. 批量压缩它把所有的消息都变成一个批量的文件并且进行合理的批量压缩减少网络IO损耗。Producer使用GZIP或者Snappy格式对消息几个进行压缩压缩的好处就是减少传输的数据量减轻对网络传输的压力。
**文件分段**
kafka 的队列topic被分为了多个区partition每个partition又分为多个段segment所以一个队列中的消息实际上是保存在N多个片段文件中。通过分段的方式每次文件操作都是对一个小文件的操作非常轻便同时也增加了并行处理能力。
**批量发送**
Kafka 允许进行批量发送消息先将消息缓存在内存中然后一次请求批量发送出去比如可以指定缓存的消息达到某个量的时候就发出去或者缓存了固定的时间后就发送出去如100条消息就发送或者每5秒发送一次这种策略将大大减少服务端的I/O次数。
## Kafka中的消息是否会丢失和重复消费
Kafka在**生产端**发送消息和**消费端**消费消息时都可能会**丢失**一些消息。
### Producer消息丢失
生产者在发送消息时会有一个ack机制当acks=0或者acks=1时都可能会丢失消息。
> 背景知识Producer发送消息时是直接与Broker中的Leader Partition进行交互的然后其他的副本再从Leader Partition中进行数据的同步。因此在发送消息的时候Producer只需要找到对应Topic的Leader Partition进行消息发送即可。
>
> 消息发送的流程:
>
> 1. 将消息发送到对应Topic下的Leader Partition
> 2. Leader Partition收到消息并将消息写入Page Cache定时刷盘进行持久化顺序写入磁盘
> 3. Foller Partition 拉取Leader Partition的消息并同Leader Partition的数据保持一致待消息拉取完毕后再给Leader Partition回复ack确认消息。
> 4. 待Leader与Foller 同步完数据并收到所有ISR中的Replica副本的ack后Leader Partition会给Producer回复ack确认消息。
Producer端为了提升发送效率减少I/O操作发送数据的时候是将多个请求合并成一个个RecordBatch并将其转换成为Request请求**异步**将数据发送出去或者按时间间隔方式每隔一定的时间自动发送出去因此Producer端消息丢失更多是因为消息根本没有发送到Kafka Broker端。
因此,**导致Producer端消息没有成功发送有以下原因**
1. 网络原因由于网络原因数据根本没有到达Broker端。
2. 数据原因消息太大超出Broker承受的范围导致Broker拒收消息。
**Producer消息确认机制**
Producer端配置了消息确认机制来确认消息是否生产成功使用ack确认机制。
1. asks=0:只要发送就自认为成功并不进行消息接收成功的ack确认。
1. 不能保证消息是否发送成功。
2. 生产环境完全不可用。
2. acks=1:当Leader Partition接收成功时进行ack确认确认后表示成功
1. 只要Leader Partition存活就可以保证不丢失保证了吞吐量。
2. 生产环境中如果需要保证吞吐量可以用这个。
3. acks=-1或者all所有Leader Partition和Foller PartitionISR都接收成功时进行ack确认确认后表示成功。
1. 保证消息不丢失,但是吞吐量低。
2. 生产环境要求数据不能丢失可以采用该方式。
### Broker端丢失场景
Broker接收到数据后会将数据进行持久化存储到磁盘为了提高吞吐量和性能采用的是**异步批量刷盘的策略**也就是说按照一定的消息量和时间间隔进行刷盘这一点和mysql、redis很像。首先数据会背存储到**PageCache**中,至于什么时候将 Cache 中的数据刷盘是由「**操作系统**」根据自己的策略决定或者调用 fsync 命令进行强制刷盘,如果此时 Broker 宕机 Crash 掉,且选举了一个落后 Leader Partition 很多的 Follower Partition 成为新的 Leader Partition那么落后的消息数据就会丢失。既然Broker是异步刷盘的那么数据就有可能会丢失比如刷盘之前操作系统崩了并且Kafka中没有提供**同步刷盘**机制。)
虽然Kafka 通过「**多 Partition (分区)多 Replica副本机制」**已经可以最大限度的保证数据不丢失,但是当数据已经写入 PageCache 中但是还没来得及刷写到磁盘,此时如果所在 Broker 突然宕机挂掉或者停电,极端情况还是会造成数据丢失。
### Consumer端丢失场景剖析
> Consumer通过Pull模式主动的去Kafka集群中拉消息
>
> 1. 在消息拉取的过程中,有个消费者组的概念,多个 Consumer 可以组成一个消费者组即 Consumer Group每个消费者组都有一个Group-Id。同一个 Consumer Group 中的 Consumer 可以消费同一个 Topic 下不同分区的数据,但是不会出现多个 Consumer 去消费同一个分区的数据。
> 2. 拉取到消息后进行业务逻辑处理,待处理完成后,会进行 ACK 确认,即提交 Offset 消费位移进度记录。
> 3. 最后 Offset 会被保存到 Kafka Broker 集群中的 **__consumer_offsets** 这个 Topic 中,且每个 Consumer 保存自己的 Offset 进度。
Consumer端丢失消息主要体现在**消费端offset的自动提交**如果开启了自动提交万一消费到数据还没处理完此时consumer直接宕机未处理完的数据丢失了下次也消费不到了因为offset已经提交完毕下次会从offset处开始消费新消息。这种丢失情况的解决方法是**采用消费端的手动提交**
### 消息重复消费
**生产端消息重复发送**
生产端发送一条消息但是未得到broker的ack生产端又重新发了一条消息。这个时候两条消息都被broker接收到了消费端从broker拉取消息时就会造成重复消费。
> kafka新版本已经在broker中保证了接收消息的幂等性比如2.4版本),只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可默认是false不开启。
>
> 新版本解决方案是producer发送消息时加上PID和Sequence NumberPID是Producer的唯一IDSequence Number是数据的序列号。
>
> broker接收到消息的时候就会检查有没有收到过这个消息根据PID和Sequence Number
**消费端消息重复消费**
消费端拉取一部分数据消费完成之后提交offset之前挂掉了此时offset未提交当前消息就会被重复消费。
解决办法添加分布式锁在offset提交之后再删key这样就保证了同一个消息只会被消费一次。
## Kafka顺序消息
Kakfa如果需要保证消息的顺序性则需要牺牲一定的性能。具体的顺序方式就是使用单一的消费者由一个消费者消费可以保证消息消费的顺序性但是消息发送的顺序性还是无法保证因为消息发送端有重传机制如果一次性发送两条消息前一条消息发送失败引发重传就会导致消息发送乱序。此时如果需要保证发送和接收的顺序那就使用发送的ack机制确认发送成功之后再发送下一条消息并且只能有一个Partition。但是这种方式会导致kafka性能低下。
**高效的解决方式**
类似于tcp发送的方式给每一个消息添加一个序号然后消费端每次拉取全部消息拉取回来之后再排序根据排序之后的数据进行处理。
## Kafka与其它MQ之间的区别为什么选择使用Kafka
**kafka相对于rocketMQ、rabbitMQ来说与它们最大的区别就是分布式存储这也是kafka高性能的最主要原因**。使用分布式存储理念,一个主题下多个分区,同时可以被多个消费者和生产者去使用,也增加了接受消息和消费消息的能力!
## 参考
+ Kafka官方文档https://kafka.apache.org/documentation
+ Kafka 设计架构原理详细解析https://blog.csdn.net/qq_32828253/article/details/110732652
+ Kafka为什么这么快https://zhuanlan.zhihu.com/p/147054382
+ Kafka如何保证消息不丢失https://zhuanlan.zhihu.com/p/459610418
+ kafka专题kafka的消息丢失、重复消费、消息积压等线上问题汇总及优化https://blog.csdn.net/qq_45076180/article/details/111561984

6
pnpm-lock.yaml generated
View File

@ -2168,8 +2168,8 @@ packages:
resolution: {integrity: sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==}
dev: false
/ip@2.0.0:
resolution: {integrity: sha512-WKa+XuLG1A1R0UWhl2+1XQSi+fZWMsYKffMZTTYsiZaUD8k2yDAj5atimTUD2TZkyCkNEeYE5NhFZmupOGtjYQ==}
/ip@2.0.1:
resolution: {integrity: sha512-lJUL9imLTNi1ZfXT+DU6rBBdbiKGBuay9B6xGSPVjUeQwaH1RIGqef8RZkUtHioLmSNpPR5M4HVKJGm1j8FWVQ==}
dev: false
/is-binary-path@2.1.0:
@ -3078,7 +3078,7 @@ packages:
resolution: {integrity: sha512-7maUZy1N7uo6+WVEX6psASxtNlKaNVMlGQKkG/63nEDdLOWNbiUMoLK7X4uYoLhQstau72mLgfEWcXcwsaHbYQ==}
engines: {node: '>= 10.13.0', npm: '>= 3.0.0'}
dependencies:
ip: 2.0.0
ip: 2.0.1
smart-buffer: 4.2.0
dev: false