太细了：美团一面连环夺命20问，搞定就60W起

爱被打了一巴掌 2024-02-19 10:04 136阅读 0赞

### 说在前面 ###

在40岁老架构师尼恩的（50+）**读者社群**中，经常有小伙伴，需要面试美团、京东、阿里、 百度、头条等大厂。

下面是一个小伙伴成功拿到通过了美团一面面试，现在把面试真题和参考答案收入咱们的宝典。

通过美团一面真题， 大家可以看看，**收个优质Offer需要学点啥？**

**总之，光代码漂亮不够， 面试，还得会吹。**

![4ee7204a171348eb8b1f8a905ed6cdc1.png][]

这里把题目以及**答案**，经过整理和梳理之后，收入咱们的《[尼恩Java面试宝典PDF][Java_PDF]》 V121版本，供后面的小伙伴参考，提升大家的 3高 架构、设计、开发、吹牛水平。

> 《尼恩 架构笔记》《尼恩高并发三部曲》《尼恩Java面试宝典》的PDF，请到公号【技术自由圈】获取

#### 文章目录 ####

*   *  说在前面
     *  美团一面
     *   *  1、redis基本数据类型，应用场景？
         *  2、redis如何实现共同关注？
         *   *  补充：Set集合三个特殊方法 sinter 、sunion 、sdiff
             *  sinter 交集的示例
             *  sunion 并集的示例
             *  sdiff 差集的示例
         *  3、redis持久化，AOF，RDB，会丢数据吗？
         *   *  1.AOF（Append-Only File）持久化：
             *  2.RDB（Redis DataBase）持久化：
             *  补充：AOF 配置
             *  AOF的优缺点
             *  RDB-AOF混合持久化
         *  4、redis穿透击穿雪崩，解决？
         *   *  缓存穿透解决措施：
             *  缓存击穿解决措施：
             *  缓存雪崩解决措施：
             *  缓存更新策略：
             *  合理的缓存策略：
         *  5、redis大Key，怎么处理？
         *   *  拆分数据：
             *  使用数据压缩：
             *  使用数据分片：
             *  定期清理：
             *  使用内存优化的数据结构：
             *  数据预热和缓存策略：
         *  6、redis分布式锁，Zookeeper分布式锁，怎么选择？
         *   *  性能：
             *  一致性：
             *  部署复杂性：
             *  可用性：
             *  应用已有基础：
             *  数据持久性需求：
             *  社区支持：
         *  7、说说熔断与限流？
         *   *  一、熔断：
             *  二、限流：
         *  8、为什么kafka多用于消息中转，很少用于实时计算？
         *  9、说说kafka高可用，高吞吐？
         *   *  高可用：
             *  高吞吐：
         *  10、说说kafka不重不丢？
         *  11、说说kafka消费端幂等性？
         *  12、说说kafka消费者群组？
         *  13、说说kafka生产端分配，路由？
         *  14、说说mysql索引，为什么B+树？
         *  15、说说主键索引，二级索引？
         *   *  \*\*1. 主键索引\*\*：
             *  \*\*2. 二级索引\*\*：
         *  16、说说mysql ACID？
         *  17、mysql事务隔离级别，RR解决幻读吗？什么场景下会幻读？
         *   *  \*\*1. 什么是幻读？\*\*
             *  \*\*2. RR隔离级别如何解决幻读？\*\*
             *  \*\*可能出现幻读问题\*\*：
         *  18、MVCC，事务版本号怎么生成，存在哪里？
         *  19、说说binlog，redolog，undolog？
         *   *  \*\*1. Binlog\*\*（二进制日志）：
             *  \*\*2. Redo Log\*\*（重做日志）：
             *  \*\*3. Undo Log\*\*（撤销日志）：
         *  20、挂了用什么log，主从同步用什么log？
         *  21、说说binlog的两阶段提交？
         *   *  \*\*1. 预提交阶段\*\*：
             *  \*\*2. 确认提交阶段\*\*：
         *  22、算法题：买卖股票
         *   *  \*\*描述\*\*：
             *  \*\*思路\*\*：
             *  实现：
     *  尼恩说在最后
     *  尼恩技术圣经系列PDF

### 美团一面 ###

#### 1、redis基本数据类型，应用场景？ ####

1.  **String**（字符串）：适用于存储文本、数字等数据。常见用途包括缓存、计数器、会话管理等。
2.  **Hash**（哈希表）：适合存储对象（如用户数据、商品信息）的多个字段及其值。能快速检索或修改字段值。
3.  **List**（列表）：用于保存有序的元素序列。适用于实现消息队列、栈、发布订阅等功能。
4.  **Set**（集合）：用于保存不重复的元素。适用于存储标签、好友列表等，也可用于计算交集、并集等操作。
5.  **Sorted Set**（有序集合）：类似于 Set，但每个元素都有一个分数，可用分数对元素进行排序。适用于实现排行榜、优先级队列等。
6.  **Bitmap**（位图）：适合存储布尔值，可用于追踪用户在线状态、用户活跃度等。能执行位运算以统计和查询状态。
7.  **HyperLogLog**：用于估算集合中不重复元素的数量，适用于统计 UV（独立访客数）等场景。
8.  **Geospatial**（地理空间）：用于存储地理位置信息，支持距离计算和附近位置的查询。适合实现地图应用、位置服务等。
9.  **Pub/Sub**（发布订阅）：用于实现消息发布和订阅机制，适用于构建实时通知、事件驱动系统等。

#### 2、redis如何实现共同关注？ ####

要实现共同关注功能（如社交网络中的好友关系），可以利用 Redis 数据结构来存储关注关系。

给个示例：

假设存在两个用户，用户 A 和用户 B，他们都可以关注其他用户。我们希望找到他们共同关注的用户。

1.  使用集合（Set）存储用户的关注列表：

用户A的关注列表：sadd userA_following userC userD userE
    用户B的关注列表：sadd userB_following userD userE userF

1.  找出用户 A 和用户 B 的共同关注：

使用集合的交集操作（`sinter`）找出两个用户的共同关注：

sinter userA_following userB_following

该操作将返回一个集合，包含用户 A 和用户 B 共同关注的用户（在此示例中是 userD 和 userE）。

1.  可以采用类似方法查找其他用户的共同关注，或执行其他操作，如取消关注、添加新关注等。

##### 补充：Set集合三个特殊方法 sinter 、sunion 、sdiff #####

redis 支持 Set集合的数据存储，其中有三个比较特殊的方法：

*  `sinter key [key …]` 查看一个集合的全部成员，该集合是所有给定集合的交集。
 *  `sunion key [key …]` 查看一个集合的全部成员，该集合是所有给定集合的并集。
 *  `sdiff key [key …]` 查看所有给定 key 与第一个 key 的差集

##### sinter 交集的示例 #####

redis> SMEMBERS group_1
    1) "LI LEI"
    2) "TOM"
    3) "JACK"
    
    redis> SMEMBERS group_2
    1) "HAN MEIMEI"
    2) "JACK"
    
    redis> SINTER group_1 group_2      # 取的是交集的数据 
    1) "JACK"

##### sunion 并集的示例 #####

redis> SMEMBERS songs
    1) "Billie Jean"
    
    redis> SMEMBERS my_songs
    1) "Believe Me"
    
    redis> SUNION songs my_songs       # 取的是集合的并集数据
    1) "Billie Jean"
    2) "Believe Me"

##### sdiff 差集的示例 #####

redis> SMEMBERS peter_movies
    1) "bet man"
    2) "start war"
    3) "2012"
    
    redis> SMEMBERS joe_movies
    1) "hi, lady"
    2) "Fast Five"
    3) "2012"
    
    redis> SDIFF peter_movies joe_movies     # 取的是两个集合的差集
    1) "bet man"
    2) "start war"

