## 前言

数据库和缓存（如`redis`、`memcached`等）双写数据一致性问题，是一个跟开发语言无关的公共问题。尤其是在高并发的场景下，这个问题变得更加严重。

今天这篇文章主要讲述数据库和缓存双写数据一致性问题常见的解决方案、这些方案中可能存在的坑，以及最优方案是什么。

## 常见方案

通常情况下，我们使用缓存的主要目的是为了提升查询的性能。大多数情况下，我们是这样使用缓存的：

> 1. 用户请求过来之后，先查询缓存有没有数据，如果有直接返回。
> 2. 如果缓存没数据，再继续查询数据库。
> 3. 如果数据库有数据，则将查询出来的数据放入缓存中，然后返回该数据。
> 4. 如果数据库也没有数据，直接返回空。
>
> 如下图所示：
>
> ```mermaid
> graph TD
> A[用户请求] --> B[查询缓存]
> B --> C{是否存在}
> C --> |是|D[返回]
> C --> |否|E[查询数据库]
> E --> F{是否存在}
> F --> |是|G[放入缓存]
> F --> |否|D
> G --> D
> ```

这是缓存非常常见的用法。初看好像没有问题，但是忽略了一个非常重要的细节：**如果数据库中的某条数据，放入缓存后，又马上被更新了，那么该如何更新缓存呢？**

> Q：不更新缓存行不行？
>
> A：当然不行，如果不更新缓存，在很长的一段时间内（取决于缓存的过期时间），用户请求从缓存中获取到的都可能是旧值，而非数据库的最新值。这就导致了数据不一致的问题。

那么，具体应该如何更新缓存呢？

目前有以下4种方案：

> 1. 先写缓存，再写数据库
> 2. 先写数据库，再写缓存
> 3. 先删缓存，再写数据库
> 4. 先写数据库，再删缓存

接下来细说这4种方案

### 先写缓存，再写数据库

对于更新缓存的方案，很多人第一个想到的可能是在写操作中直接更新缓存（写缓存），更直接明了。

那么问题来了：在写操作中，到底是先写缓存，还是先写数据库呢？

在这里先聊聊先写缓存，再写数据库的情况，因为它的问题最严重。

```mermaid
graph TD
A[用户写操作] --> B[写缓存]
B[写缓存] --> C[写数据库]
```

某一个用户的每一次写操作，如果刚写完缓存，突然网络出现了异常，导致写数据库失败了。

其结果是缓存更新成了最新数据，但数据库没有，这样缓存中的数据就变成了脏数据。如果此时该用户的查询请求，正好读到该数据，就会出现问题，因为该数据在数据库中根本不存在，这个问题非常严重。

**缓存的主要目的是把数据库的数据临时保存在内存，便于后续的查询，提升查询速度**。但如果某条数据，在数据库中都不存在，缓存这种“假数据”又有啥意义呢？

因此，先写缓存，再写数据库这种方案是不可取的，在实际工作中用的不多。

### 先写数据库，再写缓存

既然上面的方案行不通，下面聊聊先写数据库，再写缓存的方案。

```mermaid
graph TD
A[用户写操作] --> B[写数据库]
B --> C[写缓存]
```

如果是并发量比较小，对接口性能要求不太高的系统，可以这么做。

但如果在高并发的业务场景中，写数据库和写缓存，都属于远程操作。为了防止出现大事务，造成的死锁问题，通常建议写数据库和写缓存不要放在同一个事务中。

也就是说在该方案中，如果写数据库成功了，但是写缓存失败了，数据库中已写入的数据就不会回滚。

这样就造成**数据库是新数据，而缓存时旧数据，两边数据不一致的情况**。

#### 高并发下的问题

假设在高并发的场景中，针对同一个用户的同一条数据，有两个写数据请求：a和b，它们同时请求业务系统。

其中请求a获取的是旧数据，而请求b获取的是新数据，如下图所示：

> ```mermaid
> graph TD
> A[用户写操作 请求a] --> |1|B[写数据库]
> B --> |2 网络卡顿|C[写缓存]
> C --> |6 写入旧值|D[缓存]
> E[用户写操作 请求b] --> |3|F[写数据库]
> F --> |4|G[写缓存]
> G --> |5 写入新值|D
> ```
>
> 1. 请求a先过来，刚写完了数据库，但由于网络原因，卡顿了以下，还没来得及写缓存。
> 2. 这时请求b过来了，先写了数据库。
> 3. 接下来，请求b顺利写了缓存。
> 4. 此时，请求a卡顿结束，也写了缓存。

很显然，在这个过程当中，请求b在缓存中的`新数据`，被请求a的`旧数据`覆盖了。

也就是说：**在高并发场景中，如果多个线程同时执行先写数据库，再写缓存的操作，可能会出现数据库是新值，而缓存中是旧值，两边数据不一致的情况**。

#### 浪费系统资源

该方案还有一个比较大的问题就是：每个写操作，写完数据库，会马上写缓存，比较`浪费系统资源`。

如果写的缓存，并不是简单的数据内容，而是要经过非常复杂的计算得出的最终结果。这样每写一次缓存，都需要经过一次非常复杂的计算，这样就会造成系统资源的浪费，尤其是`cpu`和`内存`资源。

还有些比较特殊的业务场景，如`写多读少`，在这类业务场景中，每个用户的写操作，都要写一次缓存，有点得不偿失。

由此可见，在高并发的业务场景中，先写数据库，再写缓存，这套方案问题挺多的，也不太建议使用。

### 先删缓存，再写数据库

通过上面的内容可知，如果直接更新缓存的问题很多。

那么，我们换一种思路：不直接`更新缓存`，改为`删除缓存`。

删除缓存方案有以下两种：

> 1. 先删缓存，再写数据库。
> 2. 先写数据库，再删缓存。

首先看先删缓存，再写数据库的情况：

```mermaid
graph TD
A[用户写操作] --> B[删除缓存]
B --> C[写数据库]
```

说白了，在用户的写操作中，先执行删除操作，再去写数据库。这套方案，可以是可以，但是也存在问题。

#### 高并发下的问题

