Yifeng_20230911

itlaohuo included in interview

2023-09-11 16829 words 34 minutes

Contents

1、可重入锁

1.1 理解

可重入锁，也叫做递归锁，从名字上理解就是可以再进入的锁，重入性是指任意线程在获取到锁之后能够再次获取该锁而不会被锁阻塞

两个条件：

1 线程再次获取锁，2 可释放，线程重复n次获取了锁，随后在第n次释放该锁后，其它线程能够获取到该锁

JAVA中的实现

synchronized 关键字所使用的锁也是可重入的
ReentrantLock继承自 Lock接口
通过Sync 类，即自定义的同步组件实现，它是 ReentrantLock 里面的一个内部类，它继承自AQS(AbstractQueuedSynchronizer)，Sync 有两个子类：公平锁 FairSync 和非公平锁 NonfairSync
NonfairSync 的 lock() 方法：

1 CAS自旋操作state(0,1)
2 失败则 aqs.acquire(1),在acquire(1)中是一套锁抢占的模板,会先调tryAcquire,tryAcquire() 这个钩子方法去尝试获取锁，这个方法就是在 NonfairSync.tryAcquire()下的 nonfairTryAcquire().(先尝试CAS，state！=0就接着判断是否同一个线程所持有,如果是设置state+1)
3 如果nonfairTryAcquire的以上两种情况都不通过，则返回失败false，就会则进入 acquireQueued() 流程，也就是基于CLH队列的抢占模式；进入的时候也会去执行一次获取锁的操作，如果还是获取不到，就调用LockSupport.park() 将当前线程挂起。那么当前线程什么时候会被唤醒呢？当持有锁的那个线程调用 unlock() 的时候，会将CLH队列的头节点的下一个节点上的线程唤醒，调用的是 LockSupport.unpark() 方法。

引申 AQS原理

AQS（AbstractQueuedSynchronizer）是Java中用于构建锁和同步器的底层框架。它提供了一个灵活的方式来实现各种同步机制，如ReentrantLock、Semaphore、CountDownLatch等，并且也可用于构建自定义的同步器。AQS的核心思想是基于队列的等待，通过管理一个等待队列来实现线程的排队和唤醒。
AQS底层使用了模板方法模式，自定义同步器时需要重写下面几个AQS提供的模板方法：

1
2
3
4
5
isHeldExclusively()//该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int)//独占方式。尝试获取资源，成功则返回true，失败则返回false。
tryRelease(int)//独占方式。尝试释放资源，成功则返回true，失败则返回false。
tryAcquireShared(int)//共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
tryReleaseShared(int)//共享方式。尝试释放资源，成功则返回true，失败则返回false。