#### 3、redis持久化，AOF，RDB，会丢数据吗？ ####

两种持久化策略：

*  AOF
 *  RDB

##### 1.AOF（Append-Only File）持久化： #####

a. AOF持久化以追加的方式记录每个写操作（包括SET、INCR等）到一个日志文件中，该文件包含了恢复数据所需的所有写操作。

b. AOF持久化可以配置为每秒同步一次（默认配置），或者根据需要更频繁地同步。

c. 由于AOF记录了每个写操作，通常情况下不会丢失数据，即使Redis宕机，也可以通过AOF文件来完全恢复数据。

##### 2.RDB（Redis DataBase）持久化： #####

a. RDB持久化是通过周期性快照（快照）整个数据集到磁盘的方式。

b. RDB 文件包含特定时间点数据集的快照，因此两次快照之间的数据更改可能会丢失。

c. 默认情况下，Redis 每隔一段时间（可配置）执行一次 RDB 快照。

若 Redis 在两次快照间崩溃，可能导致数据丢失。

总的来说：

*  AOF持久化默认会丢失1s数据，也可以配置为每次都刷盘，这样不会丢失数据，因为它记录了每个写操作，但会有一定的数据恢复成本。
 *  RDB持久化在快照间可能会丢失数据，但因为RDB文件只包含快照时的数据，所以通常比AOF更快速。

##### 补充：AOF 配置 #####

在 redis.conf 配置文件的 APPEND ONLY MODE 下：

![b45760710790930455759bf344bd6355.png][]

①、**appendonly**：默认值为no，也就是说redis 默认使用的是rdb方式持久化，如果想要开启 AOF 持久化方式，需要将 appendonly 修改为 yes。 AOF 保存文件的位置和 RDB 保存文件的位置一样，都是通过 redis.conf 配置文件的 dir 配置文件的位置.

②、**appendfilename** ：aof文件名，默认是"appendonly.aof"

③、\*\*appendfsync：\*\*aof持久化策略的配置；

*  no表示不执行fsync，由操作系统保证数据同步到磁盘，速度最快，但是不太安全；
 *  always表示每次写入都执行fsync，以保证数据同步到磁盘，效率很低；
 *  everysec表示每秒执行一次fsync，可能会导致丢失这1s数据。通常选择 everysec ，兼顾安全性和效率。

④、no-appendfsync-on-rewrite：在aof重写或者写入rdb文件的时候，会执行大量IO，此时对于everysec和always的aof模式来说，执行fsync会造成阻塞过长时间，no-appendfsync-on-rewrite字段设置为默认设置为no。如果对延迟要求很高的应用，这个字段可以设置为yes，否则还是设置为no，这样对持久化特性来说这是更安全的选择。 设置为yes表示rewrite期间对新写操作不fsync,暂时存在内存中,等rewrite完成后再写入，默认为no，建议yes。Linux的默认fsync策略是30秒。可能丢失30秒数据。默认值为no。

⑤、auto-aof-rewrite-percentage：默认值为100。aof自动重写配置，当目前aof文件大小超过上一次重写的aof文件大小的百分之多少进行重写，即当aof文件增长到一定大小的时候，Redis能够调用bgrewriteaof对日志文件进行重写。当前AOF文件大小是上次日志重写得到AOF文件大小的二倍（设置为100）时，自动启动新的日志重写过程。

⑥、auto-aof-rewrite-min-size：64mb。设置允许重写的最小aof文件大小，避免了达到约定百分比但尺寸仍然很小的情况还要重写。

⑦、aof-load-truncated：aof文件可能在尾部是不完整的，当redis启动的时候，aof文件的数据被载入内存。重启可能发生在redis所在的主机操作系统宕机后，尤其在ext4文件系统没有加上data=ordered选项，出现这种现象 redis宕机或者异常终止不会造成尾部不完整现象，可以选择让redis退出，或者导入尽可能多的数据。如果选择的是yes，当截断的aof文件被导入的时候，会自动发布一个log给客户端然后load。如果是no，用户必须手动redis-check-aof修复AOF文件才可以。默认值为 yes。

##### AOF的优缺点 #####

优点：

①、AOF 持久化的方法提供了多种的同步频率，即使使用默认的同步频率每秒同步一次，Redis 最多也就丢失 1 秒的数据而已。

②、AOF 文件使用 Redis 命令追加的形式来构造，因此，即使 Redis 只能向 AOF 文件写入命令的片断，使用 redis-check-aof 工具也很容易修正 AOF 文件。

③、AOF 文件的格式可读性较强，这也为使用者提供了更灵活的处理方式。例如，如果我们不小心错用了 FLUSHALL 命令，在重写还没进行时，我们可以手工将最后的 FLUSHALL 命令去掉，然后再使用 AOF 来恢复数据。

缺点：

①、对于具有相同数据的的 Redis，AOF 文件通常会比 RDF 文件体积更大。

②、虽然 AOF 提供了多种同步的频率，默认情况下，每秒同步一次的频率也具有较高的性能。但在 Redis 的负载较高时，RDB 比 AOF 具好更好的性能保证。

③、RDB 使用快照的形式来持久化整个 Redis 数据，而 AOF 只是将每次执行的命令追加到 AOF 文件中，因此从理论上说，RDB 比 AOF 方式更健壮。官方文档也指出，AOF 的确也存在一些 BUG，这些 BUG 在 RDB 没有存在。

那么对于 AOF 和 RDB 两种持久化方式，我们应该如何选择呢？

如果可以忍受一小段时间内数据的丢失，毫无疑问使用 RDB 是最好的，定时生成 RDB 快照（snapshot）非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快，而且使用 RDB 还可以避免 AOF 一些隐藏的 bug；否则就使用 AOF 重写。但是一般情况下建议不要单独使用某一种持久化机制，而是应该两种一起用，在这种情况下,当redis重启的时候会优先载入AOF文件来恢复原始的数据，因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整。Redis后期官方可能都有将两种持久化方式整合为一种持久化模型。

##### RDB-AOF混合持久化 #####

在Redis4.0之后，在RDB和AOF两种持久化方式之外，又新增了RDB-AOF混合持久化方式。

这种方式结合了RDB和AOF的优点，既能快速加载又能避免丢失过多的数据。

具体配置为：

aof-use-rdb-preamble

设置为yes表示开启，设置为no表示禁用。

当开启混合持久化时，主进程先fork出子进程将现有内存副本全量以RDB方式写入aof文件中，然后将缓冲区中的增量命令以AOF方式写入aof文件中，写入完成后通知主进程更新相关信息，并将新的含有 RDB和AOF两种格式的aof文件替换旧的aof文件。

简单来说：混合持久化方式产生的文件一部分是RDB格式，一部分是AOF格式。

这种方式优点我们很好理解，缺点就是不能兼容Redis4.0之前版本的备份文件了。

#### 4、redis穿透击穿雪崩，解决？ ####

##### 缓存穿透解决措施： #####

a. Bloom Filter：利用布隆过滤器筛选出不在缓存中的请求，以降低对数据库的请求压力。

b. 空值缓存：虽数据库中不存在该数据，但仍将其缓存，但设置较短的过期时间，以防频繁查询。

c. 缓存空对象：当数据库查询结果为空时，也将该结果缓存，但设定较短的过期时间，避免重复查询。

##### 缓存击穿解决措施： #####

a. 互斥锁：采用互斥锁保护缓存，当缓存失效时，只允许一个请求查询数据库，其他请求等待结果，避免多个请求同时击穿缓存。

b. 热点数据预热：定期或启动时预先加载热门数据至缓存，防止因突发大量请求导致缓存击穿。