假设在高并发的场景中，同一个用户的同一条数据，有一个读数据请求a，还有另一个写数据请求b（一个更新操作），同时请求到业务系。如下图所示：

> ```mermaid
> graph TD
> A[用户写操作 请求a] --> |1|B[删除缓存]
> B --> |2 清空|D[缓存]
> B --> |8 网络卡顿|C[写数据库]
> E[用户查询操作 请求b] --> |3|F[查询缓存]
> F --> |4 缓存没数据|D
> F --> |5|G[查询数据库]
> G --> |6|H[更新缓存 ]
> H --> |7 写入旧值|D
> ```
>
> 1. 请求a先过来，把缓存删除了，但由于网络原因，卡顿了以下，还没来得及写数据库。
> 2. 这时请求b过来了，先查询缓存发现没数据，再查数据库，有数据，但是是旧值。
> 3. 请求b将数据库中的旧值，更新到缓存中。
> 4. 此时请求a卡顿结束，把新值写入数据库

在这个过程中，请求a的新值并没有被请求b写入缓存，同样会导致缓存和数据库的数据不一致的情况。

那么，这种场景的数据不一致问题，能否解决呢？

#### 缓存双删

在上面的业务场景中，一个读数据请求，一个写数据请求。当写数据请求把缓存删除之后，读数据请求，可能把当时从数据库查询出来的旧值，写入缓存当中。

有人说，那还不好办，请求d在写完数据库之后，把缓存重新删一次不就行了？

> ```mermaid
> graph TD
> A[用户写操作] --> B[删除缓存]
> B --> C[写数据库]
> C --> |间隔一段时间|D[删除缓存]
> ```
>
> 这就是我们所说的**缓存双删**，即在写数据库之前删除一次，写完数据库后，再删除一次。
>
> 该方案有个非常关键的地方是，__第二次删除缓存，并非立马就删，而是要在一定的时间间隔之后__。

我们再重新回顾一下，高并发下一个读数据请求，一个写数据请求导致数据不一致的产生过程：

> 1. 请求a先过来，把缓存删除了。但是由于网络原因，卡顿了以下，还没来得及写数据库。
> 2. 这时请求b过来了，先查缓存发现没数据，再查数据库，有数据，但是是旧值。
> 3. 请求b将数据库中的旧值，更新到缓存中。
> 4. 这时请求a卡顿结束，将新值写入数据库。
> 5. 一段时间后，比如：500ms，请求a将缓存删除。

这样来看确实可以解决缓存不一致的问题。

那么，为什么一定要间隔一段时间之后，才能删除缓存呢？

请求a卡顿结束后，把新值写入数据库后，请求b将数据库中的旧值，更新到缓存中。

接下来，还有一个问题，如果第二次删除缓存时，删除失败了怎么办？

这里先留点悬念，下文会详细说。

### 先写数据库，再删缓存

从前面得知，先删缓存，再写数据库，在并发的情况下，也可能会出现缓存和数据库的数据不一致的情况。

那么，我们只能寄希望于最后的方案了。

接下来，我们重点看看先写数据库，再删缓存的方案。

> ```mermaid
> graph TD
> A[用户写操作] --> B[写数据库]
> B --> C[删除缓存]
> ```
>
> 在高并发的场景中，有一个读数据请求，有一个写数据请求，更新过程如下：
>
> 1. 请求a先写数据库，由于网络原因卡顿了一下，没有来得及删除缓存。
> 2. 请求b查询缓存，发现缓存中有数据，直接返回该数据。
> 3. 请求a删除缓存。

在这个过程中，只有请求b读了一次旧数据，后来旧数据被请求a及时删除了，看起来问题不大。

但如果是读数据请求先过来呢？

> 1. 请求b查询缓存，发现缓存中有数据，直接返回该数据。
> 2. 请求a先写数据库。
> 3. 请求b删除缓存。

这种情况看起来也没问题呀？

答：对的。

但就怕出现下面这种情况，即缓存自己失效了：

> 1. 缓存过期时间到了，自动失效。
> 2. 请求b查询缓存，发现缓存中没有数据，查询数据库的旧值，但由于网络原因卡顿了，没有来得及更新缓存。
> 3. 请求a先写数据库，接着删除了缓存。
> 4. 请求b更新旧值到缓存中。

这时，缓存和数据库的数据同样出现不一致的情况了。

但是这种靠还是比较少的，需要同时满足以下条件才可以：

> 1. 缓存刚好自动失效。
> 2. 请求b从数据库查出旧值，更新缓存的耗时比请求a写数据库，并且删除缓存的时间还长。

我们都知道查数据库的速度，一般比写数据库要快，更何况写完数据库，还要删除缓存。所以绝大多数情况下，写数据请求比读数据情况耗时更长。

由此可见，系统同时满足上述两个条件的概率非常小。

> 推荐大家使用先写数据库，再删除缓存的方案，虽说不能100%避免数据不一致问题，但出现该问题的概率，相对于其他方案来说是最小的。

但在该方案中，如果删除缓存失败了怎么办？

## 删缓存失败怎么办

通过前面我们可以了解到，不管是先写数据库，再删缓存的方案，还是缓存双删的方案，都有一个共同的风险点，即：**如果缓存删除失败了，也会导致缓存和数据库的数据不一致**。

那么，删除缓存失败怎么办呢？

答：需要加**重试机制**。

在接口中如果更新数据库成功了，但更新缓存失败了，可以立刻重试3此。如果其中有任何一次成功，则直接返回成功。如果3次都失败了，则写入数据库，准备后续再处理。

当然，如果你在接口中直接**同步重试**，该接口并发量比较高的时候，可能有点影响接口性能。

这时，就需要改成**异步重试**了。

异步重试的方案有很多种，比如：

> 1. 每次单独起一个线程，该线程专门做重试的工作。但如果在高并发的场景下，可能会创建太多的线程，导致系统`OOM`问题，不太建议使用。
> 2. 将重试的任务交给线程池处理，但如果服务器重启，部分数据可能会丢失。
> 3. 将重试数据写表，然后使用`crontab`、`elastic-job`等定时任务进行重试。
> 4. 将重试的请求写入`mq`等消息中间件，在`mq`的`consumer`中处理。
> 5. 订阅`mysql`的`binlog`，在订阅这种，如果发现了更新数据请求，则删除相应的缓存。

