Merge 5e4c28f10a5cee7405e033fd965aff0a82eed9d6 into af606816a2fa0b362372e75c5661272dd46b8a36

teriri,fix:并发编程-重要知识点-乐观锁和悲观锁详解

docs/.vuepress/sidebar/index.ts

@@ -325,6 +325,7 @@ export default sidebar({
                 "3-commonly-used-cache-read-and-write-strategies",
                 "redis-data-structures-01",
                 "redis-data-structures-02",
+                "redis-skiplist",
                 "redis-persistence",
                 "redis-memory-fragmentation",
                 "redis-common-blocking-problems-summary",

docs/cs-basics/data-structure/bloom-filter.md

@@ -273,10 +273,10 @@ root@21396d02c252:/data# redis-cli
 
 > 注意：key : 布隆过滤器的名称，item : 添加的元素。
 
-1. **`BF.ADD`**：将元素添加到布隆过滤器中，如果该过滤器尚不存在，则创建该过滤器。格式：`BF.ADD {key} {item}`。
+1. `BF.ADD`：将元素添加到布隆过滤器中，如果该过滤器尚不存在，则创建该过滤器。格式：`BF.ADD {key} {item}`。
-2. **`BF.MADD`** : 将一个或多个元素添加到“布隆过滤器”中，并创建一个尚不存在的过滤器。该命令的操作方式`BF.ADD`与之相同，只不过它允许多个输入并返回多个值。格式：`BF.MADD {key} {item} [item ...]` 。
+2. `BF.MADD` : 将一个或多个元素添加到“布隆过滤器”中，并创建一个尚不存在的过滤器。该命令的操作方式`BF.ADD`与之相同，只不过它允许多个输入并返回多个值。格式：`BF.MADD {key} {item} [item ...]` 。
-3. **`BF.EXISTS`** : 确定元素是否在布隆过滤器中存在。格式：`BF.EXISTS {key} {item}`。
+3. `BF.EXISTS` : 确定元素是否在布隆过滤器中存在。格式：`BF.EXISTS {key} {item}`。
-4. **`BF.MEXISTS`**：确定一个或者多个元素是否在布隆过滤器中存在格式：`BF.MEXISTS {key} {item} [item ...]`。
+4. `BF.MEXISTS`：确定一个或者多个元素是否在布隆过滤器中存在格式：`BF.MEXISTS {key} {item} [item ...]`。
 
 另外， `BF.RESERVE` 命令需要单独介绍一下：
 

docs/cs-basics/network/other-network-questions.md

@@ -104,6 +104,12 @@ tag:
 
 > 类似的问题：打开一个网页，整个过程会使用哪些协议？
 
+先来看一张图（来源于《图解 HTTP》）：
+
+<img src="https://oss.javaguide.cn/github/javaguide/url%E8%BE%93%E5%85%A5%E5%88%B0%E5%B1%95%E7%A4%BA%E5%87%BA%E6%9D%A5%E7%9A%84%E8%BF%87%E7%A8%8B.jpg" style="zoom:50%" />
+
+上图有一个错误需要注意：是 OSPF 不是 OPSF。 OSPF（Open Shortest Path First，ospf）开放最短路径优先协议, 是由 Internet 工程任务组开发的路由选择协议
+
 总体来说分为以下几个步骤:
 
 1. 在浏览器中输入指定网页的 URL。

docs/cs-basics/network/other-network-questions2.md

@@ -187,7 +187,7 @@ MAC 地址具有可携带性、永久性，身份证号永久地标识一个人
 
 最后，记住，MAC 地址有一个特殊地址：FF-FF-FF-FF-FF-FF（全 1 地址），该地址表示广播地址。
 
-### ARP 协议解决了什么问题地位如何？
+### ARP 协议解决了什么问题？
 
 ARP 协议，全称 **地址解析协议（Address Resolution Protocol）**，它解决的是网络层地址和链路层地址之间的转换问题。因为一个 IP 数据报在物理上传输的过程中，总是需要知道下一跳（物理上的下一个目的地）该去往何处，但 IP 地址属于逻辑地址，而 MAC 地址才是物理地址，ARP 协议解决了 IP 地址转 MAC 地址的一些问题。
 

docs/cs-basics/network/the-whole-process-of-accessing-web-pages.md

@@ -54,6 +54,7 @@ URL（Uniform Resource Locators），即统一资源定位器。网络上的所
 
 - [HTTP vs HTTPS（应用层）](https://javaguide.cn/cs-basics/network/http-vs-https.html)
 - [HTTP 1.0 vs HTTP 1.1（应用层）](https://javaguide.cn/cs-basics/network/http1.0-vs-http1.1.html)
+- [HTTP 常见状态码总结（应用层）](https://javaguide.cn/cs-basics/network/http-status-codes.html)
 
 ## 传输层
 

docs/database/redis/redis-questions-01.md

@@ -395,7 +395,7 @@ struct sdshdr {
 
 ### 使用 Redis 实现一个排行榜怎么做？
 
-Redis 中有一个叫做 `Sorted Set` 的数据类型经常被用在各种排行榜的场景，比如直播间送礼物的排行榜、朋友圈的微信步数排行榜、王者荣耀中的段位排行榜、话题热度排行榜等等。
+Redis 中有一个叫做 `Sorted Set` （有序集合）的数据类型经常被用在各种排行榜的场景，比如直播间送礼物的排行榜、朋友圈的微信步数排行榜、王者荣耀中的段位排行榜、话题热度排行榜等等。
 
 相关的一些 Redis 命令: `ZRANGE` (从小到大排序)、 `ZREVRANGE` （从大到小排序）、`ZREVRANK` (指定元素排名)。
 
@@ -405,6 +405,16 @@ Redis 中有一个叫做 `Sorted Set` 的数据类型经常被用在各种排行
 
 ![](https://oss.javaguide.cn/github/javaguide/database/redis/image-20220719071115140.png)
 
+### Redis 的有序集合底层为什么要用跳表，而不用平衡树、红黑树或者 B+树？
+
+这道面试题很多大厂比较喜欢问，难度还是有点大的。
+
+- 平衡树 vs 跳表：平衡树的插入、删除和查询的时间复杂度和跳表一样都是 **O(log n)**。对于范围查询来说，平衡树也可以通过中序遍历的方式达到和跳表一样的效果。但是它的每一次插入或者删除操作都需要保证整颗树左右节点的绝对平衡，只要不平衡就要通过旋转操作来保持平衡，这个过程是比较耗时的。跳表诞生的初衷就是为了克服平衡树的一些缺点。跳表使用概率平衡而不是严格强制的平衡，因此，跳表中的插入和删除算法比平衡树的等效算法简单得多，速度也快得多。
+- 红黑树 vs 跳表：相比较于红黑树来说，跳表的实现也更简单一些，不需要通过旋转和染色（红黑变换）来保证黑平衡。并且，按照区间来查找数据这个操作，红黑树的效率没有跳表高。
+- B+树 vs 跳表：B+树更适合作为数据库和文件系统中常用的索引结构之一，它的核心思想是通过可能少的 IO 定位到尽可能多的索引来获得查询数据。对于 Redis 这种内存数据库来说，它对这些并不感冒，因为 Redis 作为内存数据库它不可能存储大量的数据，所以对于索引不需要通过 B+树这种方式进行维护，只需按照概率进行随机维护即可，节约内存。而且使用跳表实现 zset 时相较前者来说更简单一些，在进行插入时只需通过索引将数据插入到链表中合适的位置再随机维护一定高度的索引即可，也不需要像 B+树那样插入时发现失衡时还需要对节点分裂与合并。
+
+另外，我还单独写了一篇文章从有序集合的基本使用到跳表的源码分析和实现，让你会对 Redis 的有序集合底层实现的跳表有着更深刻的理解和掌握 ：[Redis 为什么用跳表实现有序集合](./redis-skiplist.md)。
+
 ### Set 的应用场景是什么？
 
 Redis 中 `Set` 是一种无序集合，集合中的元素没有先后顺序但都唯一，有点类似于 Java 中的 `HashSet` 。

docs/database/redis/redis-questions-02.md

@@ -645,7 +645,13 @@ public Object getObjectInclNullById(Integer id) {
 
 **2）布隆过滤器**
 
-布隆过滤器是一个非常神奇的数据结构，通过它我们可以非常方便地判断一个给定数据是否存在于海量数据中。我们需要的就是判断 key 是否合法，有没有感觉布隆过滤器就是我们想要找的那个“人”。
+布隆过滤器是一个非常神奇的数据结构，通过它我们可以非常方便地判断一个给定数据是否存在于海量数据中。我们可以把它看作由二进制向量（或者说位数组）和一系列随机映射函数（哈希函数）两部分组成的数据结构。相比于我们平时常用的 List、Map、Set 等数据结构，它占用空间更少并且效率更高，但是缺点是其返回的结果是概率性的，而不是非常准确的。理论情况下添加到集合中的元素越多，误报的可能性就越大。并且，存放在布隆过滤器的数据不容易删除。
+
+![Bloom Filter 的简单原理示意图](https://oss.javaguide.cn/github/javaguide/cs-basics/algorithms/bloom-filter-simple-schematic-diagram.png)
+
+Bloom Filter 会使用一个较大的 bit 数组来保存所有的数据，数组中的每个元素都只占用 1 bit ，并且每个元素只能是 0 或者 1（代表 false 或者 true），这也是 Bloom Filter 节省内存的核心所在。这样来算的话，申请一个 100w 个元素的位数组只占用 1000000Bit / 8 = 125000 Byte = 125000/1024 KB ≈ 122KB 的空间。
+
+![位数组](https://oss.javaguide.cn/github/javaguide/cs-basics/algorithms/bloom-filter-bit-table.png)
 
 具体是这样做的：把所有可能存在的请求的值都存放在布隆过滤器中，当用户请求过来，先判断用户发来的请求的值是否存在于布隆过滤器中。不存在的话，直接返回请求参数错误信息给客户端，存在的话才会走下面的流程。
 
@@ -653,23 +659,7 @@ public Object getObjectInclNullById(Integer id) {
 
 ![加入布隆过滤器之后的缓存处理流程图](https://oss.javaguide.cn/github/javaguide/database/redis/redis-cache-penetration-bloom-filter.png)
 
-但是，需要注意的是布隆过滤器可能会存在误判的情况。总结来说就是：**布隆过滤器说某个元素存在，小概率会误判。布隆过滤器说某个元素不在，那么这个元素一定不在。**
+更多关于布隆过滤器的详细介绍可以看看我的这篇原创：[不了解布隆过滤器？一文给你整的明明白白！](https://javaguide.cn/cs-basics/data-structure/bloom-filter/) ，强烈推荐。
-
-_为什么会出现误判的情况呢? 我们还要从布隆过滤器的原理来说！_
-
-我们先来看一下，**当一个元素加入布隆过滤器中的时候，会进行哪些操作：**
-
-1. 使用布隆过滤器中的哈希函数对元素值进行计算，得到哈希值（有几个哈希函数得到几个哈希值）。
-2. 根据得到的哈希值，在位数组中把对应下标的值置为 1。
-
-我们再来看一下，**当我们需要判断一个元素是否存在于布隆过滤器的时候，会进行哪些操作：**
-
-1. 对给定元素再次进行相同的哈希计算；
-2. 得到值之后判断位数组中的每个元素是否都为 1，如果值都为 1，那么说明这个值在布隆过滤器中，如果存在一个值不为 1，说明该元素不在布隆过滤器中。
-
-然后，一定会出现这样一种情况：**不同的字符串可能哈希出来的位置相同。** （可以适当增加位数组大小或者调整我们的哈希函数来降低概率）
-
-更多关于布隆过滤器的内容可以看我的这篇原创：[《不了解布隆过滤器？一文给你整的明明白白！》](https://javaguide.cn/cs-basics/data-structure/bloom-filter/) ，强烈推荐，个人感觉网上应该找不到总结的这么明明白白的文章了。
 
 **3）接口限流**
 

docs/database/redis/redis-skiplist.md

@@ -0,0 +1,767 @@
+---
+title: Redis为什么用跳表实现有序集合
+category: 数据库
+tag:
+  - Redis
+---
+
+## 前言
+
+近几年针对 Redis 面试时会涉及常见数据结构的底层设计，其中就有这么一道比较有意思的面试题：“Redis 的有序集合底层为什么要用跳表，而不用平衡树、红黑树或者 B+树？”。
+
+本文就以这道大厂常问的面试题为切入点，带大家详细了解一下跳表这个数据结构。
+
+本文整体脉络如下图所示，笔者会从有序集合的基本使用到跳表的源码分析和实现，让你会对 Redis 的有序集合底层实现的跳表有着更深刻的理解和掌握。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005468.png)
+
+## 跳表在 Redis 中的运用
+
+这里我们需要先了解一下 Redis 用到跳表的数据结构有序集合的使用，Redis 有个比较常用的数据结构叫**有序集合(sorted set，简称 zset)**，正如其名它是一个可以保证有序且元素唯一的集合，所以它经常用于排行榜等需要进行统计排列的场景。
+
+这里我们通过命令行的形式演示一下排行榜的实现，可以看到笔者分输入 3 名用户：**xiaoming**、**xiaohong**、**xiaowang**，它们的**score**分别是 60、80、60，最终按照成绩升级降序排列。
+
+```bash
+
+127.0.0.1:6379> zadd rankList 60 xiaoming
+(integer) 1
+127.0.0.1:6379> zadd rankList 80 xiaohong
+(integer) 1
+127.0.0.1:6379> zadd rankList 60 xiaowang
+(integer) 1
+
+# 返回有序集中指定区间内的成员，通过索引，分数从高到低
+127.0.0.1:6379> ZREVRANGE rankList 0 100 WITHSCORES
+1) "xiaohong"
+2) "80"
+3) "xiaowang"
+4) "60"
+5) "xiaoming"
+6) "60"
+```
+
+此时我们通过 `object` 指令查看 zset 的数据结构，可以看到当前有序集合存储的还是是**ziplist(压缩列表)**。
+
+```bash
+127.0.0.1:6379> object encoding rankList
+"ziplist"
+```
+
+因为设计者考虑到 Redis 数据存放于内存，为了节约宝贵的内存空间在有序集合在元素小于 64 字节且个数小于 128 的时候，会使用 ziplist，而这个阈值的默认值的设置就来自下面这两个配置项。
+
+```bash
+zset-max-ziplist-value 64
+zset-max-ziplist-entries 128
+```
+
+一旦有序集合中的某个元素超出这两个其中的一个阈值它就会转为 **skiplist**（实际是 dict+skiplist，还会借用字典来提高获取指定元素的效率）。
+
+我们不妨在添加一个大于 64 字节的元素，可以看到有序集合的底层存储转为 skiplist。
+
+```bash
+127.0.0.1:6379> zadd rankList 90 yigemingzihuichaoguo64zijiedeyonghumingchengyongyuceshitiaobiaodeshijiyunyong
+(integer) 1
+
+# 超过阈值，转为跳表
+127.0.0.1:6379> object encoding rankList
+"skiplist"
+```
+
+也就是说，ZSet 有两种不同的实现，分别是 ziplist 和 skiplist，具体使用哪种结构进行存储的规则如下：
+
+- 当有序集合对象同时满足以下两个条件时，使用 ziplist：
+  1. ZSet 保存的键值对数量少于 128 个；
+  2. 每个元素的长度小于 64 字节。
+- 如果不满足上述两个条件，那么使用 skiplist 。
+
+## 手写一个跳表
+
+为了更好的回答上述问题以及更好的理解和掌握跳表，这里可以通过手写一个简单的跳表的形式来帮助读者理解跳表这个数据结构。
+
+我们都知道有序链表在添加、查询、删除的平均时间复杂都都是**O(n)**即线性增长，所以一旦节点数量达到一定体量后其性能表现就会非常差劲。而跳表我们完全可以理解为在原始链表基础上，建立多级索引，通过多级索引检索定位将增删改查的时间复杂度变为**O(log n)**。
+
+可能这里说的有些抽象，我们举个例子，以下图跳表为例，其原始链表存储按序存储 1-10，有 2 级索引，每级索引的索引个数都是基于下层元素个数的一半。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005436.png)
+
+假如我们需要查询元素 6，其工作流程如下：
+
+1. 从 2 级索引开始，先来到节点 4。
+2. 查看 4 的后继节点，是 8 的 2 级索引，这个值大于 6，说明 2 级索引后续的索引都是大于 6 的，没有再往后搜寻的必要，我们索引向下查找。
+3. 来到 4 的 1 级索引，比对其后继节点为 6，查找结束。
+
+相较于原始有序链表需要 6 次，我们的跳表通过建立多级索引，我们只需两次就直接定位到了目标元素，其查寻的复杂度被直接优化为**O(log n)**。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005524.png)
+
+对应的添加也是一个道理，假如我们需要在这个有序集合中添加一个元素 7，那么我们就需要通过跳表找到**小于元素 7 的最大值**，也就是下图元素 6 的位置，将其插入到元素 6 的后面，让元素 6 的索引指向新插入的节点 7，其工作流程如下：
+
+1. 从 2 级索引开始定位到了元素 4 的索引。
+2. 查看索引 4 的后继索引为 8，索引向下推进。
+3. 来到 1 级索引，发现索引 4 后继索引为 6，小于插入元素 7，指针推进到索引 6 位置。
+4. 继续比较 6 的后继节点为索引 8，大于元素 7，索引继续向下。
+5. 最终我们来到 6 的原始节点，发现其后继节点为 7，指针没有继续向下的空间，自此我们可知元素 6 就是小于插入元素 7 的最大值，于是便将元素 7 插入。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005480.png)
+
+这里我们又面临一个问题，我们是否需要为元素 7 建立索引，索引多高合适？
+
+我们上文提到，理想情况是每一层索引是下一层元素个数的二分之一，假设我们的总共有 16 个元素，对应各级索引元素个数应该是：
+
+```bash
+1. 一级索引:16/2=8
+2. 二级索引:8/2 =4
+3. 三级索引:4/2=2
+```
+
+由此我们用数学归纳法可知：
+
+```bash
+1. 一级索引:16/2=16/2^1=8
+2. 二级索引:8/2 => 16/2^2 =4
+3. 三级索引:4/2=>16/2^3=2
+```
+
+假设元素个数为 n，那么对应 k 层索引的元素个数 r 计算公式为:
+
+```bash
+r=n/2^k
+```
+
+同理我们再来推断以下索引的最大高度，一般来说最高级索引的元素个数为 2，我们设元素总个数为 n，索引高度为 h，代入上述公式可得：
+
+```bash
+2= n/2^h
+=> 2*2^h=n
+=> 2^(h+1)=n
+=> h+1=log2^n
+=> h=log2^n -1
+```
+
+而 Redis 又是内存数据库，我们假设元素最大个数是**65536**，我们把**65536**代入上述公式可知最大高度为 16。所以我们建议添加一个元素后为其建立的索引高度不超过 16。
+
+因为我们要求尽可能保证每一个上级索引都是下级索引的一半，在实现高度生成算法时，我们可以这样设计：
+
+1. 跳表的高度计算从原始链表开始，即默认情况下插入的元素的高度为 1，代表没有索引，只有元素节点。
+2. 设计一个为插入元素生成节点索引高度 level 的方法。
+3. 进行一次随机运算，随机数值范围为 0-1 之间。
+4. 如果随机数大于 0.5 则为当前元素添加一级索引，自此我们保证生成一级索引的概率为**50%**，这也就保证了 1 级索引理想情况下只有一半的元素会生成索引。
+5. 同理后续每次随机算法得到的值大于 0.5 时，我们的索引高度就加 1，这样就可以保证节点生成的 2 级索引概率为**25%**，3 级索引为**12.5%**……
+
+我们回过头，上述插入 7 之后，我们通过随机算法得到 2，即要为其建立 1 级索引：
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005505.png)
+
+最后我们再来说说删除，假设我们这里要删除元素 10，我们必须定位到当前跳表**各层**元素小于 10 的最大值，索引执行步骤为：
+
+1. 2 级索引 4 的后继节点为 8，指针推进。
+2. 索引 8 无后继节点，该层无要删除的元素，指针直接向下。
+3. 1 级索引 8 后继节点为 10，说明 1 级索引 8 在进行删除时需要将自己的指针和 1 级索引 10 断开联系，将 10 删除。
+4. 1 级索引完成定位后，指针向下，后继节点为 9，指针推进。
+5. 9 的后继节点为 10，同理需要让其指向 null，将 10 删除。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005503.png)
+
+### 模板定义
+
+有了整体的思路之后，我们可以开始实现一个跳表了，首先定义一下跳表中的节点**Node**，从上文的演示中可以看出每一个**Node**它都包含以下几个元素：
+
+1. 存储的**value**值。
+2. 后继节点的地址。
+3. 多级索引。
+
+为了更方便统一管理**Node**后继节点地址和多级索引指向的元素地址，笔者在**Node**中设置了一个**forwards**数组，用于记录原始链表节点的后继节点和多级索引的后继节点指向。
+
+以下图为例，我们**forwards**数组长度为 5，其中**索引 0**记录的是原始链表节点的后继节点地址，而其余自底向上表示从 1 级索引到 4 级索引的后继节点指向。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005347.png)
+
+于是我们的就有了这样一个代码定义，可以看出笔者对于数组的长度设置为固定的 16**(上文的推算最大高度建议是 16)**，默认**data**为-1，节点最大高度**maxLevel**初始化为 1，注意这个**maxLevel**的值代表原始链表加上索引的总高度。
+
+```java
+/**
+ * 跳表索引最大高度为16
+ */
+private static final int MAX_LEVEL = 16;
+
+class Node {
+    private int data = -1;
+    private Node[] forwards = new Node[MAX_LEVEL];
+    private int maxLevel = 0;
+
+}
+```
+
+### 元素添加
+
+定义好节点之后，我们先实现以下元素的添加，添加元素时首先自然是设置**data**这一步我们直接根据将传入的**value**设置到**data**上即可。
+
+然后就是高度**maxLevel**的设置 ，我们在上文也已经给出了思路，默认高度为 1，即只有一个原始链表节点，通过随机算法每次大于 0.5 索引高度加 1，由此我们得出高度计算的算法`randomLevel()`：
+
+```java
+/**
+ * 理论来讲，一级索引中元素个数应该占原始数据的 50%，二级索引中元素个数占 25%，三级索引12.5% ，一直到最顶层。
+ * 因为这里每一层的晋升概率是 50%。对于每一个新插入的节点，都需要调用 randomLevel 生成一个合理的层数。
+ * 该 randomLevel 方法会随机生成 1~MAX_LEVEL 之间的数，且 ：
+ * 50%的概率返回 1
+ * 25%的概率返回 2
+ *  12.5%的概率返回 3 ...
+ * @return
+ */
+private int randomLevel() {
+    int level = 1;
+    while (Math.random() > PROB && level < MAX_LEVEL) {
+        ++level;
+    }
+    return level;
+}
+```
+
+然后再设置当前要插入的**Node**和**Node**索引的后继节点地址，这一步稍微复杂一点，我们假设当前节点的高度为 4，即 1 个节点加 3 个索引，所以我们创建一个长度为 4 的数组**maxOfMinArr** ，遍历各级索引节点中小于当前**value**的最大值。
+
+假设我们要插入的**value**为 5，我们的数组查找结果当前节点的前驱节点和 1 级索引、2 级索引的前驱节点都为 4，三级索引为空。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005299.png)
+
+然后我们基于这个数组**maxOfMinArr** 定位到各级的后继节点，让插入的元素 5 指向这些后继节点，而**maxOfMinArr**指向 5，结果如下图：
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005369.png)
+
+转化成代码就是下面这个形式，是不是很简单呢？我们继续：
+
+```java
+/**
+ * 默认情况下的高度为1，即只有自己一个节点
+ */
+private int leveCount = 1;
+
+/**
+ * 跳表最底层的节点，即头节点
+ */
+private Node h = new Node();
+
+public void add(int value) {
+
+    //随机生成高度
+    int level = randomLevel();
+
+    Node newNode = new Node();
+    newNode.data = value;
+    newNode.maxLevel = level;
+
+    //创建一个node数组，用于记录小于当前value的最大值
+    Node[] maxOfMinArr = new Node[level];
+    //默认情况下指向头节点
+    for (int i = 0; i < level; i++) {
+        maxOfMinArr[i] = h;
+    }
+
+    //基于上述结果拿到当前节点的后继节点
+    Node p = h;
+    for (int i = level - 1; i >= 0; i--) {
+        while (p.forwards[i] != null && p.forwards[i].data < value) {
+            p = p.forwards[i];
+        }
+        maxOfMinArr[i] = p;
+    }
+
+    //更新前驱节点的后继节点为当前节点newNode
+    for (int i = 0; i < level; i++) {
+        newNode.forwards[i] = maxOfMinArr[i].forwards[i];
+        maxOfMinArr[i].forwards[i] = newNode;
+    }
+
+    //如果当前newNode高度大于跳表最高高度则更新leveCount
+    if (leveCount < level) {
+        leveCount = level;
+    }
+
+}
+```
+
+### 元素查询
+
+查询逻辑比较简单，从跳表最高级的索引开始定位找到小于要查的 value 的最大值，以下图为例，我们希望查找到节点 8：
+
+1. 跳表的 3 级索引首先找找到 5 的索引，5 的 3 级索引**forwards[3]**指向空，索引直接向下。
+2. 来到 5 的 2 级索引，其后继**forwards[2]**指向 8，继续向下。
+3. 5 的 1 级索引**forwards[1]**指向索引 6，继续向前。
+4. 索引 6 的**forwards[1]**指向索引 8，继续向下。
+5. 我们在原始节点向前找到节点 7。
+6. 节点 7 后续就是节点 8，继续向前为节点 8，无法继续向下，结束搜寻。
+7. 判断 7 的前驱，等于 8，查找结束。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005323.png)
+
+所以我们的代码实现也很上述步骤差不多，从最高级索引开始向前查找，如果不为空且小于要查找的值，则继续向前搜寻，遇到不小于的节点则继续向下，如此往复，直到得到当前跳表中小于查找值的最大节点，查看其前驱是否等于要查找的值：
+
+```java
+public Node get(int value) {
+    Node p = h;
+    //找到小于value的最大值
+    for (int i = leveCount - 1; i >= 0; i--) {
+        while (p.forwards[i] != null && p.forwards[i].data < value) {
+            p = p.forwards[i];
+        }
+    }
+    //如果p的前驱节点等于value则直接返回
+    if (p.forwards[0] != null && p.forwards[0].data == value) {
+        return p.forwards[0];
+    }
+
+    return null;
+}
+```
+
+### 元素删除
+
+最后是删除逻辑，需要查找各层级小于要删除节点的最大值，假设我们要删除 10：
+
+1. 3 级索引得到小于 10 的最大值为 5，继续向下。
+2. 2 级索引从索引 5 开始查找，发现小于 10 的最大值为 8，继续向下。
+3. 同理 1 级索引得到 8，继续向下。
+4. 原始节点找到 9。
+5. 从最高级索引开始，查看每个小于 10 的节点后继节点是否为 10，如果等于 10，则让这个节点指向 10 的后继节点，将节点 10 及其索引交由 GC 回收。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005350.png)
+
+```java
+/**
+ * 删除
+ *
+ * @param value
+ */
+public void delete(int value) {
+    Node p = h;
+    //找到各级节点小于value的最大值
+    Node[] updateArr = new Node[leveCount];
+    for (int i = leveCount - 1; i >= 0; i--) {
+        while (p.forwards[i] != null && p.forwards[i].data < value) {
+            p = p.forwards[i];
+        }
+        updateArr[i] = p;
+    }
+    //查看原始层节点前驱是否等于value，若等于则说明存在要删除的值
+    if (p.forwards[0] != null && p.forwards[0].data == value) {
+        //从最高级索引开始查看其前驱是否等于value，若等于则将当前节点指向value节点的后继节点
+        for (int i = leveCount - 1; i >= 0; i--) {
+            if (updateArr[i].forwards[i] != null && updateArr[i].forwards[i].data == value) {
+                updateArr[i].forwards[i] = updateArr[i].forwards[i].forwards[i];
+            }
+        }
+    }
+
+    //从最高级开始查看是否有一级索引为空，若为空则层级减1
+    while (leveCount > 1 && h.forwards[leveCount] == null) {
+        leveCount--;
+    }
+
+}
+```
+
+### 完整代码以及测试
+
+完整代码如下，读者可自行参阅:
+
+```java
+public class SkipList {
+
+    /**
+     * 跳表索引最大高度为16
+     */
+    private static final int MAX_LEVEL = 16;
+
+    /**
+     * 每个节点添加一层索引高度的概率为二分之一
+     */
+    private static final float PROB = 0.5 f;
+
+    /**
+     * 默认情况下的高度为1，即只有自己一个节点
+     */
+    private int leveCount = 1;
+
+    /**
+     * 跳表最底层的节点，即头节点
+     */
+    private Node h = new Node();
+
+    public SkipList() {}
+
+    public class Node {
+        private int data = -1;
+        /**
+         *
+         */
+        private Node[] forwards = new Node[MAX_LEVEL];
+        private int maxLevel = 0;
+
+        @Override
+        public String toString() {
+            return "Node{" +
+                "data=" + data +
+                ", maxLevel=" + maxLevel +
+                '}';
+        }
+    }
+
+    public void add(int value) {
+
+        //随机生成高度
+        int level = randomLevel();
+
+        Node newNode = new Node();
+        newNode.data = value;
+        newNode.maxLevel = level;
+
+        //创建一个node数组，用于记录小于当前value的最大值
+        Node[] maxOfMinArr = new Node[level];
+        //默认情况下指向头节点
+        for (int i = 0; i < level; i++) {
+            maxOfMinArr[i] = h;
+        }
+
+        //基于上述结果拿到当前节点的后继节点
+        Node p = h;
+        for (int i = level - 1; i >= 0; i--) {
+            while (p.forwards[i] != null && p.forwards[i].data < value) {
+                p = p.forwards[i];
+            }
+            maxOfMinArr[i] = p;
+        }
+
+        //更新前驱节点的后继节点为当前节点newNode
+        for (int i = 0; i < level; i++) {
+            newNode.forwards[i] = maxOfMinArr[i].forwards[i];
+            maxOfMinArr[i].forwards[i] = newNode;
+        }
+
+        //如果当前newNode高度大于跳表最高高度则更新leveCount
+        if (leveCount < level) {
+            leveCount = level;
+        }
+
+    }
+
+    /**
+     * 理论来讲，一级索引中元素个数应该占原始数据的 50%，二级索引中元素个数占 25%，三级索引12.5% ，一直到最顶层。
+     * 因为这里每一层的晋升概率是 50%。对于每一个新插入的节点，都需要调用 randomLevel 生成一个合理的层数。
+     * 该 randomLevel 方法会随机生成 1~MAX_LEVEL 之间的数，且 ：
+     * 50%的概率返回 1
+     * 25%的概率返回 2
+     *  12.5%的概率返回 3 ...
+     * @return
+     */
+    private int randomLevel() {
+        int level = 1;
+        while (Math.random() > PROB && level < MAX_LEVEL) {
+            ++level;
+        }
+        return level;
+    }
+
+    public Node get(int value) {
+        Node p = h;
+        //找到小于value的最大值
+        for (int i = leveCount - 1; i >= 0; i--) {
+            while (p.forwards[i] != null && p.forwards[i].data < value) {
+                p = p.forwards[i];
+            }
+        }
+        //如果p的前驱节点等于value则直接返回
+        if (p.forwards[0] != null && p.forwards[0].data == value) {
+            return p.forwards[0];
+        }
+
+        return null;
+    }
+
+    /**
+     * 删除
+     *
+     * @param value
+     */
+    public void delete(int value) {
+        Node p = h;
+        //找到各级节点小于value的最大值
+        Node[] updateArr = new Node[leveCount];
+        for (int i = leveCount - 1; i >= 0; i--) {
+            while (p.forwards[i] != null && p.forwards[i].data < value) {
+                p = p.forwards[i];
+            }
+            updateArr[i] = p;
+        }
+        //查看原始层节点前驱是否等于value，若等于则说明存在要删除的值
+        if (p.forwards[0] != null && p.forwards[0].data == value) {
+            //从最高级索引开始查看其前驱是否等于value，若等于则将当前节点指向value节点的后继节点
+            for (int i = leveCount - 1; i >= 0; i--) {
+                if (updateArr[i].forwards[i] != null && updateArr[i].forwards[i].data == value) {
+                    updateArr[i].forwards[i] = updateArr[i].forwards[i].forwards[i];
+                }
+            }
+        }
+
+        //从最高级开始查看是否有一级索引为空，若为空则层级减1
+        while (leveCount > 1 && h.forwards[leveCount] == null) {
+            leveCount--;
+        }
+
+    }
+
+    public void printAll() {
+        Node p = h;
+        //基于最底层的非索引层进行遍历，只要后继节点不为空，则速速出当前节点，并移动到后继节点
+        while (p.forwards[0] != null) {
+            System.out.println(p.forwards[0]);
+            p = p.forwards[0];
+        }
+
+    }
+
+}
+```
+
+对应测试代码和输出结果如下：
+
+```java
+public static void main(String[] args) {
+        SkipList skipList = new SkipList();
+        for (int i = 0; i < 24; i++) {
+            skipList.add(i);
+        }
+
+        System.out.println("**********输出添加结果**********");
+        skipList.printAll();
+
+        SkipList.Node node = skipList.get(22);
+        System.out.println("**********查询结果:" + node+" **********");
+
+        skipList.delete(22);
+        System.out.println("**********删除结果**********");
+        skipList.printAll();
+
+
+    }
+```
+
+输出结果:
+
+```bash
+**********输出添加结果**********
+Node{data=0, maxLevel=2}
+Node{data=1, maxLevel=3}
+Node{data=2, maxLevel=1}
+Node{data=3, maxLevel=1}
+Node{data=4, maxLevel=2}
+Node{data=5, maxLevel=2}
+Node{data=6, maxLevel=2}
+Node{data=7, maxLevel=2}
+Node{data=8, maxLevel=4}
+Node{data=9, maxLevel=1}
+Node{data=10, maxLevel=1}
+Node{data=11, maxLevel=1}
+Node{data=12, maxLevel=1}
+Node{data=13, maxLevel=1}
+Node{data=14, maxLevel=1}
+Node{data=15, maxLevel=3}
+Node{data=16, maxLevel=4}
+Node{data=17, maxLevel=2}
+Node{data=18, maxLevel=1}
+Node{data=19, maxLevel=1}
+Node{data=20, maxLevel=1}
+Node{data=21, maxLevel=3}
+Node{data=22, maxLevel=1}
+Node{data=23, maxLevel=1}
+**********查询结果:Node{data=22, maxLevel=1} **********
+**********删除结果**********
+Node{data=0, maxLevel=2}
+Node{data=1, maxLevel=3}
+Node{data=2, maxLevel=1}
+Node{data=3, maxLevel=1}
+Node{data=4, maxLevel=2}
+Node{data=5, maxLevel=2}
+Node{data=6, maxLevel=2}
+Node{data=7, maxLevel=2}
+Node{data=8, maxLevel=4}
+Node{data=9, maxLevel=1}
+Node{data=10, maxLevel=1}
+Node{data=11, maxLevel=1}
+Node{data=12, maxLevel=1}
+Node{data=13, maxLevel=1}
+Node{data=14, maxLevel=1}
+Node{data=15, maxLevel=3}
+Node{data=16, maxLevel=4}
+Node{data=17, maxLevel=2}
+Node{data=18, maxLevel=1}
+Node{data=19, maxLevel=1}
+Node{data=20, maxLevel=1}
+Node{data=21, maxLevel=3}
+Node{data=23, maxLevel=1}
+```
+
+## 和其余三种数据结构的比较
+
+最后，我们再来回答一下文章开头的那道面试题: “Redis 的有序集合底层为什么要用跳表，而不用平衡树、红黑树或者 B+树？”。
+
+### 平衡树 vs 跳表
+
+先来说说它和平衡树的比较，平衡树我们又会称之为 **AVL 树**，是一个严格的平衡二叉树，平衡条件必须满足（所有节点的左右子树高度差不超过 1，即平衡因子为范围为 `[-1,1]`）。平衡树的插入、删除和查询的时间复杂度和跳表一样都是 **O(log n)**。
+
+对于范围查询来说，它也可以通过中序遍历的方式达到和跳表一样的效果。但是它的每一次插入或者删除操作都需要保证整颗树左右节点的绝对平衡，只要不平衡就要通过旋转操作来保持平衡，这个过程是比较耗时的。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005312.png)
+
+跳表诞生的初衷就是为了克服平衡树的一些缺点，跳表的发明者在论文[《Skip lists: a probabilistic alternative to balanced trees》](https://15721.courses.cs.cmu.edu/spring2018/papers/08-oltpindexes1/pugh-skiplists-cacm1990.pdf)中有详细提到：
+
+![](https://oss.javaguide.cn/github/javaguide/database/redis/skiplist-a-probabilistic-alternative-to-balanced-trees.png)
+
+> Skip lists are a data structure that can be used in place of balanced trees. Skip lists use probabilistic balancing rather than strictly enforced balancing and as a result the algorithms for insertion and deletion in skip lists are much simpler and significantly faster than equivalent algorithms for balanced trees.
+>
+> 跳表是一种可以用来代替平衡树的数据结构。跳表使用概率平衡而不是严格强制的平衡，因此，跳表中的插入和删除算法比平衡树的等效算法简单得多，速度也快得多。
+
+笔者这里也贴出了 AVL 树插入操作的核心代码，可以看出每一次添加操作都需要进行一次递归定位插入位置，然后还需要根据回溯到根节点检查沿途的各层节点是否失衡，再通过旋转节点的方式进行调整。
+
+```java
+// 向二分搜索树中添加新的元素(key, value)
+public void add(K key, V value) {
+    root = add(root, key, value);
+}
+
+// 向以node为根的二分搜索树中插入元素(key, value)，递归算法
+// 返回插入新节点后二分搜索树的根
+private Node add(Node node, K key, V value) {
+
+    if (node == null) {
+        size++;
+        return new Node(key, value);
+    }
+
+    if (key.compareTo(node.key) < 0)
+        node.left = add(node.left, key, value);
+    else if (key.compareTo(node.key) > 0)
+        node.right = add(node.right, key, value);
+    else // key.compareTo(node.key) == 0
+        node.value = value;
+
+    node.height = 1 + Math.max(getHeight(node.left), getHeight(node.right));
+
+    int balanceFactor = getBalanceFactor(node);
+
+    // LL型需要右旋
+    if (balanceFactor > 1 && getBalanceFactor(node.left) >= 0) {
+        return rightRotate(node);
+    }
+
+    //RR型失衡需要左旋
+    if (balanceFactor < -1 && getBalanceFactor(node.right) <= 0) {
+        return leftRotate(node);
+    }
+
+    //LR需要先左旋成LL型，然后再右旋
+    if (balanceFactor > 1 && getBalanceFactor(node.left) < 0) {
+        node.left = leftRotate(node.left);
+        return rightRotate(node);
+    }
+
+    //RL
+    if (balanceFactor < -1 && getBalanceFactor(node.right) > 0) {
+        node.right = rightRotate(node.right);
+        return leftRotate(node);
+    }
+    return node;
+}
+```
+
+### 红黑树 vs 跳表
+
+红黑树（Red Black Tree）也是一种自平衡二叉查找树，它的查询性能略微逊色于 AVL 树，但插入和删除效率更高。红黑树的插入、删除和查询的时间复杂度和跳表一样都是 **O(log n)**。
+
+红黑树是一个**黑平衡树**，即从任意节点到另外一个叶子叶子节点，它所经过的黑节点是一样的。当对它进行插入操作时，需要通过旋转和染色（红黑变换）来保证黑平衡。不过，相较于 AVL 树为了维持平衡的开销要小一些。关于红黑树的详细介绍，可以查看这篇文章：[红黑树](https://javaguide.cn/cs-basics/data-structure/red-black-tree.html)。
+
+相比较于红黑树来说，跳表的实现也更简单一些。并且，按照区间来查找数据这个操作，红黑树的效率没有跳表高。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005709.png)
+
+对应红黑树添加的核心代码如下，读者可自行参阅理解：
+
+```java
+private Node < K, V > add(Node < K, V > node, K key, V val) {
+
+    if (node == null) {
+        size++;
+        return new Node(key, val);
+
+    }
+
+    if (key.compareTo(node.key) < 0) {
+        node.left = add(node.left, key, val);
+    } else if (key.compareTo(node.key) > 0) {
+        node.right = add(node.right, key, val);
+    } else {
+        node.val = val;
+    }
+
+    //左节点不为红，右节点为红，左旋
+    if (isRed(node.right) && !isRed(node.left)) {
+        node = leftRotate(node);
+    }
+
+    //左链右旋
+    if (isRed(node.left) && isRed(node.left.left)) {
+        node = rightRotate(node);
+    }
+
+    //颜色翻转
+    if (isRed(node.left) && isRed(node.right)) {
+        flipColors(node);
+    }
+
+    return node;
+}
+```
+
+### B+树 vs 跳表
+
+想必使用 MySQL 的读者都知道 B+树这个数据结构，B+树是一种常用的数据结构，具有以下特点：
+
+1. **多叉树结构**：它是一棵多叉树，每个节点可以包含多个子节点，减小了树的高度，查询效率高。
+2. **存储效率高**:其中非叶子节点存储多个 key，叶子节点存储 value，使得每个节点更够存储更多的键，根据索引进行范围查询时查询效率更高。-
+3. **平衡性**：它是绝对的平衡，即树的各个分支高度相差不大，确保查询和插入时间复杂度为**O(log n)**。
+4. **顺序访问**：叶子节点间通过链表指针相连，范围查询表现出色。
+5. **数据均匀分布**：B+树插入时可能会导致数据重新分布，使得数据在整棵树分布更加均匀，保证范围查询和删除效率。
+
+![](https://oss.javaguide.cn/javaguide/database/redis/skiplist/202401222005649.png)
+
+所以，B+树更适合作为数据库和文件系统中常用的索引结构之一，它的核心思想是通过可能少的 IO 定位到尽可能多的索引来获得查询数据。对于 Redis 这种内存数据库来说，它对这些并不感冒，因为 Redis 作为内存数据库它不可能存储大量的数据，所以对于索引不需要通过 B+树这种方式进行维护，只需按照概率进行随机维护即可，节约内存。而且使用跳表实现 zset 时相较前者来说更简单一些，在进行插入时只需通过索引将数据插入到链表中合适的位置再随机维护一定高度的索引即可，也不需要像 B+树那样插入时发现失衡时还需要对节点分裂与合并。
+
+### Redis 作者给出的理由
+
+当然我们也可以通过 Redis 的作者自己给出的理由:
+
+> There are a few reasons:
+> 1、They are not very memory intensive. It's up to you basically. Changing parameters about the probability of a node to have a given number of levels will make then less memory intensive than btrees.
+> 2、A sorted set is often target of many ZRANGE or ZREVRANGE operations, that is, traversing the skip list as a linked list. With this operation the cache locality of skip lists is at least as good as with other kind of balanced trees.
+> 3、They are simpler to implement, debug, and so forth. For instance thanks to the skip list simplicity I received a patch (already in Redis master) with augmented skip lists implementing ZRANK in O(log(N)). It required little changes to the code.
+
+翻译过来的意思就是:
+
+> 有几个原因：
+>
+> 1、它们不是很占用内存。这主要取决于你。改变节点拥有给定层数的概率的参数，会使它们比 B 树更节省内存。
+>
+> 2、有序集合经常是许多 ZRANGE 或 ZREVRANGE 操作的目标，也就是说，以链表的方式遍历跳表。通过这种操作，跳表的缓存局部性至少和其他类型的平衡树一样好。
+>
+> 3、它们更容易实现、调试等等。例如，由于跳表的简单性，我收到了一个补丁（已经在 Redis 主分支中），用增强的跳表实现了 O(log(N))的 ZRANK。它只需要对代码做很少的修改。
+
+## 小结
+
+本文通过大量篇幅介绍跳表的工作原理和实现，帮助读者更进一步的熟悉跳表这一数据结构的优劣，最后再结合各个数据结构操作的特点进行比对，从而帮助读者更好的理解这道面试题，建议读者实现理解跳表时，尽可能配合执笔模拟来了解跳表的增删改查详细过程。
+
+## 参考
+
+- 为啥 redis 使用跳表(skiplist)而不是使用 red-black？:<https://www.zhihu.com/question/20202931/answer/16086538>
+- Skip List--跳表（全网最详细的跳表文章没有之一）:<https://www.jianshu.com/p/9d8296562806>
+- Redis 对象与底层数据结构详解:<https://blog.csdn.net/shark_chili3007/article/details/104171986>
+- Redis 有序集合(sorted set):<https://www.runoob.com/redis/redis-sorted-sets.html>
+- 红黑树和跳表比较:<https://zhuanlan.zhihu.com/p/576984787>
+- 为什么 redis 的 zset 用跳跃表而不用 b+ tree？:<https://blog.csdn.net/f80407515/article/details/129136998>

docs/home.md

@@ -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)
+
 ## 分布式
 
 ### 理论&算法&协议

docs/java/concurrent/java-concurrent-questions-01.md

@@ -167,6 +167,16 @@ JDK 1.2 之前，Java 线程是基于绿色线程（Green Threads）实现的，
 
 因此，对于单核 CPU 来说，如果任务是 CPU 密集型的，那么开很多线程会影响效率；如果任务是 IO 密集型的，那么开很多线程会提高效率。当然，这里的“很多”也要适度，不能超过系统能够承受的上限。
 
+## 如何创建线程？
+
+一般来说，创建线程有很多种方式，例如继承`Thread`类、实现`Runnable`接口、实现`Callable`接口、使用线程池、使用`CompletableFuture`类等等。
+
+不过，这些方式其实并没有真正创建出线程。准确点来说，这些都属于是在 Java 代码中使用多线程的方法。
+
+严格来说，Java 就只有一种方式可以创建线程，那就是通过`new Thread().start()`创建。不管是哪种方式，最终还是依赖于`new Thread().start()`。
+
+关于这个问题的详细分析可以查看这篇文章：[大家都说 Java 有三种创建线程的方式！并发编程中的惊天骗局！](https://mp.weixin.qq.com/s/NspUsyhEmKnJ-4OprRFp9g)。
+
 ## 说说线程的生命周期和状态?
 
 Java 线程在运行的生命周期中的指定时刻只可能处于下面 6 种不同状态的其中一个状态：

docs/java/concurrent/optimistic-lock-and-pessimistic-lock.md

@@ -154,7 +154,7 @@ CAS 经常会用到自旋操作来进行重试，也就是不成功就一直循
 如果 JVM 能支持处理器提供的 pause 指令那么效率会有一定的提升，pause 指令有两个作用：
 
 1. 可以延迟流水线执行指令，使 CPU 不会消耗过多的执行资源，延迟的时间取决于具体实现的版本，在一些处理器上延迟时间是零。
-2. 可以避免在退出循环的时候因内存顺序冲而引起 CPU 流水线被清空，从而提高 CPU 的执行效率。
+2. 可以避免在退出循环的时候因内存顺序冲突而引起 CPU 流水线被清空，从而提高 CPU 的执行效率。
 
 ### 只能保证一个共享变量的原子操作
 

docs/mq/kafka/kafka-basis.md

@@ -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 Partition（ISR）都接收成功时进行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 Number，PID是Producer的唯一ID，Sequence 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

docs/system-design/security/sentive-words-filter.md

@@ -13,7 +13,7 @@ tag:
 
 ### Trie 树
 
-**Trie 树** 也称为字典树、单词查找树，哈系树的一种变种，通常被用于字符串匹配，用来解决在一组字符串集合中快速查找某个字符串的问题。像浏览器搜索的关键词提示一般就是基于 Trie 树来做的。
+**Trie 树** 也称为字典树、单词查找树，哈系树的一种变种，通常被用于字符串匹配，用来解决在一组字符串集合中快速查找某个字符串的问题。像浏览器搜索的关键词提示就可以基于 Trie 树来做的。
 
 ![浏览器 Trie 树效果展示](https://oss.javaguide.cn/github/javaguide/system-design/security/brower-trie.png)
 
@@ -48,11 +48,17 @@ assertEquals("{Abi=doctor, Abigail=student}", trie.prefixMap("Abi").toString());
 assertEquals("{Chris=doctor, Christina=student}", trie.prefixMap("Chr").toString());
 ```
 
+Trie 树是一种利用空间换时间的数据结构，占用的内存会比较大。也正是因为这个原因，实际工程项目中都是使用的改进版 Trie 树例如双数组 Trie 树（Double-Array Trie，DAT）。
+
+DAT 的设计者是日本的 Aoe Jun-ichi，Mori Akira 和 Sato Takuya，他们在 1989 年发表了一篇论文[《An Efficient Implementation of Trie Structures》](https://www.co-ding.com/assets/pdf/dat.pdf)，详细介绍了 DAT 的构造和应用，原作者写的示例代码地址：<https://github.com/komiya-atsushi/darts-java/blob/e2986a55e648296cc0a6244ae4a2e457cd89fb82/src/main/java/darts/DoubleArrayTrie.java>。相比较于 Trie 树，DAT 的内存占用极低，可以达到 Trie 树内存的 1%左右。DAT 在中文分词、自然语言处理、信息检索等领域有广泛的应用，是一种非常优秀的数据结构。
+
+### AC 自动机
+
 Aho-Corasick（AC）自动机是一种建立在 Trie 树上的一种改进算法，是一种多模式匹配算法，由贝尔实验室的研究人员 Alfred V. Aho 和 Margaret J.Corasick 发明。
 
-AC 自动机算法使用 Trie 树来存放模式串的前缀，通过失败匹配指针（失配指针）来处理匹配失败的跳转。
+AC 自动机算法使用 Trie 树来存放模式串的前缀，通过失败匹配指针（失配指针）来处理匹配失败的跳转。关于 AC 自动机的详细介绍，可以查看这篇文章：[地铁十分钟 | AC 自动机](https://zhuanlan.zhihu.com/p/146369212)。
 
-相关阅读：[地铁十分钟 | AC 自动机](https://zhuanlan.zhihu.com/p/146369212)
+如果使用上面提到的 DAT 来表示 AC 自动机 ，就可以兼顾两者的优点，得到一种高效的多模式匹配算法。Github 上已经有了开源 Java 实现版本：<https://github.com/hankcs/AhoCorasickDoubleArrayTrie> 。
 
 ### DFA