##### 缓存雪崩解决措施： #####

a. 缓存失效时间随机性：为缓存失效时间增加一定随机性，使缓存不会同时失效，减轻对数据库的并发请求压力。

b. 持久化缓存：利用 AOF 和 RDB 的持久化机制确保缓存数据可靠性，即使在缓存雪崩情况下，也能从持久化数据中恢复。

c. 多级缓存：采用多级缓存（如本地内存缓存、分布式缓存、CDN 等）分担缓存压力，使某一缓存层发生雪崩时，其他层仍能提供服务。

##### 缓存更新策略： #####

a. 异步刷新：缓存失效后，后台异步更新缓存，避免请求等待缓存更新。

b. 加锁更新：缓存失效时，仅允许一个请求查询数据库并更新缓存，其他请求等待结果。

##### 合理的缓存策略： #####

a. 根据数据访问模式和业务需求，选择合适的缓存策略，如 LRU（最近最少使用）、LFU（最不常使用）、TTL（Time To Live）等。

#### 5、redis大Key，怎么处理？ ####

Redis中的大Key通常指缓存键对应的值较大，可能包含大型数据结构、大量文本或二进制数据等。这类数据会导致内存占用过高，进而影响性能。给大家一些处理Redis大Key的参考方法：

##### 拆分数据： #####

a. 如有条件，将大型数据拆分成多个小型数据，并分别存储在独立的键中。这能降低单个键的大小，减少内存占用。

b. 例如，若某个键存储了大型 JSON 对象，可将其拆分为多个子键，每个子键存储 JSON 对象的部分数据。

##### 使用数据压缩： #####

a. 在存储文本或二进制数据之前，可对其进行压缩，在读取时再解压缩。虽然 Redis 本身不支持数据压缩，但可在应用层实现压缩和解压缩操作。

##### 使用数据分片： #####

a大 Key 是由多个小 Key 组成的集合，可以使用 Redis 的数据分片或分区技术，将数据分布在多个 Redis 实例中，每个实例负责一部分数据。

这有助于减轻单个实例的内存压力。

##### 定期清理： #####

a. 若大 Key 生命周期有限，可以定期清理不再需要的数据，以释放内存。

b. 使用`DEL`命令删除不再需要的大Key。

##### 使用内存优化的数据结构： #####

对于大型集合或列表，可以考虑使用 Redis 的内存优化数据结构，如 HyperLogLog、Redis Streams 等，以降低内存占用。

##### 数据预热和缓存策略： #####

如果大 Key 是在系统启动时加载的，可以实施数据预热策略，提前将热门数据加载到缓存中，以减轻启动时的内存压力。

#### 6、redis分布式锁，Zookeeper分布式锁，怎么选择？ ####

给大家一些选择是需要考虑的因素，在什么情况用哪种分布式锁是更优选择。

##### 性能： #####

a. 通常情况下，Redis 的性能优于 ZooKeeper，因为它采用内存存储系统，而 ZooKeeper 采用磁盘存储。

b. 如果你追求高性能的分布式锁，Redis 是更好的选择。

##### 一致性： #####

a. ZooKeeper 提供强一致性，适合对一致性要求较高的分布式应用，如协调和选举等。

b. Redis 的分布式锁在某些情况下可能出现失效或死锁，因为它是基于主从复制的。

##### 部署复杂性： #####

a. ZooKeeper 需要独立的 ZooKeeper 集群，需进行维护和管理。

b. Redis 部署相对简单，特别是若你在应用中已使用 Redis。

##### 可用性： #####

a. Redis 扩展和部署较容易，因此便于实现高可用性。

b. ZooKeeper 的部署和维护较为复杂，尤其在多数据中心环境中。

##### 应用已有基础： #####

a. 若你在应用中已使用 Redis，添加 Redis 分布式锁较为容易集成。

b. 若应用已使用 ZooKeeper，选择 ZooKeeper 分布式锁更为合适。

##### 数据持久性需求： #####

a. 需更强数据持久性和一致性时，可考虑使用 ZooKeeper。

b. Redis 提供持久性，但通常在性能和可用性之间作出权衡。

##### 社区支持： #####

Redis 和 ZooKeeper 均有活跃的社区支持，需根据具体需求评估社区支持和文档资源。

综合考虑以上因素，若应用需高性能分布式锁且可容忍一定程度的一致性弱点，可选 Redis 分布式锁。若需强一致性和更复杂分布式协同操作，ZooKeeper 分布式锁更适合。

#### 7、说说熔断与限流？ ####

在分布式系统中，熔断和限流是两种关键技术，它们有助于提高系统的可用性和稳定性。

##### 一、熔断： #####

熔断机制旨在防止系统出现雪崩效应。当某个服务或组件的错误率超过设定阈值，熔断器就会启动，阻止进一步的请求访问该服务，以避免更多系统部分崩溃。熔断器开启后，会定时检查服务可用性，一旦恢复正常，便关闭熔断器，允许请求再次访问。

熔断的好处包括：

*  避免错误服务波及整个系统。
 *  减轻故障服务负担，减少不必要请求。
 *  促使系统恢复正常运行，而非持续崩溃。

**1. 熔断器实现**

import java.util.concurrent.TimeUnit;
    public class CircuitBreaker {
          
        private final ThreadPoolExecutor executor;  
        private final Semaphore permit;  
        private final long timeout;
        public CircuitBreaker(int maxThreads, long timeout, TimeUnit unit) {
          
            this.executor = new ThreadPoolExecutor(maxThreads, maxThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());  
            this.permit = new Semaphore(1);  
            this.timeout = timeout;  
        }
        public void execute(Runnable command) {
          
            permit.acquire();  
            try {
          
                executor.execute(command);  
            } catch (Exception e) {
          
                // 处理异常，例如记录日志、发送告警等  
                System.err.println("Error occurred while executing command: " + e.getMessage());  
            } finally {
          
                permit.release();  
            }  
        }
        public void reset() {
          
            executor.shutdown();  
            try {
          
                if (!executor.awaitTermination(timeout, TimeUnit.MILLISECONDS)) {
          
                    // 超过恢复时间，仍然无法恢复，则考虑进行降级或熔断  
                    System.err.println("Failed to reset circuit breaker");  
                }  
            } catch (InterruptedException e) {
          
                // 等待过程中出现中断，表示恢复失败  
                System.err.println("Interrupted while waiting for circuit breaker reset: " + e.getMessage());  
            }  
            executor = new ThreadPoolExecutor(executor.getPoolSize(), executor.getPoolSize(), 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());  
        }  
    }

**2.使用示例**

public class CircuitBreakerDemo {
          
        public static void main(String[] args) {
          
            CircuitBreaker circuitBreaker = new CircuitBreaker(10, 5000, TimeUnit.MILLISECONDS);
            for (int i = 0; i < 10; i++) {
          
                circuitBreaker.execute(() -> {
          
                    System.out.println("Executing command " + i);  
                    throw new RuntimeException("模拟异常");  
                });  
            }
            circuitBreaker.reset();
            for (int i = 0; i < 10; i++) {
          
                circuitBreaker.execute(() -> {
          
                    System.out.println("Executing command " + i);  
                });  
            }  
        }  
    }

##### 二、限流： #####

限流机制旨在控制服务请求速率，防止系统同时承受过多请求。根据应用需求，限流可以有多种实现方式，如：

*  固定速率限流：每秒最多允许N个请求。
 *  漏桶算法：请求以固定速率进入“桶”，若“桶”满，多余请求将被丢弃。
 *  令牌桶算法：每个请求需获取一个令牌，令牌以固定速率生成，若无令牌，请求将被拒绝。

限流的好处包括：