### 定时任务重试

使用定时任务重试的具体方案如下：

> 1. 当用户操作写完数据库，但删除缓存失败了，需要将用户数据写入重试表中。如下图所示：
>
>    ```mermaid
>    graph TD
>    A[用户写操作] --> B[写数据库]
>    B --> C[删除缓存]
>    C --> D{是否删除成功}
>    D --> |是|E(返回)
>    D --> |否|F[写入重试表]
>    ```
>
> 2. 在定时任务重，异步读取重试表中的用户数据。重试表需要记录一个重试次数的字段，初始值为0。然后重试5次，不断删除缓存，每重试一次该字段值+1。如果其中有任意一次成功了，则返回成功。如果重试了5次，还是失败，则需要在重试表记录一个失败的状态，等待后续进一步处理。
>
>    ```mermaid
>    graph TD
>    A[读取重试表] --> B{重试次数是否超过5次}
>    B --> |否|C[删除缓存]
>    C --> D[重试次数+1]
>    D --> E{是否删除成功}
>    E --> |否|B
>    E --> |是|F(结束)
>    B --> |是|F
>    ```

使用定时任务重试的话优缺点：

* 优点：数据是落库的，不会丢数据。
* 缺点：实时性没那么高，对于实时性要求特别高的业务场景，该方案不太适用。但是对于一般场景，还是可以用一用的。

### `mq`

在高并发的业务场景中，`mq`（消息队列）是必不可少的技术之一。它不仅可以异步解耦，还能削峰填谷。对保证系统的稳定性是非常有意义的。

使用`mq`的具体方案如下：

> 1. 当用户操作写完数据库，但删除缓存失败了，产生一条`mq`消息，发送给`mq`服务器。
> 2. `mq`消费者读取`mq`消息，重试5次删除缓存。如果其中有任意一次成功了，则返回成功。如果重试了5次依然失败，则写入死信队列中。

在该方案中，删除缓存可以完全走异步。即用户的写操作，在写完数据库之后，不用立刻删除缓存。而直接发送`mq`消息，到`mq`服务器，然后由`mq`消费者全权负责删除缓存的任务。

因为`mq`的实时性比较高，因此改良后的方案也是一种不错的选择。

### `binlog`

前面我们聊过，无论是定时任务、还是`mq`消息队列，做重试机制，对业务都有一定的侵入性。

在使用定时任务的方案中，需要在业务代码中增加额外的逻辑，如果删除缓存失败，需要将数据写入重试表。

而使用`mq`的方案中，如果删除缓存失败了，需要在业务代码中发送`mq`消息到`mq`服务器。

其实，还有一种更优雅的实现，即`监听binlog`，比如使用：`canal`等中间件。

具体方案如下：

> ```mermaid
> graph TD
> A[用户写操作] --> B[写数据库]
> B --> C[返回]
> D[binlog]
> E[binlog订阅者读取变更数据] --> F[删除缓存]
> F --> G[结束]
> B --> D
> D --> E
> ```
>
> 1. 在业务接口中写数据库之后，就不管了，直接返回成功。
> 2. `mysql`服务器会自动把变更的数据写入`binlog`中。
> 3. `binlog`订阅者获取变更的数据，然后删除缓存。

这套方案中业务接口确实简化了一些流程，只用关心数据库操作即可，而在`binlog`订阅者中做缓存删除工作。

但如果只是按照上述方案进行删除缓存，只删除了一次，也可能会失败。这时就需要加上上面聊到的`重试机制`了。如果删除缓存失败，写入重试表，使用定时任务或重试或写入`mq`，让`mq`自动重试。

在`binlog`订阅这种如果删除缓存失败，则发送一条`mq`消息到`mq`服务器，在`mq`消费者中自动重试5次。如果有任意一次成功，则返回成功。如果重试5次后还是失败，则该消息被自动放入死信队列，后续可能需要人工介入。

如何保证数据库和缓存双写一致性

### `IPv4`简介+高效存储

人们经常使用`VARCHAR(15)`列来存储`IP`地址。然而，它们实际上是4字节的无符号整数，不是字符串。用小数点将地址分成四段的表示方法只是为了让人们阅读容易。所以应该用无符号整型存储`IP`地址。`MySQL`提供`INET_ATON()`和`INET_NTOA()`函数在这两种表示方法之间转换。

### `IPv6`简介+高效存储（兼容`IPv4`）

但是，不幸的是`IPv4`地址（约43亿个）已经用完了。所以，现在很多网址使用`IPv6`。简单的来说，`IPv6`是一个16字节的整数。如何高效地存储这个16字节的整数？当然，我们可以套用存储`IPv4`的思路，使用16字节的整数存储`IPv6`。可惜的是，`MySQL`不支持16字节的整数（最大为8字节的`bigint`类型），所以无法使用整数存储`IPv6`。那么，此时`varchar`似乎又成了首选答案，但真的应该使用`varchar`吗？不妨让我们先看看`IPv6`的三种常见的表示方式。

1. 冒分十六进制表示法
   格式为`X:X:X:X:X:X:X:X`，其中每个`X`表示地址中的`16b`，以十六进制表示，例如：`ABCD:EF01:2345:6789:ABCD:EF01:2345:6789`这种表示法中，每个X的前导0是可以省略的，例如：
   `2001:0DB8:0000:0023:0008:0800:200C:417A`→`2001:DB8:0:23:8:800:200C:417A`。
2. 位压缩表示法
   在某些情况下，一个`IPv6`地址中间可能包含很长的一段0，可以把连续的一段0压缩为`::`。但为保证地址解析的唯一性，地址中`::`只能出现一次，例如：`FF01:0:0:0:0:0:0:1101` → `FF01::1101`，`0:0:0:0:0:0:0:1` → `::1，0:0:0:0:0:0:0:0` →` ::`