参考文章[JUC锁: 锁核心类AQS详解}(https://pdai.tech/md/java/thread/java-thread-x-lock-AbstractQueuedSynchronizer.html)

2、synchronized锁升级过程

Java 详细对象结构: 对象头（Mark Word、Class Pointer、数组长度三个字段组成），对象体，对齐字节 ; Mark Word主要用来表示当前 Java 对象的线程锁状态以及 GC 的标志，可以表示 4 种不同的锁状态
无锁状态：初始状态，没有线程占用锁，线程可以无竞争地进入临界区。
偏向锁状态：当只有一个线程访问临界区时，该线程会偏向于锁，以提高性能。如果其他线程尝试进入临界区，偏向锁会升级为轻量级锁。
轻量级锁状态：多个线程竞争同一把锁，但还没有争用到达一定程度。这时，锁会升级为轻量级锁，使用CAS操作来尝试获取锁，避免了传统的重量级锁的开销。
重量级锁状态：多个线程争用锁的情况下，试图抢占的线程自旋达到阈值，就会停止自旋，那么此时锁就会膨胀成重量级锁，锁会升级为重量级锁，通过操作系统的互斥原语来实现线程的阻塞和唤醒。
锁的升级过程通常是自动的，根据竞争情况和线程行为，锁会在不同状态之间切换以优化性能。这个过程在JVM内部进行管理，开发者一般不需要显式干预。详细了解锁升级过程可以帮助优化多线程程序的性能。

3、countDownLauch工作原理

其底层是由AQS提供支持，所以其数据结构可以参考AQS的数据结构，而AQS的数据结构核心就是两个虚拟队列: 同步队列sync queue 和条件队列condition queue，不同的条件会有不同的条件队列。CountDownLatch典型的用法是将一个程序分为n个互相独立的可解决任务，并创建值为n的CountDownLatch。
核心函数 - await函数；此函数将会使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断。对CountDownLatch对象的await的调用会转发为对Sync的acquireSharedInterruptibly(从AQS继承的方法)方法的调用。

4、说下Spring IOC，以及中间用了什么设计模式

使用对象时候由主动new对象转换成由外部提供对象,此过程中对象的创建权由程序转移到外部，这种思想叫做控制反转 Spring技术对此提供的实现 Spring提供了一个容器，称为IOC容器，用来充当IOC思想中的外部 IOC容器负责对象的创建、初始化等一系列工作，被创建或被管理的对象在IOC容器中统称为Bean。

1、工厂模式

Spring中在各种BeanFactory以及ApplicationContext创建中都用到了典型的工厂方法模式 ###　2、单例模式在Spring中，所有的bean默认都是单例创建的。在创建bean的代码中我们经常看到Singleton这个单词。下面我们通过代码看看单例是怎么实现的。 AbstractBeanFactory.doGetBean()

3、策略模式

在依赖注入的过程中，Spring会调用ApplicationContext 来获取Resource的实例。然而，Resource 接口封装了各种可能的资源类型，包括了：UrlResource，ClassPathResource，FileSystemResource等，Spring需要针对不同的资源采取不同的访问策略。在这里，Spring让ApplicationContext成为了资源访问策略的“决策者”。在资源访问策略的选择上，Spring采用了策略模式。当 Spring 应用需要进行资源访问时，它并不需要直接使用 Resource 实现类，而是调用 ApplicationContext 实例的 getResource() 方法来获得资源，ApplicationContext 将会负责选择 Resource 的实现类，也就是确定具体的资源访问策略，从而将应用程序和具体的资源访问策略分离开来。

4、装饰器模式

Spring中类中带有Wrapper的都是包装类

5 代理模式

AOP等等

6 责任链模式

Filter等

5、说下SpringBoot和Spring的区别?

Spring和Spring Boot基于IOC AOP理念实现，Spring Boot集成了Spring。
对于我来说，Spring框架就是提供了IOC容器、控制反转、依赖注入以及一些模块，简化了大量的代码，便捷了程序的开发，节省了开发时间，提高了效率。
在我看来Spring Boot框架是对Spring框架的补充，它消除了Spring框架配置XML的麻烦事，完善了Spring框架的开发环境，使我们可以更加高效的完成编程，并且为我们提供了 spring-boot-starter-xxx 依赖，写大量的配置，现在用Spring Boot只需要导入相关依赖。
Spring 和 Spring Boot的最大的区别在于Spring Boot的自动装配原理
在我看来是SpringBoot（spring）将java推上了神座。

6、线程池核心参数、以及说下工作原理?工作中使用什么工作队列?以及用了什么拒绝策略?

线程池的核心参数包括以下几个：

1. 核心线程数（corePoolSize）：线程池中最小的线程数量。即使是空闲状态，核心线程也不会被回收。
1. 最大线程数（maximumPoolSize）：线程池中最大的线程数量。当任务量增加时，线程池会动态地创建新的线程，直到达到最大线程数。
1. 空闲线程存活时间（keepAliveTime）：当线程池中的线程数量超过核心线程数时，多余的空闲线程的存活时间。超过这个时间，空闲线程会被回收。
1. 阻塞队列（workQueue）：用于存放待执行的任务的队列。当线程池中的线程都在忙于执行任务时，新的任务会被放入队列中等待执行。
1. 线程工厂（threadFactory）：用于创建新线程的工厂类。
1. 拒绝策略（rejectedExecutionHandler）：当线程池已经达到最大线程数，并且队列也已满时，新的任务无法被执行时的处理策略。常见的拒绝策略有：抛出异常、丢弃任务、丢弃队列中最旧的任务、调用提交任务的线程来执行任务。

线程池的执行原理如下：

1. 当有新的任务提交到线程池时，线程池会根据核心线程数和当前线程池中的线程数量来决定是创建新的线程还是将任务放入阻塞队列中。
1. 如果当前线程池中的线程数量小于核心线程数，线程池会创建新的线程来执行任务。
1. 如果当前线程池中的线程数量达到核心线程数，线程池会将任务放入阻塞队列中等待执行。
1. 如果阻塞队列已满，但线程池中的线程数量还没有达到最大线程数，线程池会创建新的线程来执行任务。
1. 如果线程池中的线程数量达到最大线程数，并且阻塞队列也已满，根据配置的拒绝策略来处理新的任务。
1. 当线程池中的线程执行完任务后，会继续从阻塞队列中取出任务来执行，直到线程池关闭或者没有待执行的任务。
通过合理地配置线程池的参数，可以根据系统的需求来控制线程的数量和任务的执行方式，提高系统的并发性能和资源利用率。

Java原生线程池的执行流程：

核心线程都在工作时，先丢队列，队列满了才考虑创建新的线程直到最大线程数
1）当前线程数量是否小于核心线程数，如果是创建核心线程执行任务，否则2；
2）尝试通过offer方法不阻塞地把任务丢到阻塞队列，如果成功就返回，否则3；
3）判断当前线程数量是否小于最大线程数量，如果是创建空闲线程执行任务，否则4；
4）执行拒绝策略，默认是抛出异常；

Java原生线程池的缺点与适用场景

缺点：

1 容易堆积任务，当线程数量大于等于核心线程数量后，就把任务丢到队列，而不是创建空闲线程执行任务，如果队列是无界，可能会造成任务堆积从而发生OOM

适用场景：

1 适合cpu密集型任务，而且任务时间不宜过长，否则会造成队列里面任务的堆积；

Tomcat线程池

适合IO密集型任务，优先增加到最大线程数；其次才是放入队列
Tomcat重点改造的是queue的offer()。即在向queue放入任务时，若发现未达到最大线程数，那么offer()返回false，即放入队列失败。此时，便继续开启maximumPoolSize线程。
通过实现自定义队列来完成逻辑的改造；自定义线程池ThreadPoolExecutor

7、说说JVM垃圾搜集器? docker容器4G，你给JVM最大堆内存分配多少?

docker中给到JVM一般分配3G约75%到80%

java默认是通过/proc/meminfo来取服务器内存数据的，而docker容器中/proc/meminfo文件记录的是宿主机的内存参数，jvm堆最大值默认是服务器的1/4，这样就可能会导致docker容器的限制内存最大值可能小于jvm堆内存上限。
现象就是在jvm触发fullGC或者majorGC之前堆内存就已经快达到容器最大内存限制，导致容器将java进程杀死
oracle在JDK 1.8u131开始对Docker容器进行支持(-XX:+UseContainerSupport )，以解决上述问题，并且在JDK1.8u191版本进行了完善，因此升级JDK版本到1.8u191之后的版本即可解决问题 .

CMS

CMS（Concurrent Mark Sweep）收集器是一种以获取最短回收停顿时间为目标的收集器。仅用于老年代
4大步骤

初始标记（CMS initial mark）并发标记（CMS concurrent mark）重新标记（CMS remark）并发清除（CMS concurrent sweep）

其中，初始标记、重新标记这两个步骤仍然需要 Stop-the-world。初始标记仅仅只是标记一下 GC Roots 能直接关联到的对象，速度很快，并发标记阶段就是进行 GC Roots Tracing 的过程，而重新标记阶段则是为了修正并发标记期间因用户程序继续运作而导致标记产生变动的那一部分对象的标记记录，这个阶段的停顿时间一般会比初始阶段稍长一些，但远比并发标记的时间短。
标记-清除算法将产生大量的内存碎片这对新生代来说是难以接受的，因此新生代的收集器并未提供 CMS 版本。

G1

G1 的特点是保持高回收率的同时，减少停顿。
G1 算法取消了堆中年轻代与老年代的物理划分，但它仍然属于分代收集器。G1 算法将堆划分为若干个区域，称作 Region，如下图中的小方格所示。一部分区域用作年轻代，一部分用作老年代，另外还有一种专门用来存储巨型对象的分区。
G1 算法允许通过 JVM 参数设置 Region 的大小，范围是 1～32MB，可以设置期望的最大 GC 停顿时间等

ZGC与分代ZGC

ZGC在标记、转移和重定位阶段几乎都是并发的，这是ZGC实现停顿时间小于10ms目标的最关键原因。

8、说下哪些场景需要打破双亲委派机制?

当某个类加载器需要加载某个.class文件时，它首先把这个任务委托给他的上级类加载器，递归这个操作，如果上级的类加载器没有加载，自己才会去加载这个类。
BootstrapClassLoader 《== ExtClassLoader 《== AppClassLoader 《== CustomClassLoader（用户自定义）
1 防止加载同一个.class。通过委托去询问上级是否已经加载过该.class，如果加载过了，则不需要重新加载。保证了数据安全。
2 保证核心.class不被篡改。通过委托的方式，保证核心.class不被篡改，即使被篡改也不会被加载，即使被加载也不会是同一个class对象，因为不同的加载器加载同一个.class也不是同一个Class对象。这样则保证了Class的执行安全。
自定义类加载器：

如果不想打破双亲委派模型，就重写ClassLoader类中的findClass()方法即可，无法被父类加载器加载的类最终会通过这个方法被加载而如果想打破双亲委派模型则需要重写ClassLoader类loadClass()方法（当然其中的坑也不会少）。典型的打破双亲委派模型的框架和中间件有tomcat与osgi

9、CPU飙高、JVM内存泄漏如何解决?

可以使用jps查看java程序的资源占用，下面介绍使用top命令的
top 命令查找CPU和内存信息，根据排行可以取得对应进程id （如：top -p 452 查找指定进程的cpu内存信息）
top -H -p 452 // 查看指定进程452的所有线程的CPU内存信息
记录CPU或内存占用高的线程ID，转化为对应的16进制，堆栈中搜索用( print命令（printf “%x \n” 15）：或其他工具）
jstack -l 108032 | grep 1a601 -A 74 ；打印线程108033（父进程是108032）的快照。 108033转成11a601

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
linux1@linuxonetest:~$ jstack -l 108032 | grep 1a601 -A 74
"main" #1 prio=5 os_prio=0 cpu=1447919.07ms elapsed=1448.76s tid=0x000003ff7c016c30 nid=0x1a601 runnable  [0x000003ff823fd000]
   java.lang.Thread.State: RUNNABLE
        at main.main(main.java:11)
        at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(java.base@17.0.8.1/Native Method)
        at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(java.base@17.0.8.1/NativeMethodAccessorImpl.java:77)
        at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(java.base@17.0.8.1/DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(java.base@17.0.8.1/Method.java:568)
        at com.sun.tools.javac.launcher.Main.execute(jdk.compiler@17.0.8.1/Main.java:419)
        at com.sun.tools.javac.launcher.Main.run(jdk.compiler@17.0.8.1/Main.java:192)
        at com.sun.tools.javac.launcher.Main.main(jdk.compiler@17.0.8.1/Main.java:132)

   Locked ownable synchronizers:
        - None

可以看到是代码 11行： at main.main(main.java:11) ; 查看代码可以知道代码里面while循序一直在处理任务，没有让出CPU，导致CPU飙升

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
 1
  2 public class main{
  3
  4
  5 public static void main(String[] args){
  6         while(true){
  7             // System.out.println("Hello wordl");
  8             //
  9             int a =  10000 + 50000000;
 10             a =  a / 1000 ;
 11         }
 12
 13 }
 14 }

内存泄漏如何排查

1
2
3
4
5
6
# 外部触发 调用 循环次数最好通过外部控制
while(true) {
    		System.out.println("一刻不停的处理任务");
    		list.add(new String(new byte[1024 * 1024]) + "处理任务分配一个1M的对象，序号为 " + i);
        // sleep 100 ms
}

首先通过jps查看进程PID是2785，然后通过top -p 2785发现内存升高到28.1%。
多次使用jstat -gc 2785查看GC日志;发现OU的内存，也就是老年代的使用内存在一直增加，有对象一直处于存活，并且一直有新对象产生。接下来就通过堆栈储文件查看内存的使用情况。
通过jmap -dump:format=b,file=height-cpu.bin 2785生产堆转储快照文件。使用Eclipse的内存分析器工具（MAT）打开height-cpu.bin文件进行堆内存分析
可以发现一个对象占了96%以上的内存。打开leak suspects页面，可以查看内存泄漏的原因 ; 点击详情可以查看具体的对象 …

GC 统计

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
jstat -gc 452 1000 10 #每秒查询一次查询10次 (452进程id  1000ms) 
# 备注 ： s --> Surive ; E --> Eden , O --> Old M --> Metaspace 不是method  C --> Capacity(容量) U--> Used  Y--> Young FGC--> Full GC  

第一行表示在应用程序启动后第一次采样时，各个内存区域和垃圾回收的情况。

例如，你可以看到：

S0C是0.0，表示survivor space 0没有分配任何空间；

S1C是4096.0，表示survivor space 1分配了4096 KB的空间；

S0U是0.0，表示survivor space 0没有使用任何空间；

S1U是4096.0，表示survivor space 1已经使用了全部空间；

EC是309248.0，表示eden space分配了309248 KB的空间；

EU是236544.0，表示eden space已经使用了236544 KB的空间；

OC是183296.0，表示old space分配了183296 KB的空间；

OU是125409.0，表示old space已经使用了125409 KB的空间；

MC是140168.0，表示metaspace分配了140168 KB的空间；

MU是135553.7，表示metaspace已经使用了135553.7 KB的空间；

CCSC是15488.0，表示compressed class space分配了15488 KB的空间；

CCSU是13814.7，表示compressed class space已经使用了13814.7 KB的空间；

YGC是236，表示从应用程序启动到采样时发生了236次young generation垃圾回收；

YGCT是3.545，表示从应用程序启动到采样时young generation垃圾回收花费了3.545秒；

FGC是0，表示从应用程序启动到采样时没有发生full GC；

FGCT是0.000，表示从应用程序启动到采样时full GC花费了0秒；

CGC是12，表示从应用程序启动到采样时发生了12次concurrent GC；

CGCT是0.188，表示从应用程序启动到采样时concurrent GC花费了0.188秒；

GCT是3.733，表示从应用程序启动到采样时垃圾回收花费了总共3.733秒。

* The jstat Command - Oracle. https://docs.oracle.com/en/java/javase/14/docs/specs/man/jstat.html 

堆内存统计

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
jstat -gccapacity 452
NGCMN：新生代最小容量
NGCMX：新生代最大容量
NGC：当前新生代容量
S0C：第一个幸存区大小
S1C：第二个幸存区的大小
EC：伊甸园区的大小
OGCMN：老年代最小容量
OGCMX：老年代最大容量
OGC：当前老年代大小
OC:当前老年代大小
MCMN:最小元数据容量
MCMX：最大元数据容量
MC：当前元数据空间大小
CCSMN：最小压缩类空间大小
CCSMX：最大压缩类空间大小
CCSC：当前压缩类空间大小
YGC：年轻代gc次数
FGC：老年代GC次数
CGC:

10、redis数据结构?

string(字符串)、list(列表)、hash(字典)、set(集合) 、 zset(有序集合)和 Stream(流)
zset(有序集合)的内部实现用的是一种叫做 「跳跃表」 的数据结构

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
typedef struct dictht {
    // 哈希表数组
    dictEntry **table;
    // 哈希表大小
    unsigned long size;
    // 哈希表大小掩码，用于计算索引值，总是等于 size - 1
    unsigned long sizemask;
    // 该哈希表已有节点的数量
    unsigned long used;
} dictht;

typedef struct dict {
    dictType *type;
    void *privdata;
    // 内部有两个 dictht 结构
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    unsigned long iterators; /* number of iterators currently running */
} dict;

可以从上面的源码中看到，实际上字典结构的内部包含两个 hashtable，通常情况下只有一个 hashtable 是有值的，但是在字典扩容缩容时，需要分配新的 hashtable，然后进行渐进式搬迁; 大字典的扩容是比较耗时间的，需要重新申请新的数组，然后将旧字典所有链表中的元素重新挂接到新的数组下面，这是一个 O(n) 级别的操作，作为单线程的 Redis 很难承受这样耗时的过程，所以 Redis 使用渐进式 rehash 小步搬迁
Stream 类型

Redis5.0带来了Stream类型。从字面上看是流类型，但其实从功能上看，应该是Redis对消息队列（MQ，Message Queue）的完善实现。用过Redis做消息队列的都了解，基于Reids的消息队列实现有很多种，例如： PUB/SUB，订阅/发布模式基于List的 LPUSH+BRPOP 的实现基于Sorted-Set的实现 Redis Stream的结构如上图所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的ID和对应的内容。消息是持久化的，Redis重启后，内容还在用来实现典型的消息队列。该Stream类型的出现，几乎满足了消息队列具备的全部内容

参考： https://cloud.tencent.com/developer/article/1667574

11、说下mysqlinnodb索引数据结构?

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
表空间
    段（segment）
    区（extent）
    页（page）
    行（row）
索引结构
    聚簇索引
    辅助索引
为什么使用 B+ 树实现索引？
二叉查找树：不平衡
    平衡二叉树：旋转耗时
    红黑树：树太高
B 树：为磁盘而生
B+ 树：更进一步的优化

页是 InnoDB 存储引擎的最小管理单位，每页大小默认是 16KB
InnoDB的索引类型分为主键索引和非主键索引
主键索引的叶子节点存的是整行数据。在 InnoDB 里，主键索引也被称为聚簇索引（clustered index）。整张表的数据其实就是存储在聚簇索引中的，聚簇索引就是表
非主键索引的叶子节点内容是主键的值。在 InnoDB 里，非主键索引也被称为二级索引（secondary index）。
B 树中每个节点（包括叶节点和非叶节点）都存储真实的数据，B+ 树中只有叶子节点存储真实的数据，非叶节点只存储键。在 MySQL 中，这里所说的真实数据，可能是行的全部数据（如 InnoDB 的聚簇索引），也可能只是行的主键（如 InnoDB 的辅助索引），或者是行所在的地址（如 MyIsam 的非聚簇索引）。

12、说下模板模式在Spring IoC的应用

ClassPathXmlApplicationContext AnnotationConfigApplicationContext ….
**refresh()**方法中大量的钩子方法： onRefresh(),postProcessBeanFactory()

13、mysql事务有哪些?默认事务是什么?

14、说下kafka生产者、消费者整套流程?生产者批量提交配置，项目是如何配

1 生产者定期向主题发送消息：主线程准备数据 ==》数据缓冲起来 ==》 sender线程批量发送数据；
2 Broker存储该特定主题配置的分区中的所有消息。它确保消息在分区之间平等共享
3 消费者订阅特定主题。 4 一旦消费者订阅主题，Kafka 将向消费者提供主题的当前偏移，并且还将偏移保存在 Zookeeper 系统中
5 消费者将定期请求 Kafka (如100 Ms)新消息。
6 一旦 Kafka 收到来自生产者的消息，它将这些消息转发给消费者。
7 消费者将收到消息并进行处理。
8 一旦消息被处理，消费者将向 Kafka 代理发送确认。
9 一旦 Kafka 收到确认，它将偏移更改为新值，并在 Zookeeper 中更新它。
10 以上流程将重复，直到消费者停止请求。
11 消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息。
配置；三个条件，满足任一即会批量发送：
batch-size ：超过收集的消息数量的最大量。默认16KB
buffer-memory ：超过收集的消息占用的最大内存 , 默认32M
linger.ms ：超过收集的时间的最大等待时长，单位：毫秒
OSR（Out of-Sync Replicas）即与Leader数据不一致的副本列表
ISR In-sync Replicas（副本同步列表）。ISR中的副本都是与Leader数据一致的副本，不在ISR中的Follower副本是与Leader数据不一致的。Leader副本天然就在 ISR 中。
AR（Assigned Replicas）；即已分配的副本列表，是指某个Partition的所有副本。AR = ISR + OSR
HW 俗称高水位，HighWatermark的缩写

16、zk的特性? zk实现注册中心的原理?服务提供者挂了，zk会怎么样?

Zookeeper的主要作用是为分布式系统提供协调服务,包括但不限于:分布式锁,统一命名服务,配置管理,负载均衡,主控服务器选举以及主从切换等。
是以高吞吐量为目标进行设计的,故而在读多写少的场合有非常好的性能表现
Zookeeper自身通常也以分布式形式存在。一个Zookeeper服务通常由多台服务器节点构成,只要其中超过一半的节点存活,Zookeeper即可正常对外提供服务,所以Zookeeper也暗含高可用的特性
Zookeeper集群的任意一个服务端节点都可以直接响应客户端的读请求
Zookeeper将全量数据存储于内存中,从内存中读取数据不需要进行磁盘IO,速度要快得多
在Zookeeper集群中,分别有Leader,Follower和Observer三种类型的服务器角色
Leader服务器在整个正常运行期间有且仅有一台,集群会通过选举的方式选举出Leader服务器,由它同统一处理集群的事务性请求以及集群内各服务器的调度.
数据模型：Zookeeper将数据存储于内存中,具体而言,Znode是存储数据的最小单元。而Znode被以层次化的结构进行组织,形容一棵树
Znode按其生命周期的长短可以分为持久结点(PERSISTENT)和临时结点(EPHEMERAL);在创建时还可选择是否由Zookeeper服务端在其路径后添加一串序号用来区分同一个父结点下多个结点创建的先后顺序。经过组合就有4种Znode结点类型
Znode结构主要由存储于其中的数据信息和状态信息两部分构成： czxid（创建时事务id），mzxid，version 等等
Zookeeper中可以通过Watcher来实现事件监听机制。客户端可以向服务端注册Watcher用以监听某些事件,一旦该事件发生,服务端即会向客户端发送一个通知
Zookeeper采用ZAB(Zookeeper Atomic Broadcast)协议来保证分布式数据一致性。ZAB并不是一种通用的分布式一致性算法,而是一种专为Zookeeper设计的崩溃可恢复的原子消息广播算法。ZAB协议包括两种基本模式:崩溃恢复模式和消息广播模式。
作为注册中心，可用性的要求要高于一致性！
在 CAP 模型中，Zookeepe 整体遵循一致性（CP）原则，即在任何时候对 Zookeeper 的访问请求能得到一致的数据结果，但是当机器下线或者宕机时，不能保证服务可用性

Leader服务器的选举流程

实现注册中心的原理

服务提供者（集成了ZK客户端）在服务启动时，会通过ZK客户端与ZK服务端建立连接，将服务提供者信息（提供者的IP地址、端口、服务接口信息等）注册到ZooKeeper服务端，这时会在ZooKeeper服务端生成一个临时节点来存储这些服务信息，这就是服务提供者的注册操作。
服务消费者（集成了ZK客户端）在服务启动时，会通过ZK客户端与ZK服务端建立连接，拉取自己感兴趣的服务提供者信息并缓存到本地，同时也会对自己感兴趣的节点（比如服务提供节点）注册Watcher监听。后续发起远程调用时，会从本地缓存的服务提供者信息通过负载均衡等策略选择一台服务提供者发起服务调用。
Watcher监听机制实现扩容与下线通知。比如提供者挂了，zk服务器心跳检测机制发现下线了会将临时节点删除，由于服务端监听的数据状态发生变化了，ZooKeeper 就会主动通知发送相应事件信息给相关会话客户端，客户端就会在本地响应式的回调相关 Watcher 的 Handle。

17、说下mysql都有哪些日志文件?每个文件的作用是什么?

MySQL 中有六种日志文件，分别是：

重做日志（redo log）和回滚日志（undo log）

是 InnoDB 存储引擎的日志
事务的隔离性由锁来实现，原子性、一致性、持久性通过数据库的 redo log 或 redo log 来完成。redo log 又称为重做日志，用来保证事务的持久性，借助redo log，InnoDB可以保证数据即使异常发生了重启，之前提交的记录也不会丢失。（crash-safe）；undo log 用来保证事务的原子性和 MVCC
InnoDB 记录了对数据文件的物理更改，并保证总是日志先行，也就是所谓的 WAL，即在持久化数据文件前，保证之前的 redo 日志已经写到磁盘。生成的redo log得先保存起来，但是又不能在没commit的时候就直接写入到redo log文件里，redo log buffer就是用来存储redo日志的，真正的将日志写入到redo log文件(ib_logfile+数字)是在执行commit语句的时候执行。
undo log 有两个作用：提供回滚和多版本并发控制下的读(MVCC)，也即非锁定读
在数据修改的时候，不仅记录了redo，还记录了相对应的 undo，如果因为某些原因导致事务失败或回滚了，可以借助该 undo 进行回滚

二进制日志（binlog）

其主要是用来记录对 MySQL 数据更新或潜在发生更新的 SQL 语句，并以 “事务”的形式保存在磁盘中。
具体的写入时间：在事务提交的时候，数据库会把 binlog cache 写入 binlog 文件中，但并没有执行fsync()操作，即只将文件内容写入到 OS 缓存中。随后根据配置判断是否执行 fsync。
Row格式：Row 格式仅保存记录被修改细节，不记录 sql 语句上下文相关信息。新版本的 MySQL 默认是 Row 格式
Statement格式：每一条会修改数据的 sql 都会记录在 binlog 中
Mixed格式：是以上两种形式的结合。
不过，新版本的 MySQL 对 Row 模式也做了优化，并不是所有的修改都会完全以 Row 形式来记录，像遇到表结构变更的时候就会以 Statement 模式来记录，如果 SQL 语句确实就是 update 或者 delete 等修改数据的语句，那么还是会记录所有行的变更；因此，现在一般使用 Row 即可

错误日志（errorlog）

慢查询日志（slow query log）

一般查询日志（general log）

中继日志（relay log）

主库写入binlog后，从库的io线程会读取主库的binlog，转储位本地的中继日志relog
从库的sql线程会读取relog日志并在本地执行，达成与主库数据一致
同步策略：异步：主库写完redolog就返回提交成功；半同步：从库读取binlog转储位relog后返回ack确认信息后主库返回提交成功；同步：从库读取relog执行万事务后才可以返回提交成功。一般都是半同步，但从库长时间未响应会退化未异步复制。

总结

首先 InnoDB 完成一次更新操作的具体步骤：

开启事务
查询待更新的记录到内存，并加 X 锁
记录 undo log 到内存 buffer
更改内存中的数据记录
记录 redo log ，prepare阶段
写binlog
提交事务，触发 redo log 刷盘
事务结束
整个update语句中牵涉到写redo log和binlog，并且redo log在前，binlog在后，并且redo log的写入被拆分成了prepare和commit两个步骤，这就是两阶段提交在数据库中的应用。

两阶段提交如何保证日志逻辑的一致性

假设redo log处于prepare阶段发生了crash，此时由于binlog没写，redo log也没有提交，所以在崩溃恢复时这个事务会回滚，binlog没写所以也不会传到备库。
假设binlog写完，但是redo log还没commit之前发生了crash，此时MySQL在崩溃恢复时会有一定的处理逻辑？
如果redo log里面的事务是完整的（有commit标识），则直接提交
如果redo log里面只有完整的prepare，则判断对应的事务binlog是否存在且完整，如果完整则提交事务，否则回滚事务。
redo log和binlog如何关联？

redo log和binlog有一个共同的数据字段是XID，在崩溃恢复时，会顺序扫描redo log：
如果redo log既有prepare，又有commit，则直接提交
如果redo log只有prepare，则会拿着XID去找binlog，如果binlog里面有则提交，否则回滚

18、说下mysql 数据结构文件有哪些?

在MySQL中每一个数据库都会在定义好（或者默认）的数据目录下存在一个以数据库名字命名的文件夹，用来存放该数据库中各种表数据文件。不同的MySQL存储引擎有各自不同的数据文件，存放位置也有区别。多数存储引擎的数据文件都存放在和MyISAM数据文件位置相同的目录下，但是每个数据文件的扩展名却各不一样。如MyISAM用“.MYD”作为扩展名，Innodb用“.ibd”，Archive用“.arc”，CSV用“.csv”，等等。
1、“.frm”文件 与表相关的元数据（meta）信息都存放在“.frm”文件中，包括表结构的定义信息等。不论是什么存储引擎，每一个表都会有一个以表名命名的“.frm”文件。所有的“.frm”文件都存放在所属数据库的文件夹下面。
2、“.MYD”文件“ .MYD”文件是MyISAM存储引擎专用，存放MyISAM表的数据。每一个MyISAM表都会有一个“.MYD”文件与之对应，同样存放于所属数据库的文件夹下，和“.frm”文件在一起。
3、“.MYI”文件 “.MYI”文件也是专属于MyISAM存储引擎的，主要存放MyISAM表的索引相关信息。对于MyISAM存储来说，可以被cache的内容主要就是来源于“.MYI”文件中。每一个MyISAM表对应一个“.MYI”文件，存放于位置和“.frm”以及“.MYD”一样。
4、“.ibd”文件和ibdata文件* 这两种文件都是存放Innodb数据的文件，之所以有两种文件来存放Innodb的数据（包括索引），是因为 Innodb 的数据存储方式能够通过配置来决定是使用共享表空间存放存储数据，还是独享表空间存放存储数据。独享表空间存储方式使用“.ibd”文件来存放数据，且每个表一个“.ibd”文件，文件存放在和MyISAM数据相同的位置。如果选用共享存储表空间来存放数据，则会使用 ibdata 文件来存放，所有表共同使用一个（或者多个，可自行配置）ibdata文件。

19、说下kafka都有哪些核心文件?

kafka

kafka消息都是根据topic这个逻辑上的概念进行分类的；生产者和消费者都是通过topic生产和消费消息的，分区partition是物理上的概念，每一个分区都对应一个log文件
假设 topic名称是 “topic-log” 那么日志文件：log对应 “topic-log-1"的文件夹 1 代表分区
日志文件和两个索引文件都是根据偏移量命名的（64位的长整数，固定20位长度，不足的前面补零如 000000000000000000001.log,000000000000000000001.index ）

1
Topic ==》 多个分区partition ==》 Log 日志 ==》 Log日志分段 ==》 每个分段对应 ：.log（日志文件） ,.index（偏移量索引文件） .timeindex（时间索引文件） 其他文件

引申：RocketMQ的核心文件有哪些？

commitLog 文件夹： commitLog文件存储消息（默认1GB，20位长度的文件名）；文件名称也是按照消息偏移量命名的；二分查找快速定位文件，消息物理偏移量（根据key从index获取） -文件名 = 消息在文件当中的位置
index 文件夹：index文件加速消息的检索；消息key与offset的对应关系； 40字节文件头；500万哈希槽；200万个index条目
consumequeue 文件夹：每个消息主题topic都包含多个消费队列，没一个队列都有一个消息文件；消息到达commitLog后异步发送到consumequeue文件；每一行固定20字节：8字节CommitLog物理偏移量+4字节消息长度+ 4字节 tag哈希码（哈希码就是为了保持固定长度）

20、说下redission分布式锁的实现原理?

首先讲一下redis 实现分布式锁的基本原理和主要步骤

主要是通过setnx命令或set 复合命令实现

1
2
3
4
# set 复合命令
# NX: IF NOT EXIST 的缩写，只有 KEY 不存在的前提下 才会设置值
# XX: IF EXIST 的缩写，只有在 KEY 存在的前提下 才会设置值
SET key value [expiration EX seconds|PX milliseconds] [NX|XX]

选取一个key作为redis锁的的key（建议命名要有意义），然后生成一个唯一的id（比如线程ID）作为该key的value
通过setnx命令设置k保存到redis，并同时设置超时时间（保证锁最终湖北释放），如果key已存在，会设置失败，表示被其他进程或线程占用，不存在则设置成功获取到锁（互斥性）
当处理完业务需要释放锁的时候，只能释放自己的锁，通过之前的value判断实现
以上步骤都可以通过lua脚本实现；从redis2.6 开始就支持lua脚本了，lua脚本将多个redis命令合在一起，可视为一个原子操作;释放锁的lua脚本示例如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
# 获取 KEYS[1] 对应的 Val
local cliVal = redis.call('get', KEYS[1])
# 判断 KEYS[1] 与 ARGV[1] 是否保持一致
if(cliVal == ARGV[1]) then 
  # 删除 KEYS[1]
  redis.call('del', KEYS[1]) 
  return 'OK' 
else
  return nil 
end

redission的实现原理

其实rdission主要就是通过lua脚本实现的，此外还有Watch dog 机制
可以从加锁机制、锁互斥机制、Watch dog 机制、可重入加锁机制、锁释放机制、等五个方面对 Redisson 实现分布式锁的底层原理进行分析
加锁原理：加锁其实是通过一段 lua 脚本实现的（hash数据结构）；field是客户端线程占有锁的唯一id，value是该线程加锁的次数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
## ARGV[2] 是加锁客户端的唯一标识， ARGV[1] 过期时间
if (redis.call('exists', KEYS[1]) == 0) then " +
   "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
   "redis.call('pexpire', KEYS[1], ARGV[1]); " +
   "return nil; " +
   "end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
    "return nil; " +
    "end; " +
"return redis.call('pttl', KEYS[1]);"

互斥原理：根据第一个参数线程id判断的；
*如果key已存在会检查对应的hash值，如果是id不是当前线程呢个的会走到另外一个分支，返回另外该key的存活时间（pttl命令）；如果是当前线程,则对计数+1（HINCRBY 命令）. 注意：在Java代码中，当锁正在被占用时，等待获取锁的进程并不是通过一个 while(true) 死循环去获取锁，而是利用了 Redis 的发布订阅机制,通过 await 方法阻塞等待锁的进程，有效的解决了无效的锁申请浪费资源的问题。
锁的续期机制：Redisson 提供了一个续期机制，只要客户端 1 一旦加锁成功，就会启动一个 Watch Dog；Watch Dog 机制其实就是一个后台定时任务线程，获取锁成功之后，会将持有锁的线程放入到一个 RedissonLock.EXPIRATION_RENEWAL_MAP里面，然后每隔 10 秒（internalLockLeaseTime / 3）检查一下
可重入加锁机制：关键在于上述的lua代码当中。在互斥性中也提到了,通过hincrby实现
释放锁机制 :

1 删除锁
2 广播释放锁的消息，通知阻塞等待的进程
3 取消watch dog看门狗，即将 RedissonLock.EXPIRATION_RENEWAL_MAP 里面的线程 id 删除，并且 cancel 掉 Netty 的那个定时任务线程

总结：

通过Redisson 实现分布式可重入锁，比原生的 SET mylock userId NX PX milliseconds + lua 实现的效果更好些，虽然基本原理都一样，但是它帮我们屏蔽了内部的执行细节。和 Zookeeper 相比较，Redisson 基于 Redis 性能更高，适合对性能要求高的场景 watch dog 机制比较好的解决了锁续期（超时时间不设置的情况下才生效，默认30秒）

21、说下lru算法的实现?

1 利用LinkedHashMap实现LRU算法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
/**
 * 利用LinkedHashMap实现的原理：
 *
 * get 能取到元素方法，会执行 ：afterNodeAccess(Node<K,V> e)  方法；move node to last； 前提 accessOrder 要设置为true
 * 默认是false； 所以LRUCache构造方法要调用
 * public LinkedHashMap(int initialCapacity,float loadFactor,boolean accessOrder) {
 *         super(initialCapacity, loadFactor);
 *         this.accessOrder = accessOrder;
 *     }
 * put方法添加新元素成功后调用 void afterNodeInsertion(boolean evict)： // possibly remove eldest:
 * afterNodeInsertion会 possibly remove eldest （从头部移除；前提：removeEldestEntry 返回true）
 * 实际是 在HashMap中实现 ： 最后一个参数evict为true会执行移除最久为使用的
 * public V put(K key, V value) {
 *         return putVal(hash(key), key, value, false, true);
 *     }
 * @param <K>
 * @param <V>
 */
class  LRUCache<K,V> extends LinkedHashMap<K,V> {

    private int capacity ;

    /**
     * LinkedHashMap 的构造方法中， accessOrder = false; 代表访问序
     * @param capacity
     * @param loadFactor
     */
    public LRUCache( int capacity,float loadFactor) {
        super(capacity, loadFactor,true);
        this.capacity = capacity;
    }

    public LRUCache(int capacity) {
        super(capacity, 0.75F,true);
        this.capacity = capacity;
    }

    /**
     *  容量不够时移除； LinkedHashMap是直接返回false，不移除
     * @return
     */
    @Override
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
        return super.size() > this.capacity;
    }
}

2 哈希表+双向链表实现

定义 size，capacity，cache（hashmap 哈希表实现，Map<K,Node> cache ;）

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

package com.example.demo;

import java.util.HashMap;
import java.util.Map;

public class LRUCache<K, V> {

    private int size;

    private int capacity;

    /**
     * 存储数据
     */
    private Map<K, Node> cache;

    /**
     * 标识作用，虚拟节点，辅助用，减少空判断,不存储具体的值
     */
    private Node header;

    /**
     * 标识作用，不存储具体的值
     */
    private Node tail;

    public LRUCache() {
        this(16);
    }

    public LRUCache(int capacity) {
        this.capacity = capacity;
        this.cache = new HashMap<>(capacity);
        this.header = new Node();
        this.tail = new Node();
        this.header.next = this.tail;
        this.tail.prev = this.header;

    }

    /**
     * 能访问到元素需要将原来的元素移动到末尾，标识最近访问
     *
     * @param k
     * @return
     */
    public V get(K k) {
        Node node = cache.get(k);
        if (node == null) {
            return null;
        }
        node.moveToLast(node);
        return node.v;
    }


    /**
     * <p>1 加入的元素放到末尾，放之前先判断是否存在同样的元素，存在则修改前后指针。</p>
     * <p> 2 不存在即就是新加入的元素，判断超过容量限制来决定删除头部元素 </p>
     *
     * @param k
     * @param v
     */
    public void put(K k, V v) {
        Node node = cache.get(k);
        if (node == null) {
            node = new Node(k,v);
            cache.put(k,node);
            size++;
            node.addToLast(node);
            if(size > capacity){
                //删除头部
                Node first = node.removeFirst();
                cache.remove(first.k);
                size -- ;

            }

        }else {
            node.v = v ;
            node.moveToLast(node);
        }


    }

    /**
     * 必须是非静态的类才能使用外部类的泛型和使用外部类的变量tail和header
     */
    class  Node {
        K k;
        V v;

        /**
         * 前驱节点
         */
        Node prev;

        Node next;

        public Node() {
        }

        public Node(K k, V v) {
            this.k = k;
            this.v = v;
        }

        private void addToLast(Node node) {
            node.prev = tail.prev;
            node.next = tail;

            node.prev.next = node;
            tail.prev = node;

        }

        private void addToFirst(Node node) {
            node.next = header.next;
            node.prev = header;

            node.next.prev = node;
            header.next = node;


        }

        private void moveToLast(Node node) {
             remove(node);
             addToLast(node);

        }

        private void moveToFirst(Node node) {
            remove(node);
            addToFirst(node);

        }

        private Node remove(Node node) {
            if (node == null) return null;
            if (node.prev == null || node.next == null) {
                return node;
            }
            node.prev.next = node.next;
            node.next.prev = node.prev;
            node.next = null;
            node.prev = null;
            return node;
        }

        private Node removeFirst() {
            Node h = header.next;
            return remove(h);
        }

        private Node removeLast() {
            Node t = tail.prev;
           return  remove(t);
        }
    }

    public static void main(String[] args) {
        LRUCache<Integer, Integer> lruCache = new LRUCache<>(10);
        for (int i = 0; i < 20; i++) {
            // 1 会一直在
            lruCache.put(1,1);
            lruCache.put(i,i);

        }
        for (int i = 18; i > 12; i--) {
            lruCache.put(i,i);

        }
        for (int i = 0 ; i< 20; i++){
            System.out.print(lruCache.get(i) + " ");
        }
    }


}