*  防止大规模请求突然涌入，导致系统过载。
 *  控制系统负载，保护后端服务免受过多请求压力。
 *  提供一种方式来保护资源，如API，免受滥用和DDoS攻击。

**1. 定义一个 TokenBucket 类**：

import java.util.concurrent.TimeUnit;
    public class TokenBucket {
          
        private final long capacity;  
        private final long tokensPerSecond;  
        private long lastRefillTime;  
        private long tokens;
        public TokenBucket(long capacity, long tokensPerSecond) {
          
            this.capacity = capacity;  
            this.tokensPerSecond = tokensPerSecond;  
            this.lastRefillTime = System.currentTimeMillis();  
            this.tokens = capacity;  
        }
        public boolean consume(long tokens) {
          
            if (tokens > capacity) {
          
                return false;  
            }
            refill();  
            if (tokens <= tokens) {
          
                tokens -= tokens;  
                return true;  
            } else {
          
                return false;  
            }  
        }
        private void refill() {
          
            long now = System.currentTimeMillis();  
            long elapsed = now - lastRefillTime;  
            long newTokens = elapsed * tokensPerSecond;  
            tokens = Math.min(capacity, tokens + newTokens);  
            lastRefillTime = now;  
        }  
    }

**2. 使用示例**：

public class Main {
          
        public static void main(String[] args) {
          
            TokenBucket bucket = new TokenBucket(10, 2);
            System.out.println(bucket.consume(5)); // 消耗 5 个令牌，返回 True  
            try {
          
                Thread.sleep(1000);  
            } catch (InterruptedException e) {
          
                e.printStackTrace();  
            }  
            System.out.println(bucket.consume(6)); // 消耗 6 个令牌，返回 False  
            try {
          
                Thread.sleep(500);  
            } catch (InterruptedException e) {
          
                e.printStackTrace();  
            }  
            System.out.println(bucket.consume(2)); // 消耗 2 个令牌，返回 True  
        }  
    }

在这个例子中，我们创建了一个 TokenBucket 类，通过 consume 方法限制每秒请求的次数。每次调用 consume 方法时，会先检查当前令牌是否足够，如果不够，则返回 False，表示限流。如果足够，则消耗令牌并返回 True。在 refill 方法中，我们会根据时间间隔和每秒添加的令牌数来补充令牌。  
这个简单的限流器可以用于保护 Java 系统的瓶颈部分，防止流量过大导致系统崩溃。在实际应用中，可以根据需求进行优化和扩展。

#### 8、为什么kafka多用于消息中转，很少用于实时计算？ ####

1.  **持久性存储**：Kafka 作为一种持久性消息队列，将消息存储在磁盘上，确保数据不会因消费者未能及时处理而丢失。这使得 Kafka 非常适合作为消息中转，让消费者能在任意时间点访问消息，而不仅限于实时计算。
2.  **消息存储和检索**：Kafka 的主要功能是存储和检索消息，而非执行复杂实时计算。它具备高吞吐量和低延迟的消息存储与检索能力，但并未提供计算框架，因此在实时计算方面相对薄弱。
3.  **数据保留**：Kafka 支持根据不同策略保留消息，如时间或存储大小。这使得 Kafka 适用于长期数据存储和数据历史查询，而不仅仅是实时计算。
4.  **消息分发和复制**：Kafka 具有强大的消息分发和复制机制，确保消息可靠地传递到多个消费者或订阅者。这使得 Kafka 成为消息分发的理想选择，但它并不提供实时计算所需的状态管理和处理。

#### 9、说说kafka高可用，高吞吐？ ####

##### 高可用： #####

1.  **分布式架构**：Kafka 作为一个分布式系统，可部署在多台服务器上，从而具备冗余性。当一台服务器出现故障时，其他服务器能够继续运行，确保消息流的可用性。
2.  **复制和副本**：Kafka 采用分区组织消息，每个分区都有多个副本。这意味着消息在多个服务器上进行复制，以防数据丢失。若某个副本不可用，其他副本仍可提供数据。
3.  **ZooKeeper**：Kafka 利用 Apache ZooKeeper 管理集群元数据和协调任务。ZooKeeper 提供了分布式锁和选举机制，确保 Kafka 集群内各组件协同工作。
4.  **消费者位移**：Kafka 记录了消费者在各分区的位移，这意味着即使消费者出现故障，也可以从上次中断的地方继续消费消息，避免数据丢失。

##### 高吞吐： #####

1.  **分区**：Kafka 通过分区实现水平扩展。每个分区可由不同服务器处理，从而平均分配负载，提高吞吐量。
2.  **批处理和零拷贝**：Kafka 采用批处理机制降低磁盘和网络开销。此外，它还利用零拷贝技术将数据从生产者传输至消费者，减少 CPU 和内存开销。
3.  **压缩**：Kafka 支持消息压缩，缩小数据传输尺寸，提高吞吐量。
4.  **持久性**：Kafka 的消息持久化至磁盘，允许在多个消费者之间共享数据，并确保即使消费者离线，数据也不会丢失。
5.  **分布式部署**：Kafka 可部署在多台服务器上，充分利用硬件资源，提供高吞吐量。

import org.apache.kafka.clients.producer.KafkaProducer;  
    import org.apache.kafka.clients.consumer.KafkaConsumer;  
    import org.apache.kafka.common.serialization.StringValueSerializer;
    public class KafkaExample {
          
        public static void main(String[] args) {
          
            // 创建生产者  
            KafkaProducer<String, String> producer = new KafkaProducer<>(  
                    new StringValueSerializer(),  
                    new StringValueSerializer()  
            );
            // 创建消费者  
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(  
                    new StringValueSerializer(),  
                    new StringValueSerializer()  
            );
            // 发送消息  
            producer.send("test-topic", "Hello, Kafka!");
            // 接收消息  
            consumer.subscribe("test-topic");  
            while (true) {
          
                ConsumerRecords<String, String> records = consumer.poll(100);  
                for (ConsumerRecord<String, String> record : records) {
          
                    System.out.printf("Received message: %s%n", record.value());  
                }  
            }  
        }  
    }

综上所述，Kafka 实现高可用、高吞吐的主要手段包括：分布式架构、顺序读写、零拷贝、文件分段、批量发送和数据压缩等技术。通过这些技术和优化，Kafka 能够在大规模消息处理场景下表现出优异的性能。

#### 10、说说kafka不重不丢？ ####

Kafka 致力于实现不重不丢，这意味着它努力确保消息不会被重复传递，也不会在传递过程中丢失。为了实现这一目标，Kafka 采用了以下关键机制：

1.  **消息复制和副本**：Kafka 使用多个副本来保存消息，每个分区的消息都有多个副本分布在不同的服务器上。这样，即使某个服务器故障，仍然可以从其他副本中获取消息。
2.  **消费者位移**：Kafka 记录了每个消费者在每个分区中的位移（offset），表示消费者已经处理到哪个位置的消息。消费者可以定期提交位移，以确保它们不会重复消费消息。
3.  **生产者确认机制**：Kafka 生产者在将消息发送到服务器后，会等待服务器的确认（acknowledgment）。只有当服务器确认接收到消息后，生产者才会认为消息已经成功发送。
4.  **事务支持**：Kafka 提供了事务支持，允许生产者在发送消息时执行事务性操作。这意味着消息要么全部成功发送，要么一个都不发送，以确保不重不丢。
5.  **幂等性生产者**：Kafka 生产者支持幂等性，即使生产者发送相同的消息多次，只有一次会生效，有助于避免重复消息。

以下是一个简单的 Java 代码示例，实现了 Kafka 生产者和消费者功能，同时展示了如何确保消息不重复和不丢失。