3. 内嵌`IPv4`地址表示法
   为了实现`IPv4`-`IPv6`互通，`IPv4`地址会嵌入`IPv6`地址中，此时地址常表示为：`X:X:X:X:X:X:d.d.d.d`，前`96b`采用冒分十六进制表示，而最后`32b`地址则使用`IPv4`的点分十进制表示，例如`::192.168.0.1`与`::FFFF:192.168.0.1`就是两个典型的例子，注意在前`96b`中，压缩0位的方法依旧适用。

如果使用`varchar`类型保存`IPv6`的话，那么在存储以冒分十六进制表示的`IPv6`时，就需要`39 + 1 = 40 `个字节。可`IPv6`本质上一个16字节的数字，难道真的没有其它办法了吗？在`《高性能MySQL 第三版》`中有句话`与CHAR和VARCHAR类似的类型还有BINARY和VARBINARY，它们存储的是二进制字符串。二进制字符串和常规字符串非常相似，但是二进制字符串存储的是字节码而不是字符。`同时，要注意到`VARBINARY(m)`，其中`m`表示字节数。这样我们可以使用`VARBINARY(16)`存储`IPv6`地址，并且兼容存储`IPv4`地址。

```mysql
mysql> CREATE TABLE `test` (
    ->   `ip` varbinary(16) DEFAULT NULL
    -> ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
Query OK, 0 rows affected (0.02 sec)

mysql> insert into test(ip) values(INET6_ATON('2409:8924:1201:1e3e:3d41:3233:7791:ef3a'));
Query OK, 1 row affected (0.01 sec)

mysql> insert into test(ip) values(INET6_ATON('2408:8426:d481:f9e4::1'));
Query OK, 1 row affected (0.00 sec)

mysql> insert into test(ip) values(INET6_ATON('172.241.112.79'));
Query OK, 1 row affected (0.01 sec)

mysql> select *, INET6_NTOA(ip) from test;
+------------------------------------+-----------------------------------------+
| ip                                 | INET6_NTOA(ip)                          |
+------------------------------------+-----------------------------------------+
| 0x2409892412011E3E3D4132337791EF3A | 2409:8924:1201:1e3e:3d41:3233:7791:ef3a |
| 0x24088426D481F9E40000000000000001 | 2408:8426:d481:f9e4::1                  |
| 0xACF1704F                         | 172.241.112.79                          |
+------------------------------------+-----------------------------------------+
3 rows in set (0.00 sec)
```
**[INET6_ATON()](https://dev.mysql.com/doc/refman/8.0/en/miscellaneous-functions.html#function_inet6-aton) 将IPv6的字符串地址转换成数值(MySQL5.6+)，
[INET6_NTOA()](https://dev.mysql.com/doc/refman/8.0/en/miscellaneous-functions.html#function_inet6-ntoa) 将IPv6的数值转换成字符串地址(MySQL5.6+)。**

MySQL IPv4/IPv6高效存储

当您创建表并意识到自己犯了一个错误，或者应用程序的要求发生了变化时，您可以删除该表并重新创建它。但是，如果表中已经有了数据，或者该表被其他数据库对象（例如外键约束）引用，那么这不是一个方便的选择。因此，`PostgreSQL`提供了一组命令来修改现有表。

## 添加列

```sql
ALTER TABLE products ADD COLUMN description text;
```

你也可以在添加列的时候定义约束：

```sql
ALTER TABLE products ADD COLUMN description text CHECK (description <> '');
```

## 删除列 

```
ALTER TABLE products DROP COLUMN description;
```

列中的任何数据都会消失。涉及该列的表约束也会被删除。但是，如果该列被另一个表的外键约束引用，`PostgreSQL`不会默默地删除该约束。您可以通过添加`CASCADE`来授权删除依赖于该列的所有内容：

```sql
ALTER TABLE products DROP COLUMN description CASCADE;
```

## 添加约束

```
ALTER TABLE products ADD CHECK (name <> '');
ALTER TABLE products ADD CONSTRAINT some_name UNIQUE (product_no);
ALTER TABLE products ADD FOREIGN KEY (product_group_id) REFERENCES product_groups;
```

添加非空约束

```
ALTER TABLE products ALTER COLUMN product_no SET NOT NULL;
```

该约束会被立即检查，因此表数据必须满足约束才可以添加。

## 删除约束 

要删除约束，您需要知道它的名称。如果你忘记了约束的名称，可以通过`psql`的`\d tablename`命令或其他方式查看。

```
ALTER TABLE products DROP CONSTRAINT some_name;
```

***与删除列一样，如果您要删除其他内容所依赖的约束，则需要添加`CASCADE`。例如，外键约束依赖于引用列上的唯一或主键约束。***

这对于除非空约束之外的所有约束类型都适用。要删除非空约束，请使用：:

```
ALTER TABLE products ALTER COLUMN product_no DROP NOT NULL;
```

## 更改列的默认值 

```
ALTER TABLE products ALTER COLUMN price SET DEFAULT 7.77;
```

请注意，这不会影响表中的任何现有行，它只会更改以后执行`INSERT`命令时的默认值。

删除默认值:

```
ALTER TABLE products ALTER COLUMN price DROP DEFAULT;
```

这实际上与将默认值设置为 `null` 相同。

## 更改列的数据类型 

```
ALTER TABLE products ALTER COLUMN price TYPE numeric(10,2);
```

只有被更改的列中的所有值都可以通过隐式转换转换为新类型时，此操作才会成功。

`PostgreSQL`将尝试将列的默认值（如果有）以及涉及该列的任何约束转换为新类型。但这些转换可能会失败，或可能产生意想不到的结果。通常最好在更改列类型之前删除列上的所有约束，然后在之后添加适当修改的约束。

## 重命名列

```
ALTER TABLE products RENAME COLUMN product_no TO product_number;
```

## 重命名表 

```
ALTER TABLE products RENAME TO items;
```

PostgreSQL修改表结构

首先查看配置文件位置：

```shell
[postgres@localhost ~]$ psql
psql (16.3)
Type "help" for help.

postgres=# show config_file;
              config_file               
----------------------------------------
 /var/lib/pgsql/16/data/postgresql.conf
(1 row)
```

然后修改`/var/lib/pgsql/16/data/postgresql.conf`，修改`listen_addresses`配置：

```
listen_addresses = '*'					# what IP address(es) to listen on;
                                        # comma-separated list of addresses;
                                        # defaults to 'localhost'; use '*' for all
                                        # (change requires restart)
```

接下来修改`/var/lib/pgsql/16/data/pg_hba.conf`，加入以下配置：

```
host    all             all             0.0.0.0/0               scram-sha-256
```

然后重启`postgresql`：

```bash
$ sudo systemctl restart postgresql-16.service
```





PostgreSQL开启远程访问

> *提示：本文基于Kafka 3.7.0写成*

`Kafka`发行版中自带的配置示例可以用来安装单机服务，主要用于概念验证，并不能满足大型集群的配置需求。`Kafka`有很多配置参数，涉及安装和调优的方方面面，其中的大多数参数可以使用默认值，除非你对调优有特别的需求。

## 常规配置参数

一些配置参数在单机安装时可以使用默认值，但部署到其他环境中时要格外小心。这些参数是针对单台服务器最基本的配置，其中大多数参数需要经过修改后才能用在集群中。

### `broker.id`

默认值：-1

每个`broker`都需要有一个整数标识符，该标识符是使用`broker.id`指定的。可以被设置成其他任意整数。这个值在整个`Kafka`集群中必须是唯一的，并且可以在服务器节点间移动。建议把`ID`设置成与主机名具有相关性的整数，这样就可以很容易地将`ID`与主机名映射起来。如果主机名包含唯一性的数字（比如`host1.example.com`、`host2.example.com`等），那么用`1`、`2`这些数字来设置`broker.id`就再好不过了。

### `listeners`

默认值：`PLAINTEXT://:9092`

`listeners`配置参数是一个用逗号分隔的`URI`列表，也就是要监听的地址和端口。如果没有为监听器指定安全协议，则还需要额外配置`listener.security.protocol.map`参数。监听器的格式为`<protocol>://<hostname>:<port>`，例如，`PLAINTEXT://localhost:9092,SSL://:9091`就是一个合法的配置。如果主机名是`0.0.0.0`，那么将绑定所有的网络接口地址。如果主机名为空，那么将绑定默认的网络接口地址。***需要注意的是，如果指定的端口号小于`1024`，则必须用`root`权限启动`Kafka`，但不建议这么配置***。

### `zookeeper.connect`

默认值：`null`

用于保存`broker`元数据的`ZooKeeper`地址是通过`zookeeper.connect`来指定的。示例配置使用了一个运行在`2181`端口上的`ZooKeeper`，所以指定了`localhost:2181`。这个参数的值是用逗号分隔的一组`hostname:port/path`，每一部分的含义如下。

* `hostname`

  `ZooKeeper`服务器的主机名或IP地址。

* `port`

  `ZooKeeper`的客户端连接端口。

* `/path`

  可选的`ZooKeeper`路径，以作为`Kafka`集群的`chroot`。如果不指定，则默认使用根路径。如果指定的`chroot`路径（一般作为应用程序的根目录）不存在，那么`broker`会在启动时创建它。

### `log.dirs`

默认值：`/tmp/kafka-logs`

`Kafka`把所有消息都保存在磁盘上，存放日志片段的目录是通过`log.dir`来指定的。如果有多个目录，则可以用`log.dirs`来指定。如果没有指定这个参数，则默认使用`log.dir`。***`log.dirs`是一组用逗号分隔的本地文件系统路径。如果指定了多条路径，那么`broker`会根据“最少使用”原则，把同一个分区的日志片段保存到同一条路径下。***需要注意的是，`broker`会向分区数量最少的目录新增分区，而不是向可用磁盘空间最小的目录新增分区，所以并不能保证数据会被均匀地分布在多个目录中。

### `num.recovery.threads.per.data.dir`

默认值：1

`Kafka`使用线程池来处理日志片段。目前，线程池被用于以下3种情形：

* 当服务器正常启动时，用于打开每个分区的日志片段。
* 当服务器发生崩溃并重启时，用于检查和截短每个分区的日志片段。
* 当服务器正常关闭时，用于关闭日志片段。

在默认情况下，每个日志目录只使用一个线程。因为这些线程只在服务器启动和关闭时使用，所以可以多设置一些线程来实现并行操作。特别是对包含大量分区的服务器来说，一旦发生崩溃，在从错误中恢复时可以通过并行操作省下数小时的时间。***需要注意的是，这个参数对应的是`log.dirs`中的一个目录，也就是说，如果`num.recovery.threads.per.data.dir`被设为8，并且`log.dirs`指定了3条路径，那么总共需要24个线程***。

### `auto.create.topics.enable`

默认值：`true`

在默认情况下，`Kafka`会在如下几种情形中自动创建主题：

* 在默认情况下，Kafka会在如下几种情形中自动创建主题。
* 在默认情况下，Kafka会在如下几种情形中自动创建主题。
* 在默认情况下，Kafka会在如下几种情形中自动创建主题。

根据`Kafka`协议，如果一个主题不提前被创建，则无法知道它是否存在。如果你选择通过手动的方式或其他配置系统来显式地创建主题，那么可以把`auto.create.topics.enable`设为`false`。

### `auto.leader.rebalance.enable`

默认值：`true`

为了确保主题的所有权不会集中在一台`broker`上，可以将这个参数设置为`true`，让主题的所有权尽可能地在集群中保持均衡。如果启用了这个功能，那么就会有一个后台线程定期检查分区的分布情况（这个时间间隔可以通过`leader.imbalance.check.interval.seconds`来配置）。如果不均衡的所有权超出了`leader.imbalance.per.broker.percentage`指定的百分比，则会启动一次分区首领再均衡。

### `delete.topic.enable`

默认值：`true`

根据应用场景和数据保留策略的不同，你可能希望将集群锁定，以防主题被随意删除。把这个参数设置为`false`就可以禁用主题删除功能。

## 主题的默认配置

`Kafka`为新创建的主题提供了很多默认的配置参数。

### `num.partitions`

默认值：1

`num.partitions`参数指定了新创建的主题将包含多少个分区，特别是如果启用了主题自动创建功能（默认是启用的），那么主题的分区数就是这个参数指定的值。**默认是1个分区**。需要注意的是，***可以增加主题的分区数，但不能减少***。所以，如果要让一个主题的分区数小于`num.partitions`指定的值，则需要手动创建主题。

### `default.replication.factor`

默认值：1

如果启用了自动创建主题功能，那么这个参数的值就是新创建主题的复制系数。

### `log.retention.ms`

默认值：`null`

`Kafka`通常根据配置的时间长短来决定数据可以被保留多久。我们使用`log.retention.hours`参数来配置时间，默认为168小时，也就是1周。除此以外，还有另外两个参数`log.retention.minutes`和`log.retention.ms`。这3个参数的作用是一样的（**都是用于确定消息将在多久以后被删除**），不过还是推荐使用`log.retention.ms`，因为如果指定了不止一个参数，那么`Kafka`会优先使用具有最小单位值的那个。

### `log.retention.bytes`

默认值：-1

另一种数据保留策略是通过计算已保留的消息的字节总数来判断旧消息是否过期。这个字节总数阈值通过参数`log.retention.bytes`来指定，对应的是每一个分区。也就是说，如果一个主题包含8个分区，并且`log.retention.bytes`被设置为1 GB，那么这个主题最多可以保留8 GB的数据。需要注意的是，**所有保留都是针对单个分区而不是主题执行的**。

所以，如果配置了这个参数，那么当主题增加了新分区，整个主题可以保留的数据也会随之增加。***如果这个值被设置为–1，那么分区就可以无限期地保留数据***。

> 如果同时指定了`log.retention.bytes`和`log.retention.ms`（或另一个按时间保留的参数），那么只要任意一个条件得到满足，消息就会被删除。假设`log.retention.ms`被设置为86 400 000（也就是1天），`log.retention.bytes`被设置为1 000 000 000（也就是1 GB），如果消息字节总数不到一天就超过了1 GB，那么旧数据就会被删除。相反，如果消息字节总数小于1 GB，那么一天之后这些消息也会被删除，尽管分区的数据总量小于1 GB。为简单起见，建议只选择其中的一种保留策略，要么基于数据大小，要么基于时间，或者两种都不选择，以防发生意外的数据丢失。不过，对于复杂的场景，可以两种都使用。

### `log.segment.bytes`

默认值：1073741824 (1GB)

前面介绍的参数都作用在日志片段而不是单条消息上。当消息到达`broker`时，它们会被追加到分区的当前日志片段上。当日志片段大小达到`log.segment.bytes`指定的上限（默认是1 GB）时，当前日志片段会被关闭，一个新的日志片段会被打开。一旦日志片段被关闭，就可以开始进入过期倒计时。这个参数的值越小，关闭和分配新文件就会越频繁，从而降低整体的磁盘写入效率。

如果主题的消息量不是很大，那么如何设置这个参数就变得尤为重要。如果一个主题每天只接收100 MB的消息，并且`log.segment.bytes`使用了默认设置，那么填满一个日志片段将需要10天。因为在日志片段被关闭之前消息是不会过期的，所以如果`log.retention.ms`被设为604 800 000（也就是1周），那么日志片段最多需要17天才会过期。这是因为关闭日志片段需要10天，而根据配置的过期时间，还需要再保留数据7天（要等到日志片段的最后一条消息过期才能将其删除）。

### `log.roll.ms`

默认值：`null`

另一个可用于控制日志片段关闭时间的参数是`log.roll.ms`，它指定了多长时间之后日志片段可以被关闭。就像`log.retention.bytes`和`log.retention.ms`一样，`log.segment.bytes`和`log.roll.ms`并不互斥。日志片段会在大小或时间达到上限时被关闭，就看哪个条件先得到满足。在默认情况下，`log.roll.ms`没有设定值，所以使用`log.roll.hours`设定的默认值——168小时，也就是7天。

> 在使用基于时间的日志片段时，需要考虑并行关闭多个日志片段对磁盘性能的影响。如果多个分区的日志片段一直未达到大小的上限，就会出现这种情况。这是因为`broker`在启动时会计算日志片段的过期时间，一旦满足条件，就会并行关闭它们，尽管它们的数据量可能很少。

### `min.insync.replicas`

默认值：1

为了提升集群的数据持久性，可以将`min.insync.replicas`设置为2，确保至少有两个副本跟生产者保持“同步”。生产者需要配合将`ack`设置为`all`，这样就可以确保至少有两个副本（首领和另一个副本）确认写入成功，从而防止在以下情况下丢失数据：首领确认写入，然后发生停机，所有权被转移到一个副本，但这个副本没有写入成功。如果没有这些配置，则生产者会认为已经写入成功，但实际上消息丢失了。不过，这样做是有副作用的，因为需要额外的开销，所以效率会有所降低。因此，对于能够容忍偶尔消息丢失的高吞吐量集群，不建议修改这个参数的默认值。

### `message.max.bytes`

默认值：1048576 (1MB)

`broker`通过设置`message.max.bytes`参数来限制单条消息的大小，默认值是1 000000，也就是1 MB。如果生产者尝试发送超过这个大小的消息，那么不仅消息不会被`broker`接收，还会收到`broker`返回的错误信息。与其他`broker`配置参数一样，这个参数指的是压缩后的消息大小，也就是说，消息的实际大小可以远大于`message.max.bytes`，只要压缩后小于这个值即可。

这个参数对性能有显著的影响。值越大，负责处理网络连接和请求的线程用在处理请求上的时间就越长。它还会增加磁盘写入块的大小，从而影响`I/O`吞吐量。其他的存储解决方案，比如大对象存储或分层存储，可能也是解决大磁盘写入问题的一种方案。

消费者客户端设置的`fetch.max.bytes`需要与服务器端设置的消息大小保持一致。如果这个参数的值比`message.max.bytes`小，那么消费者就无法读取比较大的消息，进而造成阻塞，无法继续处理消息。在配置`broker`的`replica.fetch.max.bytes`参数时，也遵循同样的原则。

Kafka消费者配置

> *提示：本文基于Kafka 3.7.0写成*

`Kafka`生产者有很多可配置的参数，在[Kafka生产者配置文档](https://kafka.apache.org/documentation/#producerconfigs)中都有说明。它们大部分有合理的默认值，没有必要进行修改。不过有几个参数在内存使用、性能和可靠性方面对生产者影响比较大，本文将详细介绍它们。

## `client.id`

`client.id`是客户端标识符，它的值可以是任意字符串，`broker`用它来识别从客户端发送过来的消息。`client.id`可以被用在日志、指标和配额中。选择一个好的客户端标识符可以让故障诊断变得更容易些。

## `acks`

默认值：`all`

`acks`指定了生产者在多少个分区副本收到消息的情况下才会认为消息写入成功。在默认情况下，`Kafka`会在首领副本收到消息后向客户端回应消息写入成功。这个参数对写入消息的持久性有重大影响，对于不同的场景，使用默认值可能不是最好的选择。

* `acks=0`

  如果`acks=0`，则生产者不会等待任何来自`broker`的响应。也就是说，如果`broker`因为某些问题没有收到消息，那么生产者便无从得知，消息也就丢失了。不过，因为生产者不需要等待`broker`返回响应，所以它们能够以网络可支持的最大速度发送消息，从而达到很高的吞吐量。

* `acks=1`

  如果`acks=1`，那么只要集群的首领副本收到消息，生产者就会收到消息成功写入的响应。如果消息无法到达首领副本（比如首领副本发生崩溃，新首领还未选举出来），那么生产者会收到一个错误响应。为了避免数据丢失，生产者会尝试重发消息。不过，在首领副本发生崩溃的情况下，如果消息还没有被复制到新的首领副本，则消息还是有可能丢失。

* `acks=all`或`acks=-1`

  如果`acks=all`，那么只有当所有副本全部收到消息时，生产者才会收到消息成功写入的响应。这种模式是最安全的，它可以保证不止一个`broker`收到消息，就算有个别`broker`发生崩溃，整个集群仍然可以运行。不过，它的延迟比`acks=1`高，因为生产者需要等待不止一个`broker`确认收到消息。

> 从上面可以看到，为`acks`设置的值越小，生产者发送消息的速度就越快。也就是说，我们通过牺牲可靠性来换取较低的生产者延迟。不过，端到端延迟是指从消息生成到可供消费者读取的时间，这对3种配置来说都是一样的。这是因为为了保持一致性，在消息被写入所有同步副本之前，`Kafka`不允许消费者读取它们。因此，如果你关心的是端到端延迟，而不是生产者延迟，那么就不需要在可靠性和低延迟之间做权衡了：你可以选择最可靠的配置，但仍然可以获得相同的端到端延迟。

## 消息传递时间

### `max.block.ms`

默认值: 60000 (1分钟)

这个参数用于控制在调用`send()`或通过`partitionsFor()`显式地请求元数据时生产者可以发生阻塞的时间。当生产者的发送缓冲区被填满或元数据不可用时，这些方法就可能发生阻塞。当达到`max.block.ms`配置的时间时，就会抛出一个超时异常。

### `delivery.timeout.ms`

默认值: 540000(9分钟)

这个参数用于控制从消息准备好发送（`send()`方法成功返回并将消息放入批次中）到`broker`响应或客户端放弃发送（包括重试）所花费的时间。如果配置的时间不满足这一点，则会抛出异常。通常，成功发送消息的速度要比`delivery.timeout.ms`快得多。如果生产者在重试时超出了`delivery.timeout.ms`，那么将执行回调，并会将`broker`之前返回的错误传给它。如果消息批次还没有发送完毕就超出了`delivery.timeout.ms`，那么也将执行回调，并会将超时异常传给它。**这个时间应该大于或等于`linger.ms`和`request.timeout.ms`**。

### `request.timeout.ms`

默认值：30000 (30秒)

这个参数用于控制生产者在发送消息时等待服务器响应的时间。需要注意的是，这是指生产者在放弃之前等待每个请求的时间，不包括重试、发送之前所花费的时间等。如果设置的值已触及，但服务器没有响应，那么生产者将重试发送，或者执行回调，并传给它一个`TimeoutException`。

## `linger.ms`

默认值：0

这个参数指定了生产者在发送消息批次之前等待更多消息加入批次的时间。生产者会在批次被填满或等待时间达到`linger.ms`时把消息批次发送出去。在默认情况下，只要有可用的发送者线程，生产者都会直接把批次发送出去，就算批次中只有一条消息。把`linger.ms`设置成比0大的数，可以让生产者在将批次发送给服务器之前等待一会儿，以使更多的消息加入批次中。虽然这样会增加一点儿延迟，但也极大地提升了吞吐量。这是因为一次性发送的消息越多，每条消息的开销就越小，如果启用了压缩，则计算量也更少了。

## `retries` 和`retry.backoff.ms`

`retries`默认值：2147483647

`retry.backoff.ms`默认值：1000 (1秒)

当生产者收到来自服务器的错误消息时，这个错误有可能是暂时的（例如，一个分区没有首领）。在这种情况下，`retries`参数可用于控制生产者在放弃发送并向客户端宣告失败之前可以重试多少次。在默认情况下，重试时间间隔是`100`毫秒，但可以通过`retry.backoff.ms`参数来控制重试时间间隔。

并不建议在当前版本的`Kafka`中使用这些参数。相反，你可以测试一下`broker`在发生崩溃之后需要多长时间恢复（也就是直到所有分区都有了首领副本），并设置合理的`delivery.timeout.ms`，让重试时间大于`Kafka`集群从崩溃中恢复的时间，以免生产者过早放弃重试。

生产者并不会重试所有的错误。有些错误不是暂时的，生产者就不会进行重试（例如，“消息太大”错误）。通常，对于可重试的错误，生产者会自动进行重试，所以不需要在应用程序中处理重试逻辑。你要做的是集中精力处理不可重试的错误或者当重试次数达到上限时的情况。

**如果想完全禁用重试，那么唯一可行的方法是将retries设置为0。**

## `buffer.memory`

默认值：33554432

这个参数用来设置生产者要发送给服务器的消息的内存缓冲区大小。如果应用程序调用`send()`方法的速度超过生产者将消息发送给服务器的速度，那么生产者的缓冲空间可能会被耗尽，后续的`send()`方法调用会等待内存空间被释放，如果在`max.block.ms`之后还没有可用空间，就抛出异常。需要注意的是，这个异常与其他异常不一样，它是`send()`方法而不是`Future`对象抛出来的。

## `compression.type`

默认值：`none`

在默认情况下，生产者发送的消息是未经压缩的。这个参数可以被设置为`none`, `gzip`、 `snappy`、`lz4`、或`zstd`，这指定了消息被发送给`broker`之前使用哪一种压缩算法。`snappy`压缩算法由谷歌发明，虽然占用较少的`CPU`时间，但能提供较好的性能和相当可观的压缩比。如果同时有性能和网络带宽方面的考虑，那么可以使用这种算法。`gzip`压缩算法通常会占用较多的`CPU`时间，但提供了更高的压缩比。如果网络带宽比较有限，则可以使用这种算法。使用压缩可以降低网络传输和存储开销，而这些往往是向Kafka发送消息的瓶颈所在。

## `batch.size`

默认值：16384

当有多条消息被发送给同一个分区时，生产者会把它们放在同一个批次里。这个参数指定了一个批次可以使用的内存大小。需要注意的是，该参数是按照字节数而不是消息条数来计算的。当批次被填满时，批次里所有的消息都将被发送出去。但是生产者并不一定都会等到批次被填满时才将其发送出去。那些未填满的批次，甚至只包含一条消息的批次也有可能被发送出去。所以，就算把批次大小设置得很大，也不会导致延迟，只是会占用更多的内存而已。但如果把批次大小设置得太小，则会增加一些额外的开销，因为生产者需要更频繁地发送消息。

## `max.in.flight.requests.per.connection`

默认值：5

这个参数指定了生产者在收到服务器响应之前可以发送多少个消息批次。它的值越大，占用的内存就越多，不过吞吐量也会得到提升。实验数据表明，在单数据中心环境中，该参数被设置为2时可以获得最佳的吞吐量，但使用默认值5也可以获得差不多的性能。

**`Kafka`可以保证同一个分区中的消息是有序的**。也就是说，如果生产者按照一定的顺序发送消息，那么`broker`会按照这个顺序把它们写入分区，消费者也会按照同样的顺序读取它们。在某些情况下，顺序是非常重要的。

假设我们把`retries`设置为非零的整数，并把`max.in.flight.requests.per.connection`设置为比1大的数。如果第一个批次写入失败，第二个批次写入成功，那么`broker`会重试写入第一个批次，等到第一个批次也写入成功，两个批次的顺序就反过来了。

我们希望至少有2个正在处理中的请求（出于性能方面的考虑），并且可以进行多次重试（出于可靠性方面的考虑），这个时候，最好的解决方案是将`enable.idempotence`设置为`true`。这样就可以在最多有5个正在处理中的请求的情况下保证顺序，并且可以保证重试不会引入重复消息。

## `enable.idempotence`

默认值：`true`

从0.11版本开始，`Kafka`支持精确一次性`exactly once`语义。幂等生产者是它的一个简单且重要的组成部分。

假设为了最大限度地提升可靠性，你将生产者的`acks`设置为`all`，并将`delivery.timeout.ms`设置为一个比较大的数，允许进行尽可能多的重试。这些配置可以确保每条消息被写入`Kafka`至少一次。但在某些情况下，消息有可能被写入`Kafka`不止一次。假设一个`broker`收到了生产者发送的消息，然后消息被写入本地磁盘并成功复制给了其他`broker`。此时，这个`broker`还没有向生产者发送响应就发生了崩溃。而生产者将一直等待，直到达到`request.timeout.ms`，然后进行重试。重试发送的消息将被发送给新的首领，而这个首领已经有这条消息的副本，因为之前写入的消息已经被成功复制给它了。现在，你就有了一条重复的消息。

为了避免这种情况，可以将`enable.idempotence`设置为`true`。当幂等生产者被启用时，生产者将给发送的每一条消息都加上一个序列号。如果`broker`收到具有相同序列号的消息，那么它就会拒绝第二个副本，而生产者则会收到`DuplicateSequenceException`，这个异常对生产者来说是无害的。

如果要启用幂等性，那么`max.in.flight.requests.per.connection`应小于或等于5、`retries`应大于0，并且acks被设置为all。如果设置了不恰当的值，则会抛出`ConfigException`异常。

## `max.request.size`

默认值：1048576

这个参数用于控制生产者发送的请求的大小。它限制了可发送的单条最大消息的大小和单个请求的消息总量的大小。假设这个参数的值为`1 MB`，那么可发送的单条最大消息就是`1 MB`，或者生产者最多可以在单个请求里发送一条包含`1024`个大小为`1 KB`的消息。另外，`broker`对可接收的最大消息也有限制`message.max.bytes`，其两边的配置最好是匹配的，以免生产者发送的消息被`broker`拒绝。

## `receive.buffer.bytes`和`send.buffer.bytes`

`receive.buffer.bytes`默认值：65536

`send.buffer.bytes`默认值：131072

这两个参数分别指定了`TCP socket`接收和发送数据包的缓冲区大小。如果它们被设为`–1`，就使用操作系统默认值。如果生产者或消费者与`broker`位于不同的数据中心，则可以适当加大它们的值，因为跨数据中心网络的延迟一般都比较高，而带宽又比较低。

MySQL

如何保证数据库和缓存双写一致性

MySQL IPv4/IPv6高效存储

最近发布

PostgreSQL修改表结构

PostgreSQL开启远程访问

Kafka消费者配置

Kafka生产者配置

分类

算法精解公众号

鬼畜动图小程序