**生产者端**（Producer.java）：

import org.apache.kafka.clients.producer.KafkaProducer;  
    import org.apache.kafka.clients.producer.Producer;  
    import org.apache.kafka.clients.producer.ProducerRecord;  
    import org.apache.kafka.common.serialization.StringSerializer;
    import java.time.Duration;  
    import java.util.Properties;
    public class Producer {
          
        public static void main(String[] args) {
          
            // 配置生产者参数  
            Properties props = new Properties();  
            props.put("bootstrap.servers", "localhost:9092");  
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            // 创建生产者实例  
            Producer<String, String> producer = new KafkaProducer<>(props);
            // 发送消息  
            for (int i = 0; i < 10; i++) {
          
                String message = "Hello, Kafka!" + i;  
                producer.send(new ProducerRecord<>("test-topic", message));  
                System.out.printf("Sent message: %s%n", message);  
                try {
          
                    Thread.sleep(1000);  
                } catch (InterruptedException e) {
          
                    e.printStackTrace();  
                }  
            }
            // 关闭生产者  
            producer.close();  
        }  
    }

这个生产者实例使用了 KafkaProducer 类，并发送了 10 条消息到名为 test-topic 的 topic。在 main 方法中，调用 send 方法发送消息，并在发送消息后打印消息内容。生产者在运行过程中，会持续发送消息。当需要结束程序时，调用 close 方法关闭生产者。

**消费者端**（Consumer.java）：

import org.apache.kafka.clients.consumer.KafkaConsumer;  
    import org.apache.kafka.clients.consumer.Consumer;  
    import org.apache.kafka.clients.consumer.ConsumerRecord;  
    import org.apache.kafka.clients.consumer.ConsumerRecords;  
    import org.apache.kafka.common.serialization.StringSerializer;
    import java.time.Duration;  
    import java.util.Collections;  
    import java.util.Properties;
    public class Consumer {
          
        public static void main(String[] args) {
          
            // 配置消费者参数  
            Properties props = new Properties();  
            props.put("bootstrap.servers", "localhost:9092");  
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
            props.put("enable.auto.commit", "true");  
            props.put("auto.commit.interval.ms", "1000");
            // 创建消费者实例  
            Consumer<String, String> consumer = new KafkaConsumer<>(props);
            // 订阅 topic  
            consumer.subscribe(Collections.singletonList("test-topic"));
            // 消费消息  
            while (true) {
          
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
          
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());  
                }  
            }
            // 关闭消费者  
            consumer.close();  
        }  
    }

这个消费者实例使用了 KafkaConsumer 类，并订阅了名为 test-topic 的 topic。在 main 方法中，使用 poll 方法轮询消息，并在接收到消息时打印 offset、key 和 value。消费者在运行过程中，会持续接收和处理消息。当需要结束程序时，调用 close 方法关闭消费者。

#### 11、说说kafka消费端幂等性？ ####

Kafka 消费者端的幂等性是指消费者能够处理来自 Kafka 主题的消息，而不会导致重复数据或意外的结果。保持幂等性对于确保数据处理的正确性和稳定性至关重要。

以下是一些建议和方法，用于实现 Kafka 消费者端的幂等性：

1.  **消费者位移管理**：Kafka 消费者应恰当管理位移，以避免消息的重复处理。消费者应定期提交已成功处理的消息的位移，以确保它们不会再次消费相同的消息。
2.  **消息处理的幂等性**：消费者的消息处理逻辑应是幂等的，即无论处理相同的消息一次还是多次，结果应相同。这可以通过设计消息处理逻辑来实现，例如，检查消息的唯一标识符，以避免重复插入相同的数据。
3.  **幂等性标识符**：在某些处理场景中，可以在消息中包含幂等性标识符。消费者在处理消息之前检查这个标识符，以确保不会重复处理相同的消息。
4.  **事务性处理**：Kafka 支持事务，消费者可使用事务性处理来确保消息的幂等性。在处理消息之前，消费者可以启动事务，并在成功处理后提交事务。这样确保消息只会被处理一次。
5.  **异常处理**：消费者需正确处理异常情况。如果消息处理失败，消费者应能够重新处理消息而不引入额外的副作用。这可能涉及到将消息从未处理状态切换到已处理状态的机制。
6.  **幂等性测试**：对于关键的消息处理逻辑，建议编写单元测试来验证其幂等性。这些测试可以模拟重复消息处理，以确保不会引入重复数据。

以下是一个简单的 Java 示例，展示了如何实现 Kafka 消费端的幂等性：

import org.apache.kafka.clients.consumer.ConsumerRecord;  
    import org.apache.kafka.clients.consumer.ConsumerRecords;  
    import org.apache.kafka.clients.consumer.KafkaConsumer;  
    import org.apache.kafka.common.serialization.StringDeserializer;
    import java.time.Duration;  
    import java.util.Collections;  
    import java.util.Properties;
    public class KafkaConsumerIdempotenceDemo {
          
        public static void main(String[] args) {
          
            // 配置消费者参数  
            Properties props = new Properties();  
            props.put("bootstrap.servers", "localhost:9092");  
            props.put("group.id", "test-group");  
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
            props.put("auto.offset.reset", "earliest");
            // 创建消费者实例  
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            // 订阅 topic  
            consumer.subscribe(Collections.singletonList("test-topic"));
            // 消费消息  
            while (true) {
          
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));  
                for (ConsumerRecord<String, String> record : records) {
          
                    // 处理消息  
                    processMessage(record.value());
                    // 提交偏移量  
                    consumer.commit(record.offset());  
                }  
            }
            // 关闭消费者  
            consumer.close();  
        }
        private static void processMessage(String message) {
          
            // 实现幂等处理逻辑，例如使用数据库事务或悲观锁  
            // 这里仅作为示例，模拟数据库操作  
            System.out.println("Processing message: " + message);  
        }  
    }

在这个示例中，我们创建了一个 Kafka 消费者，订阅了名为 test-topic 的 topic。在 poll 方法中获取消息，并对每条消息进行幂等处理。处理完成后，使用 commit 方法提交消费偏移量。

需要注意的是，这个示例仅实现了消费端的幂等性，并未涉及生产端的消息幂等。

在实际应用中，为了确保消息的幂等性，需要在生产端和消费端都进行相应的处理。此外，Kafka 的幂等性仅保证了单个分区内的消息不重复，不同分区之间仍有可能出现重复消息。如需保证多个分区的幂等性，可以考虑使用 Kafka 的事务功能。

#### 12、说说kafka消费者群组？ ####

Kafka 消费者群组是 Kafka 中用于协同消费主题中消息的机制。它允许多个消费者协同工作，以从一个或多个主题中消费消息。

1.  **多个消费者**：一个消费者群组可以包括多个消费者。这些消费者可以在不同的应用程序或服务器上运行。
2.  **主题分区**：Kafka 主题通常被划分为多个分区，每个分区包含一部分消息。消费者群组可以同时消费多个分区中的消息。
3.  **负载均衡**：Kafka 自动分配分区给消费者，以实现负载均衡。每个分区通常只分配给一个消费者来避免重复消费。
4.  **水平扩展**：通过增加消费者，可以水平扩展消费者群组以处理更多的消息。
5.  **消费者位移管理**：Kafka 为每个消费者群组中的消费者维护位移（offset），表示它们在每个分区中的消费位置。这确保了消费者能够从上次停止的位置继续消费。
6.  **消费者协作**：在同一个消费者群组中，每个分区只能由一个消费者消费。这有助于避免重复消费。
7.  **消息处理并行性**：每个消费者可以在独立的线程中处理消息，从而提高消息的处理并行性。
8.  **自动重平衡**：如果有新的消费者加入或旧的消费者退出，Kafka 自动触发群组的重平衡，以重新分配分区，确保负载均衡。
9.  **消费者状态监控**：Kafka 提供监控工具来跟踪消费者群组的状态，包括消费速度和位移。

#### 13、说说kafka生产端分配，路由？ ####

Kafka 生产端的主要职责是将消息发送至 Kafka 集群，并确保消息被正确路由至适宜的主题和分区。在这一过程中，Kafka 采用了分区策略来对消息进行分区，从而实现消息在各个分区间的均衡分布。

以下是一些相关概念：

1.  **主题**：Kafka 中的主题作为消息的逻辑容器，生产者可以将消息发送至一个或多个主题。主题通常代表一类消息，如日志、事件或其他数据类型。
2.  **分区**：Kafka 主题可以被划分为多个分区，每个分区是消息的物理存储单元。分区可以并行处理消息，并且每个分区都有独立的偏移量（offset）来跟踪已消费的消息。
3.  **分区策略**：用于确定消息将被发送至哪个分区的规则。Kafka 提供了多种分区策略，包括轮询（Round-robin）、哈希（Hashing）和自定义分区策略。生产者可根据需求选择合适的分区策略。
4.  **轮询分区策略**：轮询分区策略是最简单的策略，它按顺序将消息发送至不同分区。这确保了消息在不同分区间的均匀分布，适用于负载均衡场景。
5.  **哈希分区策略**：哈希分区策略根据消息的键（Key）进行哈希计算，将相同键的消息路由至相同分区。这保证了具有相同键的消息始终进入同一分区。
6.  **自定义分区策略**：开发人员可以编写自定义分区策略，根据特定业务逻辑将消息路由至分区。这使得消息路由更具灵活性。
7.  **Producer API**：Kafka 为各种编程语言提供了生产者 API，方便开发人员将消息发送至 Kafka 集群，并可根据需求配置分区策略。
8.  **生产者确认**：Kafka 生产者可以配置确认机制，以确保消息成功写入分区。这包括确认（acknowledgment）机制，生产者在等待分区确认后才会认为消息发送成功。
9.  **消息分布**：消息将根据分区策略分布至不同分区，生产者可以向不同分区发送消息，以确保消息的分布和处理。

Kafka 生产端（Producer）在发送消息时，需要考虑消息的分发策略。Kafka 生产端通过路由来将消息发送到与主题（Topic）关联的分区（Partition），从而实现高效的消息处理和数据分布。以下是一个使用 Java 实现 Kafka 生产端的路由和分配的示例：

**1. 添加 Maven 依赖**：

<dependency>  
        <groupId>org.apache.kafka</groupId>  
        <artifactId>kafka-clients</artifactId>  
        <version>2.8.0</version>  
    </dependency>

**2. 创建 Kafka 生产者并配置参数**：

import org.apache.kafka.clients.producer.KafkaProducer;  
    import org.apache.kafka.clients.producer.Producer;  
    import org.apache.kafka.clients.producer.ProducerConfig;  
    import org.apache.kafka.common.serialization.StringSerializer;
    import java.util.Properties;
    public class KafkaProducerDemo {
          
        public static void main(String[] args) {
          
            // 配置生产者参数  
            Properties props = new Properties();  
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");  
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());  
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            // 创建 Kafka 生产者实例  
            Producer<String, String> producer = new KafkaProducer<>(props);
            // 发送消息  
            for (int i = 0; i < 10; i++) {
          
                String message = "Hello, Kafka!" + i;  
                producer.send(new KeyValue<>(message, message));  
            }
            // 关闭生产者  
            producer.close();  
        }  
    }

在上面的示例中，我们创建了一个 Kafka 生产者，并配置了 bootstrap.servers、key.serializer 和 value.serializer 等参数。生产者将消息发送到名为"test-topic"的主题，该主题有两个分区（Partition）。

**3. 实现路由和分配**：

Kafka 生产端通过路由和分配策略来决定将消息发送到哪个分区。这可以通过实现自定义的 Partitioner 类来实现。以下是一个简单的示例，将消息发送到分区序号小于等于消息序号的分区：

import org.apache.kafka.clients.producer.Partitioner;  
    import org.apache.kafka.common.requests.SendResult;
    import java.util.List;
    public class CustomPartitioner implements Partitioner {
          
        @Override  
        public int partition(String topic, Object key, int numPartitions) {
          
            int partition = (int) (key.hashCode() % numPartitions);  
            if (partition < 0) {
          
                partition = 0;  
            }  
            return partition;  
        }
        @Override  
        public void send(SendResult sendResult, List<Object> records) {
          
            // 实现自定义发送逻辑，例如记录发送结果  
        }  
    }

**4. 使用自定义分区器发送消息**：

在之前的生产者示例中，我们将 Partitioner 类替换为自定义的 CustomPartitioner。然后，将消息发送到与主题关联的分区：

// 创建自定义分区器  
    CustomPartitioner partitioner = new CustomPartitioner();
    // 发送消息  
    for (int i = 0; i < 10; i++) {
          
        String message = "Hello, Kafka!" + i;  
        SendResult sendResult = producer.send(new KeyValue<>(message, message), partitioner);  
        System.out.printf("Message %s sent to partition %d with offset %d%n", message, sendResult.getPartition(), sendResult.getOffset());  
    }

#### 14、说说mysql索引，为什么B+树？ ####

1.  **平衡性**：B+树是一种自平衡数据结构，能保持树的高度相对较低。这意味着在最坏情况下，查找特定条目的时间复杂度为 O(log n)，其中 n 是索引中的条目数量。这对于实现高效检索操作至关重要。
2.  **顺序访问性能**：B+树的内部节点包含指向子节点的指针，使 B+树在范围查询中具有较高效率。例如，如需查询某一范围内的数据，B+树可沿叶子节点顺序遍历，从而提高顺序访问性能。
3.  **磁盘读写性能**：B+树的节点大小通常与数据库页大小相同，有助于减少磁盘读写操作。较大节点可容纳更多键值对，降低磁盘 I/O 频率。
4.  **有序性**：B+树的叶子节点形成有序链表，便于实现范围查询和排序。数据库能快速遍历有序数据。
5.  **范围查询优势**：得益于 B+树的有序性，范围查询和排序效率较高。可快速找到指定范围内的数据，无需扫描整个表。
6.  **支持多列索引**：B+树索引支持多列组合索引，有助于应对复杂查询条件。
7.  **高扇出性能**：B+树具有高扇出性能，即每个内部节点拥有多个子节点。降低树高度，减少磁盘读取次数。
8.  **支持并发操作**：B+树索引支持并发插入和删除操作，适用于多用户、多线程的数据库环境。

#### 15、说说主键索引，二级索引？ ####

##### **1. 主键索引**： #####

a. 这是一种用于唯一标识每条记录的索引。每个表格仅能设立一个主键索引。

b. 主键索引通常用于加速检索特定记录或进行数据修改操作。

c. 主键索引要求索引列的值唯一且非空。

d. 主键索引作为表的聚集索引，意味着数据按照主键索引的顺序进行物理存储。

e. 主键索引通常具有较高的查找效能，因为它能快速定位到特定行。

##### **2. 二级索引**： #####

a. 二级索引是除主键索引以外的一种索引，用于加速特定查询条件的查找。

b. 表可以有多个二级索引，用于加速不同类型的查询，如根据非主键列的条件检索数据。

c. 二级索引的值可以重复，不要求唯一性，因为它们是辅助索引，主要用于快速定位主键值。

d. 二级索引通常包含索引列的值和对应的主键值，以便在查找时可直接找到对应行。

e. 二级索引能提升查询性能，但也会占用额外的存储空间和增加更新操作的开销。

#### 16、说说mysql ACID？ ####

**原子性**（Atomicity）：

a. 原子性确保事务是不可分割的操作单元，要么全部执行，要么全部不执行。若事务的任何部分失败，整个事务将回滚，数据库状态恢复至初始状态。

b. 原子性旨在防止不完整或部分执行的事务，确保数据库一致性。

**一致性**（Consistency）：

a. 一致性确保事务在执行前后保持数据库的一致性状态。即事务执行后，数据库应从一个一致状态转变为另一个一致状态。

b. 一致性要求事务操作遵循数据库的完整性约束和业务规则，维持数据合法性。

**隔离性**（Isolation）：

a. 隔离性确保并发执行的事务不会相互干扰，每个事务都仿佛在无其他事务干扰的情况下执行。

b. 隔离性分为不同级别（如读未提交、读已提交、可重复读和串行化），以控制并发事务间的相互影响。

**持久性**（Durability）：

a. 持久性确保事务成功提交后，其结果永久存储在数据库中，即使系统崩溃或断电也不会丢失。

b. 数据库系统通常采用日志文件实现持久性，以便在系统崩溃后恢复事务。

#### 17、mysql事务隔离级别，RR解决幻读吗？什么场景下会幻读？ ####

##### **1. 什么是幻读？** #####

幻读，又称不可重复读，是一种并发事务问题。它发生在多个事务之间，其中一个事务在某个范围内插入新行，而另一个事务在此范围内尝试查询数据。这可能导致查询事务看到新插入的行，即使在其开始查询之前这些行并不存在。幻读与脏读类似，但关注的是插入操作而非修改操作。

##### **2. RR隔离级别如何解决幻读？** #####

RR 隔离级别通过使用锁或多版本并发控制（MVCC）来解决幻读问题。在 RR 隔离级别下，事务会获取一个范围锁，确保在事务进行中查询的范围内的数据在事务结束前不会被其他事务修改或插入。

例如，如果事务 A 在 RR 隔离级别下查询某个范围的数据，另一个事务 B 想要在相同范围内插入新行，事务 B 将被阻塞，直到事务 A 完成。这样可以防止事务 A 看到事务 B 插入的新行，从而解决幻读问题。

##### **可能出现幻读问题**： #####

假设一个在线购物系统，多个用户同时浏览某个商品的库存情况。如果一个用户正在查询库存时，另一个用户刚好购买了最后一件商品，事务A可能在查询时看到商品的数量是1，但在实际购买时，库存已经为0了，这就是幻读问题。

#### 18、MVCC，事务版本号怎么生成，存在哪里？ ####

MVCC（多版本并发控制）是一种并发控制机制，允许数据库系统在同一时间点存在多个版本的数据，以支持事务隔离和并发查询。每个数据行在 MVCC 中都有一个或多个版本号，用于标识数据的不同版本。版本号的生成和存储取决于数据库管理系统的具体实现。

通常，MVCC系统中的版本号是在数据行上生成的，并且通常包括以下信息：

1.  **事务ID**（Transaction ID）：版本号通常包含生成该版本的事务的唯一标识符或 ID。这使得数据库能够跟踪哪个事务生成了哪个版本的数据。
2.  **时间戳**（Timestamp）：版本号通常包括生成该版本的时间戳。时间戳可以是事务开始或提交的时间，以及其他时间单位，用于确定版本的时间顺序。

版本号的生成和存储方式因 DBMS 而异，但通常存储在数据行的元数据中，以便系统能够在查询时识别和访问不同版本的数据。数据库系统还维护一个版本控制的数据结构，通常称为版本链或版本表，以跟踪每个数据行的不同版本及其关系。

MVCC 的主要优点是它允许高度并发的读取操作，因为每个事务都可以看到一致性的数据快照，而不会阻塞其他事务的写入操作。不同数据库管理系统的 MVCC 实现方式可能有所不同，但它们都旨在提供高并发性和事务隔离。在查询时，数据库系统会根据当前事务的 ID 或时间戳选择适当版本的数据，以确保事务之间的隔离。

#### 19、说说binlog，redolog，undolog？ ####

##### **1. Binlog**（二进制日志）： #####

a. `binlog` 是 MySQL 数据库中的二进制日志，用于记录数据库的变更操作，如插入、更新和删除。它以二进制形式记录了 SQL 语句或数据变动事件的日志，而不是实际的数据值。

b. 主要用途在于数据库的备份、主从复制和故障恢复。通过分析 `binlog`，可以还原数据库的历史状态。

##### **2. Redo Log**（重做日志）： #####

a. `redo log` 是数据库管理系统中的一种日志，主要用于记录数据变动操作。它以物理方式记录了对数据库页的更改，而非 SQL 语句。

b. 主要用途是确保事务的持久性（Durability），在数据库系统发生崩溃或故障时，可以使用 `redo log` 来重放事务，以确保数据的一致性。

##### **3. Undo Log**（撤销日志）： #####

a. `undo log` 也是数据库管理系统中的一种日志，用于记录事务的撤销操作。它包含了事务执行前的数据状态，以便在需要时回滚事务。

b. 主要用途是支持事务的回滚操作和多版本并发控制（MVCC）。在事务发生回滚时，可以使用 `undo log` 将数据还原到之前的状态。

简单记忆：

*  `binlog` 用于记录数据更改的逻辑日志，通常用于备份和复制。
 *  `redo log` 用于记录物理数据页的更改，以确保持久性。
 *  `undo log` 用于支持事务的回滚和多版本并发控制。

#### 20、挂了用什么log，主从同步用什么log？ ####

当数据库系统发生崩溃或非正常关闭时，崩溃恢复日志（通常是 `redo log`）用于重放未完成的事务，以确保数据的持久性。主从同步通常依赖于二进制日志（Binary log）来保持主数据库和从数据库之间的数据一致性，以支持数据库复制和高可用性方案。

数据库管理中存在多种不同类型的日志，各自用于不同的目的，主要包括：

**1. Crash Recovery Log**（崩溃恢复日志）：

*  用于在数据库系统发生崩溃或非正常关闭时，恢复数据至一致状态。通常，这包括数据库的 `redo log`，用于重放未完成的事务以确保数据的持久性。

**2. Binary Log**（二进制日志或Binlog）：

*  用于记录数据库的修改操作，如插入、更新和删除。它通常用于数据库备份、主从复制和数据库同步。

**3. Error Log**（错误日志）：

*  用于记录数据库系统的错误信息、警告和异常情况。这对于诊断和解决问题非常有帮助。

**4. Transaction Log**（事务日志）：

*  用于记录事务的操作，以支持数据库的事务性和回滚操作。这包括数据库的 `undo log`，它用于支持事务的回滚操作。

#### 21、说说binlog的两阶段提交？ ####

两阶段提交（Two-Phase Commit）是分布式系统中的一种事务协议，用于在多个节点上执行原子操作。

在两阶段提交中，事务分为两个阶段：预提交（Pre-Commit）和确认提交（Commit）。

以下是对两阶段提交的具体解释和 Java 实现。

##### **1. 预提交阶段**： #####

在这个阶段，事务需要在所有参与者节点上执行，并将执行结果存储在事务日志（如 MySQL 的 binlog）中。此时，事务仍然可以被回滚（Rollback）。

##### **2. 确认提交阶段**： #####

在这个阶段，事务已经完成在所有参与者节点上的执行，并且日志已经被持久化。此时，事务不能被回滚，只能向前推进（Commit）或回滚（Rollback）。

以下是 Java 实现的两阶段提交代码示例：

public class TwoPhaseCommit {
          
        private final Logger logger = LoggerFactory.getLogger(TwoPhaseCommit.class);
        private final AtomicBoolean committed = new AtomicBoolean(false);
        public void preCommit() {
          
            logger.info("Entering pre-commit stage");
            // 在这里执行事务操作，如更新、插入等  
            // ...
            logger.info("Finished pre-commit stage");  
        }
        public void commit() {
          
            if (committed.getAndUpdate(true, x -> true)) {
          
                logger.info("Entering commit stage");
                // 在这里执行提交操作，如持久化日志、发送确认消息等  
                // ...
                logger.info("Finished commit stage");  
            } else {
          
                logger.warn("Commit failed");  
            }  
        }
        public void rollback() {
          
            if (committed.getAndUpdate(false, x -> false)) {
          
                logger.info("Entering rollback stage");
                // 在这里执行回滚操作，如撤销更新、删除日志等  
                // ...
                logger.info("Finished rollback stage");  
            } else {
          
                logger.warn("Rollback failed");  
            }  
        }  
    }

在这个示例中，我们使用了一个原子布尔变量 `committed` 来标记事务是否已经进入确认提交阶段。

在预提交阶段，事务执行操作并将结果记录在日志中。

然后，事务进入确认提交阶段，执行持久化操作并发送确认消息。如果事务已经进入确认提交阶段，那么不能再回滚事务。

使用这个两阶段提交框架，可以确保事务在所有节点上的一致性和原子性。在实际应用中，还需要考虑如何处理事务日志、异常处理、并发控制等问题。

#### 22、算法题：买卖股票 ####

##### **描述**： #####

给定一个数组 `prices` ，它的第 `i` 个元素 `prices[i]` 表示一支给定股票第 `i` 天的价格。

你只能选择某一天买入这只股票，并在未来的某一天卖出。

设计一个算法来计算你所能获得的最大利润。

返回你从这笔交易中能获得的最大利润。如果无法获得任何利润，返回 `0` 。

##### **思路**： #####

1.  首先，我们需要找到买入和卖出的最佳时机，因此我们需要计算每个元素的价格差。
2.  接下来，我们需要找到最大的价格差，即最大利润。
3.  为了找到最大利润，我们可以使用双指针法，一个指针从左边开始，另一个指针从右边开始，比较左右两边的价格差，更新最大利润。

##### 实现： #####

以下是 Java 代码实现：

public class Main {
          
        public static void main(String[] args) {
          
            int[] prices = {
        7, 1, 5, 3, 6, 4};  
            System.out.println(maxProfit(prices));  
        }
        public static int maxProfit(int[] prices) {
          
            int buy = Integer.MIN_VALUE, sell = Integer.MIN_VALUE;  
            int maxProfit = 0;
            for (int i = 0; i < prices.length; i++) {
          
                int temp = Math.max(buy, prices[i] - sell);  
                buy = Math.max(buy, temp);  
                sell = Math.min(sell, prices[i] - temp);  
                maxProfit = Math.max(maxProfit, sell - buy);  
            }
            return maxProfit;  
        }  
    }

当输入 `[7, 1, 5, 3, 6, 4]` 时，输出结果为 `5`，表示从这笔交易中能获得的最大利润为 5。

### 尼恩说在最后 ###

在尼恩的（50+）读者社群中，很多、很多小伙伴需要进大厂、拿高薪。

尼恩团队，会持续结合一些大厂的面试真题，给大家梳理一下学习路径，看看大家需要学点啥？

前面用多篇文章，给大家介绍阿里、百度、字节、滴滴的真题：

《[炸裂，靠“吹牛”过京东一面，月薪40k][40k]》

《[太猛了，靠“吹牛”过顺丰一面，月薪30K][30K]》

《[问懵了…美团一面索命44问，过了就60W+][44_60W]》

《[炸裂了…京东一面索命40问，过了就50W+][40_50W]》

《[问麻了…阿里一面索命27问，过了就60W+][27_60W]》

《[百度狂问3小时，大厂offer到手，小伙真狠！][3_offer]》

《[饿了么太狠：面个高级Java，抖这多硬活、狠活][Java]》

《[字节狂问一小时，小伙offer到手，太狠了！][offer]》

《[收个滴滴Offer：从小伙三面经历，看看需要学点啥？][Offer]》

这些真题，都会收入到 史上最全、持续升级的 PDF电子书 《[尼恩Java面试宝典][Java_PDF]》。

本文收录于 《尼恩Java面试宝典》。

基本上，把尼恩的 《尼恩Java面试宝典》吃透，大厂offer很容易拿到滴。另外，下一期的 大厂面经大家有啥需求，可以发消息给尼恩。

### 尼恩技术圣经系列PDF ###

*  《[NIO圣经：一次穿透NIO、Selector、Epoll底层原理][NIO_NIO_Selector_Epoll]》
 *  《[Docker圣经：大白话说Docker底层原理，6W字实现Docker自由][Docker_Docker_6W_Docker]》
 *  《[K8S学习圣经：大白话说K8S底层原理，14W字实现K8S自由][K8S_K8S_14W_K8S]》
 *  《[SpringCloud Alibaba 学习圣经，10万字实现SpringCloud 自由][SpringCloud Alibaba _10_SpringCloud]》
 *  《[大数据HBase学习圣经：一本书实现HBase学习自由][HBase_HBase]》
 *  《[大数据Flink学习圣经：一本书实现大数据Flink自由][Flink_Flink]》
 *  《[响应式圣经：10W字，实现Spring响应式编程自由][10W_Spring]》
 *  《[Go学习圣经：Go语言实现高并发CRUD业务开发][Go_Go_CRUD]》

……完整版尼恩技术圣经PDF集群，请找尼恩领取

> 《尼恩 架构笔记》《尼恩高并发三部曲》《尼恩Java面试宝典》PDF，请到下面公号【技术自由圈】取↓↓↓

[4ee7204a171348eb8b1f8a905ed6cdc1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/6c7f565c6a3a47678ce56fd308c721d8.png
[Java_PDF]: https://blog.csdn.net/crazymakercircle/article/details/132927258
[b45760710790930455759bf344bd6355.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/19/53f6de4a6eda4ed2928c0b7244d41a86.png
[40k]: https://blog.csdn.net/crazymakercircle/article/details/132150378
[30K]: https://blog.csdn.net/crazymakercircle/article/details/132000420
[44_60W]: https://blog.csdn.net/crazymakercircle/article/details/131678859
[40_50W]: https://blog.csdn.net/crazymakercircle/article/details/131615296
[27_60W]: https://blog.csdn.net/crazymakercircle/article/details/131147192
[3_offer]: https://blog.csdn.net/crazymakercircle/article/details/131027012
[Java]: https://blog.csdn.net/crazymakercircle/article/details/130989697
[offer]: https://blog.csdn.net/crazymakercircle/article/details/130956376
[Offer]: https://blog.csdn.net/crazymakercircle/article/details/129317821
[NIO_NIO_Selector_Epoll]: https://blog.csdn.net/crazymakercircle/article/details/133186797
[Docker_Docker_6W_Docker]: https://blog.csdn.net/crazymakercircle/article/details/129482349
[K8S_K8S_14W_K8S]: https://blog.csdn.net/crazymakercircle/article/details/130073643
[SpringCloud Alibaba _10_SpringCloud]: https://blog.csdn.net/crazymakercircle/article/details/129559428
[HBase_HBase]: https://blog.csdn.net/crazymakercircle/article/details/132589856
[Flink_Flink]: https://blog.csdn.net/crazymakercircle/article/details/132347440
[10W_Spring]: https://blog.csdn.net/crazymakercircle/article/details/129022714
[Go_Go_CRUD]: https://blog.csdn.net/crazymakercircle/article/details/131383065