linux/linux常见命令/linux命令.md

@@ -0,0 +1,96 @@
+# 什么是source命令？
+
+运行shell时，会同时存在两种变量：
+
+- **1) 局部变量** 局部变量在脚本或命令中定义，仅在当前shell实例中有效，其他shell启动的程序不能访问局部变量。
+- **2) 环境变量** 所有的程序，包括shell启动的程序，都能访问环境变量，有些程序需要环境变量来保证其正常运行。必要的时候shell脚本也可以定义环境变量。
+
+在 Linux 中，source 命令用于执行指定的 shell 脚本文件，并将脚本文件中的所有命令都当作当前 shell 环境的命令来运行。简单地说，source 命令可以让你在当前的 shell 环境中执行一个脚本文件，并将其中定义的环境变量和函数等内容加载到当前的 shell 环境中。
+
+具体而言，source 命令可以实现如下功能：
+
+1. 在当前 shell 环境中载入指定脚本文件的内容，包括环境变量、函数等。
+2. 不创建新的子 shell 进程，而是在当前 shell 环境中执行脚本文件中的命令，从而实现在当前 shell 中设置环境变量或执行函数等操作。
+
+使用 source 命令的常见场景包括：
+
+1. 执行 shell 脚本文件，以实现设置环境变量、执行别名或函数等功能。
+2. 加载特定的 shell 配置文件，例如 ~/.bashrc 或 /etc/profile，以初始化 shell 环境。
+
+总之，source 命令可以帮助你在当前的 shell 环境中执行指定的脚本文件，并将其中定义的环境变量、函数等内容加载到当前的 shell 操作中，非常有用。
+
+## 简单例子
+
+```sh
+export var=999
+```
+
+如果`./my.sh`，运行后，执行`echo $var`是空的。
+
+如果`source ./my.sh`运行后，执行`echo $var`会显示999。
+
+因为调用`./my.sh`来执行`shell`是在一个子`shell`里运行的，所以执行后，里面定义的变量并没有反应到本`shell`里，但是 `source`不同，是在本`shell`中执行的，所以本shell能够看到其定义的变量。
+
+# export命令
+
+在 Linux 中，export 命令用于向当前 shell 环境中添加新的环境变量，使其能被当前 shell 环境及其子进程所访问和使用。简单地说，export 命令可以将一个变量设置为环境变量，从而可以在当前 shell 会话中以及当前 shell 的子进程中使用该变量。
+
+具体而言，export 命令可以实现如下功能：
+
+1. 将指定的变量设置为当前 shell 环境的环境变量。
+2. 将环境变量持久化保存到系统文件中，以便在下次启动时仍然可用。
+3. 环境变量的作用范围不仅限于当前 shell 环境，还包括当前 shell 环境的所有子进程，因此可以实现父子进程之间的环境变量共享。
+
+使用 export 命令可以将一个普通变量设置为环境变量，例如：
+
+```shell
+MY_VAR="hello"
+export MY_VAR
+```
+
+这样就将 MY_VAR 变量设置为当前 shell 环境的环境变量，以便在当前 shell 会话和当前 shell 的子进程中使用。
+
+同时，你也可以直接在一行中同时定义和导出一个环境变量，例如：
+
+```shell
+export MY_VAR="hello"
+```
+
+这样也可以将 MY_VAR 设置为环境变量，并将其导出到当前 shell 环境中。
+
+总之，export 命令可以将指定变量设置为环境变量，并将其在当前 shell 环境及其子进程中进行共享和使用，非常有用。
+
+# #!/bin/bash 作用
+
+`#!/bin/bash` 是一种脚本文件的头部声明，也称为 Shebang 或 Hashbang。在 Linux 和 Unix 系统中，这行命令用于指定解释器的路径，让操作系统知道该如何执行该脚本文件。
+
+具体而言，当一个脚本文件被执行时，操作系统会读取脚本文件的第一行，如果该行指定了解释器的路径，操作系统就会使用该解释器来执行该脚本文件；否则，操作系统会使用默认的解释器来执行该脚本文件（在大多数情况下是 /bin/sh）。
+
+因此，在编写 Shell 脚本时，将 `#!/bin/bash` 添加到脚本文件的开头可以确保脚本文件将使用 Bash 解释器来执行，而不是其他解释器。同时，还可以通过修改 Shebang 行的内容来指定使用其他解释器来执行脚本文件。
+
+需要注意的是，`#!/bin/bash` 命令在脚本文件的开头必须是第一行，并且不能有任何空格或其他字符紧随其后，否则该命令将不会被正确识别，从而导致脚本无法执行。
+
+总之，`#!/bin/bash` 命令的作用是指定解释器的路径，让操作系统知道该如何执行该脚本文件。在编写 Shell 脚本时，通常需要添加该命令到脚本文件的开头。
+
+# -e作用
+
+`-e` 是 Bash 脚本中的一个选项，用于指定在脚本执行过程中如果出现错误就立即退出并返回非零（非正常）状态码。
+
+具体而言，当在 Bash 脚本中添加了 `set -e` 命令后，表示开启了该选项（也可以使用 `set -o errexit`），此时如果在脚本的任何一行命令或管道执行失败，Bash 就会立即退出并返回非零状态码。这个特性可以确保脚本在执行过程中出现错误后可以立即停止执行，避免产生更多的错误。
+
+举个例子，假设有一个名为 `test.sh` 的 Shell 脚本文件，它的内容如下：
+
+```shell
+#!/bin/bash
+set -e
+
+echo "start"
+command_not_exist
+echo "end"
+```
+
+其中，`set -e` 指定了开启 `-e` 选项，`command_not_exist` 是一条不存在的命令。
+
+如果在执行该脚本时，命令 `command_not_exist` 执行失败，那么该脚本就会立即退出，并且不会输出 "end"。同时，脚本会返回一个非零状态码（通常是 1），表示脚本执行过程中发生错误。
+
+总之，`-e` 选项可以确保在 Bash 脚本执行过程中出现错误就立即退出并返回非零状态码，从而更好地确保脚本执行的可靠性。

linux/linux服务器运维/vmware exsi/11、交换机安装.md

@@ -0,0 +1,49 @@
+## 安装VPS交换机
+
+1. 在vsphere中选择部署OVF模板
+
+![image-20230525143835817](assets/image-20230525143835817.png)
+
+2. 选择三个模板文件
+
+![image-20230525143521677](assets/image-20230525143521677.png)
+
+3. 按照具体选择虚拟机位置
+
+![image-20230525143554957](assets/image-20230525143554957.png)
+
+4. 选择一个exsi机器，这里选择默认的即可
+
+![image-20230525143606116](assets/image-20230525143606116.png)
+
+5. 选择安装磁盘
+
+![image-20230525143655534](assets/image-20230525143655534.png)
+
+6. 网络选择自定义的server网络
+
+![image-20230525143806979](assets/image-20230525143806979.png)
+
+7. 打开虚拟机配置IP，网关，子网掩码
+
+![image-20230525161825139](assets/image-20230525161825139.png)
+
+8. 打开上述配置的IP，默认用户名密码是nsroot
+
+![image-20230525161811186](assets/image-20230525161811186.png)
+
+9. 安装配置完毕
+
+![image-20230526141545083](assets/image-20230526141545083.png)
+
+## 升级
+
+1. 登录citrix站点，下载最新的安装包
+2. FTP 客户端上传安装包到此设备
+3. 将升级所需要的安装包从计算机复制到设备上的 /var/nsinstall 目录。
+4. 使用安全的SSH客户端打开与设备的 SSH 连接。
+5. 在命令提示窗口中，键入： `shell`
+6. 进入安装目录，在命令提示符下键入：`cd /var/nsinstall`
+7. 需要解压安装软件，输入：tar –xvzf build_X_XX.tgz，其中 Build_x_xx.tgz 是要升级到的内部版本的名称。
+8. 开始安装，在命令提示符下键入：`./installns`
+9. 安装完成后，重新启动服务器

linux/linux服务器配置/配置基础环境.md

@@ -0,0 +1,37 @@
+```sh
+#!/bin/bash
+set -e
+
+# 关闭防火墙
+systemctl stop firewalld && systemctl disable firewalld
+# 配置yum源
+yum install wget -y
+mv /etc/yum.repos.d/CentOS-Base.repo /etc/yum.repos.d/CentOS-Base.repo.bak
+wget -O /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-7.repo
+yum clean all 
+yum makecache
+# 配置时间同步
+yum install ntp -y
+systemctl enable ntpd && systemctl start ntpd
+timedatectl set-timezone Asia/Shanghai
+
+# 安装必要软件
+yum install git vim -y
+
+# 安装java
+yum -y remove *java*
+yum install -y java-1.8.0-openjdk*
+
+# 安装maven
+yum install maven
+sed -i '/<mirrors>/,/<\/mirrors>/c\
+  <mirrors>\
+    <mirror>\
+      <id>alimaven</id>\
+      <url>https://maven.aliyun.com/repository/public</url>\
+      <mirrorOf>central</mirrorOf>\
+    </mirror>\
+  </mirrors>
+' /etc/maven/settings.xml
+```
+

后端/Java/JAVA高阶/JUC编程/线程池源码.md

@@ -0,0 +1,379 @@
+## 线程池
+
+池化技术的作用：把一些能够复用的东西（比如说连接、线程）放到初始化好的池中，便于资源统一管理。
+
+**execute()执行流程图**
+
+![image-20230515102824954](assets/image-20230515102824954.png)
+
+## execute源码解析
+
+### 基础的变量
+
+```java
+// 使用原子操作类AtomicInteger的ctl变量，前3位记录线程池的状态，后29位记录线程数
+private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
+// Integer的范围为[-2^31,2^31 -1], Integer.SIZE-3 =32-3= 29，用来辅助左移位运算
+private static final int COUNT_BITS = Integer.SIZE - 3;
+// 高三位用来存储线程池运行状态，其余位数表示线程池的容量
+private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
+
+// 线程池状态以常量值被存储在高三位中
+// RUNNING状态其实就是全1
+private static final int RUNNING    = -1 << COUNT_BITS; // 线程池接受新任务并会处理阻塞队列中的任务
+private static final int SHUTDOWN   =  0 << COUNT_BITS; // 线程池不接受新任务，但会处理阻塞队列中的任务
+private static final int STOP       =  1 << COUNT_BITS; // 线程池不接受新的任务且不会处理阻塞队列中的任务，并且会中断正在执行的任务
+private static final int TIDYING    =  2 << COUNT_BITS; // 所有任务都执行完成，且工作线程数为0，将调用terminated方法
+private static final int TERMINATED =  3 << COUNT_BITS; // 最终状态，为执行terminated()方法后的状态
+
+// ctl变量的封箱拆箱相关的方法
+private static int runStateOf(int c)     { return c & ~CAPACITY; } // 获取线程池运行状态
+private static int workerCountOf(int c)  { return c & CAPACITY; } // 获取线程池运行线程数
+private static int ctlOf(int rs, int wc) { return rs | wc; } // 获取ctl对象
+private static boolean isRunning(int c) { return c < SHUTDOWN; } // 是否在运行
+```
+
+### 核心方法
+
+```java
+public void execute(Runnable command) {
+    if (command == null) // 任务为空，抛出NPE
+        throw new NullPointerException();
+        
+    int c = ctl.get(); // 获取当前工作线程数和线程池运行状态（共32位，前3位为运行状态，后29位为运行线程数）
+    if (workerCountOf(c) < corePoolSize) { // 如果当前工作线程数小于核心线程数
+        if (addWorker(command, true)) // 在addWorker中创建工作线程并执行任务
+            return;
+        c = ctl.get();
+    }
+    
+    // 核心线程数已满（工作线程数>核心线程数）才会走下面的逻辑
+    if (isRunning(c) && workQueue.offer(command)) { // 如果当前线程池状态为RUNNING，并且任务成功添加到阻塞队列
+        int recheck = ctl.get(); // 双重检查，因为从上次检查到进入此方法，线程池可能已成为SHUTDOWN状态
+        if (! isRunning(recheck) && remove(command)) // 如果当前线程池状态不是RUNNING则从队列删除任务
+            reject(command); // 执行拒绝策略
+        else if (workerCountOf(recheck) == 0) // 当线程池中的workerCount为0时，此时workQueue中还有待执行的任务，则新增一个addWorker，消费workqueue中的任务
+            addWorker(null, false);
+    }
+    // 阻塞队列已满才会走下面的逻辑
+    else if (!addWorker(command, false)) // 尝试增加工作线程执行command
+        // 如果当前线程池为SHUTDOWN状态或者线程池已饱和
+        reject(command); // 执行拒绝策略
+}
+```
+
+### 添加worker
+
+```java
+private boolean addWorker(Runnable firstTask, boolean core) {
+    retry: // 循环退出标志位
+    for (;;) { // 无限循环
+        int c = ctl.get();
+        int rs = runStateOf(c); // 线程池状态
+
+        // Check if queue empty only if necessary.
+        if (rs >= SHUTDOWN && 
+            ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()) // 换成更直观的条件语句
+            // (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())
+           )
+           // 返回false的条件就可以分解为：
+           //（1）线程池状态为STOP，TIDYING，TERMINATED
+           //（2）线程池状态为SHUTDOWN，且要执行的任务不为空
+           //（3）线程池状态为SHUTDOWN，且任务队列为空
+            return false;
+
+        // cas自旋增加线程个数
+        for (;;) {
+            int wc = workerCountOf(c); // 当前工作线程数
+            if (wc >= CAPACITY ||
+                wc >= (core ? corePoolSize : maximumPoolSize)) // 工作线程数>=线程池容量 || 工作线程数>=(核心线程数||最大线程数)
+                return false;
+            if (compareAndIncrementWorkerCount(c)) // 执行cas操作，添加线程个数
+                break retry; // 添加成功，退出外层循环
+            // 通过cas添加失败
+            c = ctl.get();  
+            // 线程池状态是否变化，变化则跳到外层循环重试重新获取线程池状态，否者内层循环重新cas
+            if (runStateOf(c) != rs)
+                continue retry;
+            // else CAS failed due to workerCount change; retry inner loop
+        }
+    }
+    // 简单总结上面的CAS过程：
+    //（1）内层循环作用是使用cas增加线程个数，如果线程个数超限则返回false，否者进行cas
+    //（2）cas成功则退出双循环，否者cas失败了，要看当前线程池的状态是否变化了
+    //（3）如果变了，则重新进入外层循环重新获取线程池状态，否者重新进入内层循环继续进行cas
+
+    // 走到这里说明cas成功，线程数+1，但并未被执行
+    boolean workerStarted = false; // 工作线程调用start()方法标志
+    boolean workerAdded = false; // 工作线程被添加标志
+    Worker w = null;
+    try {
+        w = new Worker(firstTask); // 创建工作线程实例
+        final Thread t = w.thread; // 获取工作线程持有的线程实例
+        if (t != null) {
+            final ReentrantLock mainLock = this.mainLock; // 使用全局可重入锁
+            mainLock.lock(); // 加锁，控制并发
+            try {
+                // Recheck while holding lock.
+                // Back out on ThreadFactory failure or if
+                // shut down before lock acquired.
+                int rs = runStateOf(ctl.get()); // 获取当前线程池状态
+
+                // 线程池状态为RUNNING或者（线程池状态为SHUTDOWN并且没有新任务时）
+                if (rs < SHUTDOWN ||
+                    (rs == SHUTDOWN && firstTask == null)) {
+                    if (t.isAlive()) // 检查线程是否处于活跃状态
+                        throw new IllegalThreadStateException();
+                    workers.add(w); // 线程加入到存放工作线程的HashSet容器，workers全局唯一并被mainLock持有
+                    int s = workers.size();
+                    if (s > largestPoolSize)
+                        largestPoolSize = s;
+                    workerAdded = true;
+                }
+            } finally {
+                mainLock.unlock(); // finally块中释放锁
+            }
+            if (workerAdded) { // 线程添加成功
+                t.start(); // 调用线程的start()方法
+                workerStarted = true;
+            }
+        }
+    } finally {
+        if (! workerStarted) // 如果线程启动失败，则执行addWorkerFailed方法
+            addWorkerFailed(w);
+    }
+    return workerStarted;
+}
+```
+
+### 添加worker失败，工作队列里面的任务
+
+```java
+private void addWorkerFailed(Worker w) {
+    final ReentrantLock mainLock = this.mainLock;
+    mainLock.lock();
+    try {
+        if (w != null)
+            workers.remove(w); // 线程启动失败时，需将前面添加的线程删除
+        decrementWorkerCount(); // ctl变量中的工作线程数-1
+        tryTerminate(); // 尝试将线程池转变成TERMINATE状态
+    } finally {
+        mainLock.unlock();
+    }
+}
+```
+
+### 尝试去执行
+
+```java
+final void tryTerminate() {
+    for (;;) {
+        int c = ctl.get();
+        // 以下情况不会进入TERMINATED状态：
+        //（1）当前线程池为RUNNING状态
+        //（2）在TIDYING及以上状态
+        //（3）SHUTDOWN状态并且工作队列不为空
+        //（4）当前活跃线程数不等于0
+        if (isRunning(c) ||
+            runStateAtLeast(c, TIDYING) ||
+            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
+            return;
+        if (workerCountOf(c) != 0) { // 工作线程数!=0
+            interruptIdleWorkers(ONLY_ONE); // 中断一个正在等待任务的线程
+            return;
+        }
+
+        final ReentrantLock mainLock = this.mainLock;
+        mainLock.lock();
+        try {
+            // 通过CAS自旋判断直到当前线程池运行状态为TIDYING并且活跃线程数为0
+            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
+                try {
+                    terminated(); // 调用线程terminated()
+                } finally {
+                    ctl.set(ctlOf(TERMINATED, 0)); // 设置线程池状态为TERMINATED，工作线程数为0
+                    termination.signalAll(); // 通过调用Condition接口的signalAll()唤醒所有等待的线程
+                }
+                return;
+            }
+        } finally {
+            mainLock.unlock();
+        }
+        // else retry on failed CAS
+    }
+}
+```
+
+## Worker源码解读
+
+`Worker`是`ThreadPoolExecutor`类的内部类，此处只讲最重要的构造函数和run方法
+
+```Java
+private final class Worker extends AbstractQueuedSynchronizer implements Runnable
+{
+    // 该worker正在运行的线程
+    final Thread thread;
+    
+    // 将要运行的初始任务
+    Runnable firstTask;
+    
+    // 每个线程的任务计数器
+    volatile long completedTasks;
+
+    // 构造方法   
+    Worker(Runnable firstTask) {
+        setState(-1); // 调用runWorker()前禁止中断
+        this.firstTask = firstTask;
+        this.thread = getThreadFactory().newThread(this); // 通过ThreadFactory创建一个线程
+    }
+
+    // 实现了Runnable接口的run方法
+    public void run() {
+        runWorker(this);
+    }
+    
+    ... // 此处省略了其他方法
+}
+```
+
+#### runWorker方法
+
+```Java
+final void runWorker(Worker w) {
+    Thread wt = Thread.currentThread();
+    Runnable task = w.firstTask; // 获取工作线程中用来执行任务的线程实例
+    w.firstTask = null;
+    w.unlock(); // status设置为0，允许中断
+    boolean completedAbruptly = true; // 线程意外终止标志
+    try {
+        // 如果当前任务不为空，则直接执行；否则调用getTask()从任务队列中取出一个任务执行
+        while (task != null || (task = getTask()) != null) {
+            w.lock(); // 加锁，保证下方临界区代码的线程安全
+            // 如果状态值大于等于STOP且当前线程还没有被中断，则主动中断线程
+            if ((runStateAtLeast(ctl.get(), STOP) ||
+                 (Thread.interrupted() &&
+                  runStateAtLeast(ctl.get(), STOP))) &&
+                !wt.isInterrupted())
+                wt.interrupt(); // 中断当前线程
+            try {
+                beforeExecute(wt, task); // 任务执行前的回调，空实现，可以在子类中自定义
+                Throwable thrown = null;
+                try {
+                    task.run(); // 执行线程的run方法
+                } catch (RuntimeException x) {
+                    thrown = x; throw x;
+                } catch (Error x) {
+                    thrown = x; throw x;
+                } catch (Throwable x) {
+                    thrown = x; throw new Error(x);
+                } finally {
+                    afterExecute(task, thrown); // 任务执行后的回调，空实现，可以在子类中自定义
+                }
+            } finally {
+                task = null; // 将循环变量task设置为null，表示已处理完成
+                w.completedTasks++; // 当前已完成的任务数+1
+                w.unlock();
+            }
+        }
+        completedAbruptly = false;
+    } finally {
+        processWorkerExit(w, completedAbruptly);
+    }
+}
+```
+
+#### 从任务队列中取出一个任务
+
+```Java
+private Runnable getTask() {
+    boolean timedOut = false; // 通过timeOut变量表示线程是否空闲时间超时了
+    // 无限循环
+    for (;;) {
+        int c = ctl.get(); // 线程池信息
+        int rs = runStateOf(c); // 线程池当前状态
+
+        // 如果线程池状态>=SHUTDOWN并且工作队列为空 或 线程池状态>=STOP，则返回null，让当前worker被销毁
+        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
+            decrementWorkerCount(); // 工作线程数-1
+            return null;
+        }
+
+        int wc = workerCountOf(c); // 获取当前线程池的工作线程数
+
+        // 当前线程是否允许超时销毁的标志
+        // 允许超时销毁：当线程池允许核心线程超时 或 工作线程数>核心线程数
+        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
+
+        // 如果(当前线程数大于最大线程数 或 (允许超时销毁 且 当前发生了空闲时间超时))
+        // 且(当前线程数大于1 或 阻塞队列为空)
+        // 则减少worker计数并返回null
+        if ((wc > maximumPoolSize || (timed && timedOut))
+            && (wc > 1 || workQueue.isEmpty())) {
+            if (compareAndDecrementWorkerCount(c))
+                return null;
+            continue;
+        }
+
+        try {
+            // 根据线程是否允许超时判断用poll还是take（会阻塞）方法从任务队列头部取出一个任务
+            Runnable r = timed ?
+                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
+                workQueue.take();
+            if (r != null)
+                return r; // 返回从队列中取出的任务
+            timedOut = true;
+        } catch (InterruptedException retry) {
+            timedOut = false;
+        }
+    }
+}
+```
+
+总结一下哪些情况getTask()会返回null：
+
+> 1. 线程池状态为SHUTDOWN且任务队列为空
+> 2. 线程池状态为STOP、TIDYING、TERMINATED
+> 3. 线程池线程数大于最大线程数
+> 4. 线程可以被超时回收的情况下等待新任务超时
+
+#### 工作线程退出
+
+```Java
+private void processWorkerExit(Worker w, boolean completedAbruptly) {
+    // 如果completedAbruptly为true则表示任务执行过程中抛出了未处理的异常
+    // 所以还没有正确地减少worker计数，这里需要减少一次worker计数
+    if (completedAbruptly) 
+        decrementWorkerCount();
+
+    final ReentrantLock mainLock = this.mainLock;
+    mainLock.lock();
+    try {
+        // 把将被销毁的线程已完成的任务数累加到线程池的完成任务总数上
+        completedTaskCount += w.completedTasks;
+        workers.remove(w); // 从工作线程集合中移除该工作线程
+    } finally {
+        mainLock.unlock();
+    }
+
+    // 尝试结束线程池
+    tryTerminate();
+
+    int c = ctl.get();
+    // 如果是RUNNING 或 SHUTDOWN状态
+    if (runStateLessThan(c, STOP)) {
+        // worker是正常执行完
+        if (!completedAbruptly) {
+            // 如果允许核心线程超时则最小线程数是0，否则最小线程数等于核心线程数
+            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
+            // 如果阻塞队列非空，则至少要有一个线程继续执行剩下的任务
+            if (min == 0 && ! workQueue.isEmpty())
+                min = 1;
+            // 如果当前线程数已经满足最小线程数要求，则不需要再创建替代线程
+            if (workerCountOf(c) >= min)
+                return; // replacement not needed
+        }
+        // 重新创建一个worker来代替被销毁的线程
+        addWorker(null, false);
+    }
+}
+```

后端/项目/开源之夏/申请书.md

@@ -0,0 +1,429 @@
+> [TOC]
+
+# 1、项目背景
+
+项目名称：增加跨工作流的参数传递功能
+
+项目主导师: 鲍亮
+
+申请人：孙浩博
+
+申请日期：2023/6/4
+
+邮箱：sunhaobo@stu.xidian.edu.cn
+
+## 1.1、项目基本背景
+
+为开源项目dolphinscheduler设计两个任务，并且为上述两个任务编写详细的设计文档以及使用文档
+
+1. 工作流可以选择自己的输出参数，作为工作流的输出参数，输出给下游任务使用。
+
+场景1： shellA -> subprocessB -> shellC
+
+shellA 查出所有学生信息 （a,b,c） 并将 users 输出给下游任务subprocessB
+
+subprocessB 是一个子工作流， 负责计算所有学生个数 userCount，并将userCount 作为工作流的输出传递给下游
+
+shellC 负责将 userCount 输出到控制台
+
+2. 依赖任务也可以选择将被依赖任务的输出参数继承过来使用。增加依赖参数继承功能
+
+需要在依赖任务上增加一个字段：是否继承被依赖任务的参数
+
+场景2：
+
+工作流A ： taska1 -> taska2, taska2 输出了参数 count=10
+
+工作流B : taskb1 -> dependentb2 -> taskb3
+
+dependentb2 依赖了taska2 就可以将count继承过来作为dependentb2的输出参数，给taskb3使用
+
+
+## 1.2、开源项目仓库
+
+- https://github.com/apache/dolphinscheduler
+
+# 2、方法可行性
+
+## 2.1、vue相关
+
+本人有vue开发相关经验，曾从事实验室项目（煤矿大数据平台--大屏模块）的开发，熟练运用前端知识，完成项目开发。主要包括，使用flex和grid完成自适应布局；基于websocket的实时数据展示；基于 echarts封装常见组件如柱状图、饼状图、折线图；对接三方监控，了解chrome播放策略等。
+
+## 2.2、java开发经验
+
+丰富的linux使用经验
+在中电六所实习的经验，从事软件项目开发
+实验室项目开发经验，从事煤炭大数据后端模块的开发
+个人博客编写习惯，编写个人博客并在github开源，[博客地址](https://www.seamew.top)
+
+# 3、项目实现细节--（主要分析start-process-instance接口）
+
+dolphinscheduler项目主要实现为master和worker节点的交互，以下内容分析执行start-process-instance接口的实现细节
+
+![image-20230604104941343](assets/image-20230604104941343.png)
+
+## 3.1、master执行流程分析
+
+首先调用execService的execProcessInstance方法
+
+```java
+        Map<String, Object> result = execService.execProcessInstance(loginUser, projectCode, processDefinitionCode,
+                scheduleTime, execType, failureStrategy,
+                startNodeList, taskDependType, warningType, warningGroupId, runMode, processInstancePriority,
+                workerGroup, tenantCode, environmentCode, timeout, startParamMap, expectedParallelismNumber, dryRun,
+                testFlag,
+                complementDependentMode, version, allLevelDependent);
+        return returnDataList(result);
+```
+
+execProcessInstance方法实现细节
+
+```java
+        /**
+         * 主要调用creat command方法将command存储到DB中
+         */
+        int create =
+                this.createCommand(triggerCode, commandType, processDefinition.getCode(), taskDependType,
+                        failureStrategy,
+                        startNodeList,
+                        cronTime, warningType, loginUser.getId(), warningGroupId, runMode, processInstancePriority,
+                        workerGroup, tenantCode,
+                        environmentCode, startParams, expectedParallelismNumber, dryRun, testFlag,
+                        complementDependentMode, allLevelDependent);
+
+
+    private int createCommand() {
+        // 忽略大部分代码
+        } else {
+            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
+        	// 实际上执行createCommand方法
+            int count = commandService.createCommand(command);
+            if (count > 0) {
+                triggerRelationService.saveTriggerToDb(ApiTriggerType.COMMAND, triggerCode, command.getId());
+            }
+            return count;
+        }
+    }
+
+    public int createCommand(Command command) {
+        int result = 0;
+        if (command == null) {
+            return result;
+        }
+        // add command timezone
+        Schedule schedule = scheduleMapper.queryByProcessDefinitionCode(command.getProcessDefinitionCode());
+        if (schedule != null) {
+            Map<String, String> commandParams =
+                    StringUtils.isNotBlank(command.getCommandParam()) ? JSONUtils.toMap(command.getCommandParam())
+                            : new HashMap<>();
+            commandParams.put(Constants.SCHEDULE_TIMEZONE, schedule.getTimezoneId());
+            command.setCommandParam(JSONUtils.toJsonString(commandParams));
+        }
+        command.setId(null);
+        // 这一步将command存储到DB中
+        result = commandMapper.insert(command);
+        return result;
+    }
+
+```
+
+## 3.2、master分配任务细节
+
+主要是MasterSchedulerBootstrap线程线性扫描DB解析command，然后使用分片算法分发给worker
+
+run方法解析
+
+```java
+    @Override
+    public void run() {
+        // 省略大部分代码
+        // 这一步直接扫描数据库获取command，注意这里同时也会删除数据库习惯记录
+        List<Command> commands = findCommands();
+        commands.parallelStream()
+            .forEach(command -> {
+                try {
+                    WorkflowExecuteRunnable workflowExecuteRunnable =
+                        workflowExecuteRunnableFactory.createWorkflowExecuteRunnable(command);
+                    ProcessInstance processInstance = workflowExecuteRunnable.getProcessInstance();
+                    if (processInstanceExecCacheManager.contains(processInstance.getId())) {
+                        log.error(
+                            "The workflow instance is already been cached, this case shouldn't be happened");
+                    }
+                    // 这一步是启动一个监听任务，监听该实例执行情况
+                    processInstanceExecCacheManager.cache(processInstance.getId(), workflowExecuteRunnable);
+                    // 通过事件的方式去传播任务
+                    workflowEventQueue.addEvent(
+                        new WorkflowEvent(WorkflowEventType.START_WORKFLOW, processInstance.getId()));
+    }
+```
+
+WorkflowEventLooper类用来处理MasterSchedulerBootstrap方法添加的事件
+
+run方法解析
+
+```java
+    public void run() {
+        // 省略大部分代码
+        WorkflowEvent workflowEvent;
+        while (RUNNING_FLAG.get()) {
+            try {
+                // 获取workflowEventQueue队列的事件
+                workflowEvent = workflowEventQueue.poolEvent();
+            try (
+                WorkflowEventHandler workflowEventHandler =
+                        workflowEventHandlerMap.get(workflowEvent.getWorkflowEventType());
+                // 这里通过策略模式处理事件
+                workflowEventHandler.handleWorkflowEvent(workflowEvent);
+    }
+                
+    @Override
+    public void handleWorkflowEvent(final WorkflowEvent workflowEvent) throws WorkflowEventHandleError {
+	    // 省略大部分代码
+        // 核心调用call方法去处理任务
+        CompletableFuture.supplyAsync(workflowExecuteRunnable::call, workflowExecuteThreadPool)
+    }
+                
+    @Override
+    public WorkflowSubmitStatus call() {
+        // 省略大部分代码
+        try {
+            if (workflowRunnableStatus == WorkflowRunnableStatus.CREATED) {
+                // 新建甘特图
+                buildFlowDag();
+                workflowRunnableStatus = WorkflowRunnableStatus.INITIALIZE_DAG;
+                log.info("workflowStatue changed to :{}", workflowRunnableStatus);
+            }
+            if (workflowRunnableStatus == WorkflowRunnableStatus.INITIALIZE_DAG) {
+                // 初始化任务队列
+                initTaskQueue();
+                workflowRunnableStatus = WorkflowRunnableStatus.INITIALIZE_QUEUE;
+                log.info("workflowStatue changed to :{}", workflowRunnableStatus);
+            }
+            if (workflowRunnableStatus == WorkflowRunnableStatus.INITIALIZE_QUEUE) {
+                // 提交任务
+                submitPostNode(null);
+                workflowRunnableStatus = WorkflowRunnableStatus.STARTED;
+                log.info("workflowStatue changed to :{}", workflowRunnableStatus);
+            }
+            return WorkflowSubmitStatus.SUCCESS;
+        }
+    }
+```
+
+现在为止我们分析到了提交任务阶段，submitPostNode方法提交任务之后实际会执行submitStandByTask方法提交
+
+```java
+    private void submitPostNode(String parentNodeCode) throws StateEventHandleException {
+        // 省略大部分代码
+        submitStandByTask();
+        updateProcessInstanceState();
+    }
+
+    public void submitStandByTask() throws StateEventHandleException {
+        // 省略大部分代码
+        TaskInstance task;
+        // 循环peek处理任务队列
+        while ((task = readyToSubmitTaskQueue.peek()) != null) {
+            DependResult dependResult = getDependResultForTask(task);
+            if (DependResult.SUCCESS == dependResult) {
+                log.info("The dependResult of task {} is success, so ready to submit to execute", task.getName());
+                // 实际处理任务
+                if (!executeTask(task)) {
+        }
+    }
+    private boolean executeTask(TaskInstance taskInstance) {
+        // 省略大部分代码
+        // 核心代码，分发任务
+        taskExecuteRunnable.dispatch();
+    }
+```
+
+任务分发之后，GlobalTaskDispatchWaitingQueueLooper类会处理分发的任务
+
+```java
+   @Override
+    public void run() {
+        // 省略大部分代码
+        DefaultTaskExecuteRunnable defaultTaskExecuteRunnable;
+        while (RUNNING_FLAG.get()) {
+            try {
+                // 获取任务
+                defaultTaskExecuteRunnable = globalTaskDispatchWaitingQueue.takeNeedToDispatchTaskExecuteRunnable();
+            }
+            try {
+                final TaskDispatcher taskDispatcher = taskDispatchFactory
+                        .getTaskDispatcher(defaultTaskExecuteRunnable.getTaskInstance().getTaskType());
+                // 分发任务
+                taskDispatcher.dispatchTask(defaultTaskExecuteRunnable);
+    }
+                @Override
+    public void dispatchTask(TaskExecuteRunnable taskExecuteRunnable) throws TaskDispatchException {
+        // 省略大部分代码
+        // 实际分发任务
+        doDispatch(taskExecuteRunnable);
+        // 添加分发任务监听事件
+        addDispatchEvent(taskExecuteRunnable);
+    }
+            
+    protected void doDispatch(TaskExecuteRunnable taskExecuteRunnable) throws TaskDispatchException {
+        // 省略大部分代码
+        // 核心代码，通过RPC方式分发任务给task处理线程
+        Message message = masterRpcClient.sendSyncCommand(Host.of(taskExecutionContext.getHost()),
+                                                          taskDispatchRequest.convert2Command());
+    }
+```
+
+## 3.3、worker线程处理task
+
+NettyServerHandler类用来接收master传递的RPC消息
+
+```java
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        // 接受消息
+        processReceived(ctx.channel(), (Message) msg);
+    }
+
+    private void processReceived(final Channel channel, final Message msg) {
+        // 省略大部分代码
+        final Pair<NettyRequestProcessor, ExecutorService> pair = processors.get(messageType);
+        if (pair != null) {
+            Runnable r = () -> {
+                try {
+                    // 通过策略模式分发消息
+                    pair.getLeft().process(channel, msg);
+    }
+```
+
+WorkerTaskDispatchProcessor类来分发消息
+
+```java
+    @Counted(value = "ds.task.execution.count", description = "task execute total count")
+    @Timed(value = "ds.task.execution.duration", percentiles = {0.5, 0.75, 0.95, 0.99}, histogram = true)
+    @Override
+    public void process(Channel channel, Message message) {
+            // 省略大部分代码
+            WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable
+            // 推送任务
+            if (!workerManager.offer(workerTaskExecuteRunnable)) {
+                log.error("submit task: {} to wait queue error, current queue size: {} is full",
+                        taskExecutionContext.getTaskName(), workerManager.getWaitSubmitQueueSize());
+                sendDispatchRejectResult(channel, message, taskExecutionContext);
+            } 
+        }
+    }
+```
+
+WorkerManagerThread类实际处理消息
+
+```java
+    @Override
+    public void run() {
+        // 省略大部分代码
+        Thread.currentThread().setName("Worker-Execute-Manager-Thread");
+        while (!ServerLifeCycleManager.isStopped()) {
+            try {
+                if (!ServerLifeCycleManager.isRunning()) {
+                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
+                }
+                if (this.getThreadPoolQueueSize() <= workerExecThreads) {
+                    // 获取任务
+                    final WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable = waitSubmitQueue.take();
+                    // 提交任务
+                    workerExecService.submit(workerDelayTaskExecuteRunnable);
+                } 
+            } 
+        }
+    }
+```
+
+封装的WorkerTaskExecuteRunnable类的run方法
+
+```java
+    @Override
+    public void run() {
+        // 省略大部分代码
+        // 处理任务
+        executeTask(taskCallBack);
+    }
+    @Override
+    public void executeTask(TaskCallBack taskCallBack) throws TaskException {
+        if (task == null) {
+            throw new IllegalArgumentException("The task plugin instance is not initialized");
+        }
+        // 通过策略模式调用任务插件处理任务
+        task.handle(taskCallBack);
+    }
+```
+
+## 3.4、处理任务1
+
+根据上述的执行流程分析我们知道任务是以process为调度单元，任务通过任务ID来确定先后顺序
+
+![image-20230604112340116](assets/image-20230604112340116.png)
+
+```java
+    @Override
+    public int compareTo(Delayed o) {
+        if (o == null) {
+            return 1;
+        }
+        return Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
+    }
+```
+
+我们可以将任务队列的任务改为submit/future模式，通过添加params参数确定返回值和入参，通过future.get方法来阻塞等待任务将返回值输入到下一个任务中，可以将执行结果缓存到map中，类似于
+
+```java
+    @Override
+    public Object executeTask(TaskCallBack taskCallBack) throws TaskException {
+        if (task == null) {
+            throw new IllegalArgumentException("The task plugin instance is not initialized");
+        }
+        // 通过策略模式调用任务插件处理任务
+        return task.handle(taskCallBack);
+    }
+
+```
+
+WorkerManagerThread类中有一个map缓存结果
+
+```java
+ConcurrentHashMap concurrentHashMap = new ConcurrentHashMap();
+    @Override
+    public void run() {
+        // 省略大部分代码
+                if (this.getThreadPoolQueueSize() <= workerExecThreads) {
+                    // 获取任务
+                    final WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable = waitSubmitQueue.take();
+                    // 判断是否需要入参
+                    if (需要入参) {
+                        改造workerDelayTaskExecuteRunnable
+                    }
+                    // 提交任务
+                    future f = workerExecService.submit(workerDelayTaskExecuteRunnable);
+                    f.put(ID， f.get())
+                } 
+            } 
+        }
+    }
+```
+
+# 4、规划
+
+> 有丰富的后端开发经验，可以保证较快的进度，
+> 身边的同学也同时参加了开源之夏，可以相互帮助进步
+> 也是第一次参与开源项目，有兴趣长期投入事件参与开源事业，为他添砖加瓦
+
+## 4.1、项目研发第一阶段（6月-7月）：
+
+* 6月份大致完成任务1
+* 7月份完成任务2
+
+## 4.2、项目研发第二阶段（7月-8月）：
+
+* 完善任务1，2
+* 编写前端UI
+* 编写文档
+* 总结
+

大数据/kafka/1.kafka笔记.md

@@ -438,8 +438,9 @@ kafka的顺序消费使⽤场景不多，因为牺牲掉了性能，但是⽐如
 
 ![image-20210914195129983](../../照片/image-20210914195129983.png)
 * kafka中创建创建相应的主题
-* 消费者消费该主题的消息（轮询）
-* 消费者消费消息时判断消息的创建时间和当前时间是否超过30分钟（前提是订单没⽀付）
- *  如果是：去数据库中修改订单状态为已取消
- *  如果否：记录当前消息的offset，并不再继续消费之后的消息。等待1分钟后，再次 向kafka拉取该offset及之后的消息，继续进⾏判断，以此反复。
+  * 消费者消费该主题的消息（轮询）
+  * 消费者消费消息时判断消息的创建时间和当前时间是否超过30分钟（前提是订单没⽀付）
+  * 如果是：去数据库中修改订单状态为已取消
+
+ * 如果否：记录当前消息的offset，并不再继续消费之后的消息。等待1分钟后，再次 向kafka拉取该offset及之后的消息，继续进⾏判断，以此反复。
 

大数据/kafka/4.kafka事务.md

@@ -0,0 +1,67 @@
+## kafka 的事务机制
+
+Kafka 事务与数据库的事务定义基本类似，主要是一个原子性：**多个操作要么全部成功，要么全部失败**。Kafka 中的事务可以使应用程序将**消费消息、生产消息、提交消费位移**当作原子操作来处理。
+
+- KAFKA的事务机制，在底层依赖于幂等生产者，幂等生产者是 kafka 事务的必要不充分条件；
+- 事实上，开启 kafka事务时，kafka 会自动开启幂等生产者。
+
+## kafka的幂等性
+
+当 kafka producer 向 broker 中的 topic发送数据时，可能会因为网络抖动等各种原因，造成 producer 收不到 broker 的 ack 确认信息。kafka幂等性就会保证在生产者内部逻辑问题引起的消息重复消费的时候，只有一个数据可以被正确的发送。
+
+> **需要注意的是如果使用try/catch捕获，用send手动发送，则会被视为不同的消息**
+
+**原理**
+
+- 在 producer 端，每个 producer 都被 broker 自动分配了一个 Producer Id (PID)， producer 向 broker 发送的每条消息，在内部都附带着该 pid 和一个递增的 sequence number；
+- 在 broker 端，broker 为每个 topic 的每个 partition 都维护了一个当前写成功的消息的最大 PID-Sequence Number 元组；
+- 当 broker 收到一个比当前最大 PID-Sequence Number 元组小的 sequence number 消息时，就会丢弃该消息，以避免造成数据重复存储；
+- 当 broker 失败重新选举新的 leader 时, 以上去重机制仍然有效：因为 broker 的 topic 中存储的消息体中附带了 PID-sequence number 信息，且 leader 的所有消息都会被复制到 followers 中。当某个原来的 follower 被选举为新的 leader 时，它内部的消息中已经存储了PID-sequence number 信息，也就可以执行消息去重了。
+
+## kafka事务基本流程
+
+* **initTransactions**：方法用来初始化事务，这个方法能够执行的前提是配置了transactionalId，如果没有则会报出IllegalStateException：
+* **beginTransaction**：方法用来开启事务；
+* **sendOffsetsToTransaction**：方法为消费者提供在事务内的位移提交的操作；将偏移量提交到事务中，仅当整个交易(消费和生产)成功时，它才会提交。
+* **commitTransaction**：方法用来提交事务；
+* **abortTransaction**：方法用来中止事务，类似于事务回滚。
+
+```java
+producer.initTransactions();
+try {
+    producer.beginTransaction();
+    for (ProducerRecord<String, String> record : payload) {
+        producer.send(record);
+    }
+
+    Map<TopicPartition, OffsetAndMetadata> groupCommit = new HashMap<TopicPartition, OffsetAndMetadata>() {
+        {
+            put(new TopicPartition(TOPIC, 0), new OffsetAndMetadata(42L, null));
+        }
+    };
+    producer.sendOffsetsToTransaction(groupCommit, "groupId");
+    producer.commitTransaction();
+} catch (ProducerFencedException e) {
+    producer.close();
+} catch (KafkaException e) {
+    producer.abortTransaction();
+}
+```
+
+## 事务基本流程
+
+1. **存储对应关系，通过请求增加分区**
+   * Producer 在向新分区发送数据之前，首先向 TransactionalCoordinator 发送请求，使 TransactionalCoordinator 存储对应关系 (transactionalId, TopicPartition) 到主题 __transaction_state 中。
+2. **生产者发送消息**
+   * 基本与普通的发送消息相同，生产者调用 `producer.send()` 方法，发送数据到分区；
+   * 发送的请求中，包含 pid, epoch, sequence number 字段；
+3. **增加消费 offset 到事务**
+   * 生产者通过 `producer.senOffsetsToTransaction()` 接口，发送分区的 Offset 信息到事务协调者，协调者将分区信息增加到事务中；
+4. **事务提交位移**
+   * 在前面生产者调用事务提交 offset 接口后，会发送一个 TxnOffsetCommitRequest 请求到消费组协调者，消费组协调者会把 offset 存储到 Kafka 内部主题 __consumer_offsets 中。协调者会根据请求的 pid 与 epoch 验证生产者是否允许发起这个请求。
+   * 只有当事务提交之后，offset 才会对外可见。
+5. **提交或回滚事务**
+   * 用户调用 `producer.commitTransaction()` 或 `abortTransaction()` 方法，提交或回滚事务；
+   * 生产者完成事务之后，客户端需要显式调用结束事务，或者回滚事务。前者使消息对消费者可见，后者使消息标记为 abort 状态，对消费者不可见。无论提交或者回滚，都会发送一个 EndTxnRequest 请求到事务协调者，同时写入 PREPARE_COMMIT 或者 PREPARE_ABORT 信息到事务记录日志中。
+
+> **需要注意的是：如果事务性生产者（Transactional Producer）发送的消息没有被提交，消费者是不会读取该消息之后的数据的。**

算法/多线程/三个线程打印ABC.md

@@ -1,3 +1,7 @@
+## 三个线程轮流打印ABC，打印N次
+
+### synchronized锁同步
+
 ```java
 public void print(String str, int target) throws InterruptedException {
     for (int i = 0; i < n; i++) {
@@ -15,22 +19,10 @@ public void print(String str, int target) throws InterruptedException {
         }
     }
 }
-
-public void print(String str, int target) throws InterruptedException {
-    for (int i = 0; i < n; ) {
-        synchronized (obj) {
-            while (state % 3 != target) {
-                obj.wait();
-            }
-            state++;
-            i++;
-            System.out.print(str);
-            obj.notifyAll();
-        }
-    }
-}
 ```
 
+### lock锁同步
+
 ```java
 public void print(String str, int target) throws InterruptedException {
     for (int i = 0; i < n; ) {

面经/工作/秋招.md

@@ -0,0 +1,8 @@
+| 公司     | 网站                                                    | 投递时间 | 进度 |
+| -------- | ------------------------------------------------------- | -------- | ---- |
+| 百度     | https://talent.baidu.com/jobs/center                    | 7月10号  | 初筛 |
+| 科大讯飞 | https://campus.iflytek.com/official-pc/delivery         | 7月6号   | 初筛 |
+| oppo     | https://careers.oppo.com/university/oppo/center/history | 7月3号   | 初筛 |
+| 米哈游   | https://campus.mihoyo.com/#/campus/applyRecord          | 7月5号   | 初筛 |
+| 大疆     | https://we.dji.com/zh-CN/user                           | 7月7号   | 初筛 |
+

面经/问答/Mysql.md

@@ -14,13 +14,32 @@
 5. 定期维护数据库：数据库的性能和稳定性需要定期进行维护，包括定期清理无用的数据和索引，进行数据库备份和日志管理，定期更新统计信息等。定期维护可以保持数据库的健康状态，提升查询性能。
 6. 可以考虑使用分库分表技术，例如对于用户中奖详情单我选择了分库分表，因为有多个用户和不同活动同时竞争这个表，数据库访问压力比较大，所以就要选择分库分表
 
-## mysql低层数据结构
+## mysql底层数据结构
 
 1. B+ 树索引：MySQL 使用 B+ 树作为默认的索引结构。索引数据最好能按顺序排列，这样可以使用「二分查找法」高效定位数据，而且还要支持排序操作。这就天然要求我们使用平衡二叉树作为默认的数据结构，而平衡二叉树本身是一个B树，每个节点只能有两个子节点那么当节点个数越多的时候，树的高度也会相应变高，这样就会增加磁盘的 I/O 次数，从而影响数据查询的效率。B 树的每一个节点最多可以包括 M 个子节点，M 称为 B 树的阶，所以 B 树就是一个多叉树。B+ 树就是对 B 树做了一个升级，他和B树相比有如下几个优点
    1. B+ 树的非叶子节点不存放实际的记录数据，仅存放索引，因此数据量相同的情况下，相比存储即存索引又存记录的 B 树查询效率更高，因为IO次数更少
    2. B+ 树在删除根节点的时候，由于存在冗余的节点，所以不会发生复杂的树的变形，B 树则不同，B 树没有冗余节点，删除节点的时候非常复杂
    3. B+ 树所有叶子节点间还有一个链表进行连接，这种设计对范围查找非常有帮助
-2. 数据文件：MySQL 数据库将数据存储在磁盘上的数据文件中，低层由表空间构成，表空间由段（segment）、区（extent）、页（page）、行（row）组成
+2. 数据文件：MySQL 数据库将数据存储在磁盘上的数据文件中，底层由表空间构成，表空间由段（segment）、区（extent）、页（page）、行（row）组成
+
+![image-20230524111644588](assets/image-20230524111644588.png)
+
+为什么分区？
+
+页要连续，顺序IO读写快，每个区都维护一个链表，代表空闲区的指针
+
+- `FREE`链表：同一个段中，所有页都是空闲的区对应的`XDES Entry`结构会被加入到这个链表。注意和直属于表空间的`FREE`链表区别开了，此处的`FREE`链表是附属于某个段的。
+- `NOT_FULL`链表：同一个段中，仍有空闲空间的区对应的`XDES Entry`结构会被加入到这个链表。
+- `FULL`链表：同一个段中，已经没有空闲空间的区对应的`XDES Entry`结构会被加入到这个链表。
+
+为什么分段？
+
+区分叶子节点和非叶子节点
+
+页？
+
+稀疏索引，分组
+
 3. InnoDB 存储引擎：InnoDB 是 MySQL 的一种常用存储引擎，它使用了多版本并发控制（MVCC）和行级锁来实现高并发性和事务支持。InnoDB 存储引擎的核心数据结构包括页（page）、记录（record）、索引（index）和事务日志（transaction log）等。
 
 ## 联合索引定义要注意哪些点

面经/问答/kafka.md

@@ -0,0 +1,44 @@
+## kafka为什么快？
+
+### 顺序读写
+
+kafka的存储方案是**顺序追加写日志 + 稀疏哈希索引**
+
+![image-20230519214345638](assets/image-20230519214345638.png)
+
+1. kafka 中消息是以主题 Topic 为基本单位进行归类的，这里的 Topic 是逻辑上的概念，实际上在磁盘存储是根据分区 Partition 存储的, 即每个 Topic 被分成多个 Partition，分区 Partition 的数量可以在主题 Topic 创建的时候进行指定。
+2. Partition 分区主要是为了解决 Kafka 存储的水平扩展问题而设计的， 如果一个 Topic 的所有消息都只存储到一个 Kafka Broker上的话， 对于 Kafka 每秒写入几百万消息的高并发系统来说，这个 Broker 肯定会出现瓶颈， 故障时候不好进行恢复，所以 Kafka 将 Topic 的消息划分成多个 Partition， 然后均衡的分布到整个 Kafka Broker 集群中。
+3. Partition 分区内每条消息都会被分配一个唯一的消息 id,即我们通常所说的 偏移量 Offset, 因此 kafka 只能保证每个分区内部有序性,并不能保证全局有序性。
+4. 然后每个 Partition 分区又被划分成了多个 LogSegment，这是为了防止 Log 日志过大，Kafka 又引入了日志分段(LogSegment)的概念，将 Log 切分为多个 LogSegement，相当于一个巨型文件被平均分割为一些相对较小的文件，这样也便于消息的查找、维护和清理。这样在做历史数据清理的时候，直接删除旧的 LogSegement 文件就可以了。
+4. Log 日志在物理上只是以文件夹的形式存储，而每个 LogSegement 对应磁盘上的一个日志文件和两个索引文件，以及可能的其他文件(比如以".snapshot"为后缀的快照索引文件等)
+
+/index文件中会生成三份文件XX.log,XX.index,XX.timeindex，日志是分块存储的，时间戳索引从XX.timeindex中找到对应的offset，再从XX.index中索引XX.log。
+
+### 页缓存
+
+页缓存相对来说比较简单，页缓存在操作系统层面是保存数据的一个基本单位，Kafka 避免使用 JVM，直接使用操作系统的页缓存特性提高处理速度，进而避免了JVM GC 带来的性能损耗。
+
+### 零拷贝
+
+kafka就采用零拷贝技术来消费数据
+
+### 批量操作
+
+在 kafka 中页提高了大量批处理的 API ，可以对数据进行统一的压缩合并，通过更小的数据包在网络中进行数据发送，再进行后续处理，这在大量数据处理中，效率提高是非常明显的。
+
+## kafka如何保证顺序消费
+
+1. 1 个 Topic 只对应一个 Partition。
+2. （推荐）发送消息的时候指定 key/Partition。
+
+## Kafka 如何保证消息不重复消费
+
+**kafka 出现消息重复消费的原因：**
+
+- 服务端侧已经消费的数据没有成功提交 offset（根本原因）。
+- Kafka 侧 由于服务端处理业务时间长或者网络链接等等原因让 Kafka 认为服务假死，触发了分区 rebalance。
+
+**解决方案：**
+
+* 消费消息服务做幂等校验，比如 Redis 的 set、MySQL 的主键等天然的幂等功能。这种方法最有效。
+* 将 **`enable.auto.commit`** 参数设置为 false，关闭自动提交，开发者在代码中手动提交 offset。

面经/问答/并发编程.md

@@ -79,7 +79,7 @@ CAS是Compare and Swap的缩写，即比较并交换。它是一种无锁算法
 
 * AQS
 
-AQS抽象队列同步器。它是Java并发包中锁和同步工具的核心实现上面的所说的LOCK锁就是基于AQS实现的
+AQS抽象队列同步器。它是Java并发包中锁和同步工具的核心实现上面的所说的LOCK锁就是基于AQS实现的。
 
 AQS提供了一种通用的框架，用于实现线程间的协作和同步操作。它的核心思想是使用一个先进先出的等待队列来管理线程状态，同时支持独占模式和共享模式两种同步方式。
 
@@ -99,7 +99,7 @@ AQS提供了一种通用的框架，用于实现线程间的协作和同步操
 
 * LockSupport不需要获取锁对象，因此避免了可能出现的死锁问题。
 
-* LockSupport可以响应中断，而Object.wait()/notify()方法无法响应中断。
+* notify只能随机释放一个线程，并不能指定某个特定线程，notifyAll是释放锁对象中的所有线程。而unpark方法可以唤醒指定的线程。
 
 * LockSupport可以先执行unpark()方法，然后再执行park()方法，而Object.notify()方法必须在对应的wait()方法之前执行。
 
@@ -158,7 +158,9 @@ AQS提供了一种通用的框架，用于实现线程间的协作和同步操
 * **`ThreadPoolExecutor.DiscardPolicy`：** 不处理新任务，直接丢弃掉。
 * **`ThreadPoolExecutor.DiscardOldestPolicy`：** 此策略将丢弃最早的未处理的任务请求。
 
-### 如何设定线程池的大小？
+### 如何设定线程池的大小？（CPU 核心数 - N）
+
+Runtime.getRuntime().availableProcessors()
 
 * **CPU 密集型任务(N+1)**
 
@@ -171,13 +173,13 @@ CPU 密集型简单理解就是利用 CPU 计算能力的任务比如你在内
 * Future类的get方法
 
 ```java
-private static final int NEW          = 0;
-private static final int COMPLETING   = 1;
-private static final int NORMAL       = 2;
-private static final int EXCEPTIONAL  = 3;
-private static final int CANCELLED    = 4;
-private static final int INTERRUPTING = 5;
-private static final int INTERRUPTED  = 6;
+private static final int NEW          = 0; // 初始状态，FutureTask刚被创建，正在计算中都是该状态。
+private static final int COMPLETING   = 1; // 中间状态，表示计算已完成正在对结果进行赋值，或正在处理异常
+private static final int NORMAL       = 2; // 终止状态，表示计算已完成，结果已经被赋值。
+private static final int EXCEPTIONAL  = 3; // 终止状态，表示计算过程已经被异常打断。
+private static final int CANCELLED    = 4; // 终止状态，表示计算过程已经被cancel操作终止。
+private static final int INTERRUPTING = 5; // 中间状态，表示计算过程已开始并且被中断，正在修改状态。
+private static final int INTERRUPTED  = 6; // 终止状态，表示计算过程已开始并且被中断，目前已完全停止。
 
 public V get() throws InterruptedException, ExecutionException {
     int s = state;
@@ -210,7 +212,7 @@ public void run() {
                 ran = false;
                 setException(ex);
             }
-            if (ran)
+            if (ran)·
                 // 改变状态
                 set(result);
         }

面经/问答/操作系统.md

@@ -189,3 +189,95 @@ TLB即为页表缓存、转址旁路缓存、快表等。有了 TLB 后，那么
 1. **静态分配策略**：一个进程必须在执行前就申请到它所需要的全部资源
 2. **层次分配策略**：一个进程得到某一次的一个资源后，它只能再申请较高一层的资源
 
+## 零拷贝
+
+**传统的拷贝**
+
+![image-20230524103611405](assets/image-20230524103611405.png)
+
+**一共需要2次CPU拷贝，很消耗时间。**
+
+
+
+零拷贝（Zero-Copy）是一种 `I/O` 操作优化技术，可以快速高效地将数据从文件系统移动到网络接口，而不需要将其从内核空间复制到用户空间。
+
+技术的核心就要是减少CPU占用和上下文切换
+
+它主要由以下4点技术结合实现：
+
+### 1. DMA技术
+
+DMA（Direct Memory Access）是一种硬件实现的技术，能够直接将数据从设备传输到内存中，而无需CPU参与其中。
+
+### 2. 缓冲区技术
+
+在实现零拷贝时，需要使用多个缓冲区来存储数据。通过缓冲区技术可以实现不同层级的内存之间的数据传输。
+
+### 3. 文件映射技术
+
+文件映射技术可以将文件或磁盘块映射到内存空间中，使得应用程序可以直接访问这些文件或磁盘块，从而实现零拷贝。
+
+**mmap + write**
+
+![image-20230524101601923](assets/image-20230524101601923.png)
+
+`mmap()` 系统调用函数会直接把内核缓冲区里的数据「**映射**」到用户空间，这样，操作系统内核与用户空间就不需要再进行任何的数据拷贝操作。
+
+具体过程如下：
+
+- 应用进程调用了 `mmap()` 后，DMA 会把磁盘的数据拷贝到内核的缓冲区里。接着，应用进程跟操作系统内核「共享」这个缓冲区；
+- 应用进程再调用 `write()`，操作系统直接将内核缓冲区的数据拷贝到 socket 缓冲区中，这一切都发生在内核态，由 CPU 来搬运数据；
+- 最后，把内核的 socket 缓冲区里的数据，拷贝到网卡的缓冲区里，这个过程是由 DMA 搬运的。
+
+但这还不是最理想的零拷贝，因为仍然需要通过 CPU 把内核缓冲区的数据拷贝到 socket 缓冲区里，而且仍然需要 4 次上下文切换，因为系统调用还是 2 次
+
+### 4. 网络协议技术
+
+网络协议技术可以帮助实现在网络上进行零拷贝传输，例如在TCP/IP协议栈中使用sendfile系统调用。
+
+**sendfile**
+
+![image-20230524101812043](assets/image-20230524101812043.png)
+
+- 第一步，通过 DMA 将磁盘上的数据拷贝到内核缓冲区里；
+- 第二步，缓冲区描述符和数据长度传到 socket 缓冲区，这样网卡的 SG-DMA 控制器就可以直接将内核缓存中的数据拷贝到网卡的缓冲区里，此过程不需要将数据从操作系统内核缓冲区拷贝到 socket 缓冲区中，这样就减少了一次数据拷贝；
+
+## PageCache 有什么作用？
+
+**零拷贝使用了 PageCache 技术**，**PageCache其实本质上就是一种磁盘高速缓存**，通过 DMA 把磁盘里的数据搬运到内存里，这样就可以用读内存替换读磁盘。
+
+**PageCache 来缓存最近被访问的数据**，当空间不足时淘汰最久未被访问的缓存。所以，读磁盘数据的时候，优先在 PageCache 找，如果数据存在则可以直接返回；如果没有，则从磁盘中读取，然后缓存 PageCache 中。
+
+还有一点，读取磁盘数据的时候，需要找到数据所在的位置，但是对于机械磁盘来说，就是通过磁头旋转到数据所在的扇区，再开始「顺序」读取数据，但是旋转磁头这个物理动作是非常耗时的，为了降低它的影响，**PageCache 使用了「预读功能」**。
+
+比如，假设 read 方法每次只会读 `32 KB` 的字节，虽然 read 刚开始只会读 0 ～ 32 KB 的字节，但内核会把其后面的 32～64 KB 也读取到 PageCache，这样后面读取 32～64 KB 的成本就很低，如果在 32～64 KB 淘汰出 PageCache 前，进程读取到它了，收益就非常大。
+
+所以，PageCache 的优点主要是两个：
+
+- 缓存最近被访问的数据；
+- 预读功能；
+
+这两个做法，将大大提高读写磁盘的性能。
+
+## 大文件传输用什么方式实现？
+
+**在传输大文件（GB 级别的文件）的时候，PageCache 会不起作用，那就白白浪费 DMA 多做的一次数据拷贝，造成性能的降低，即使使用了 PageCache 的零拷贝也会损失性能**
+
+**在高并发的场景下，针对大文件的传输的方式，应该使用「异步 I/O + 直接 I/O」来替代零拷贝技术**。
+
+异步I/O：
+
+它把读操作分为两部分：
+
+- 前半部分，内核向磁盘发起读请求，但是可以**不等待数据就位就可以返回**，于是进程此时可以处理其他任务；
+- 后半部分，当内核将磁盘中的数据拷贝到进程缓冲区后，进程将接收到内核的**通知**，再去处理数据；
+
+直接 I/O 应用场景常见的两种：
+
+- 应用程序已经实现了磁盘数据的缓存，那么可以不需要 PageCache 再次缓存，减少额外的性能损耗。在 MySQL 数据库中，可以通过参数设置开启直接 I/O，默认是不开启；
+- 传输大文件的时候，由于大文件难以命中 PageCache 缓存，而且会占满 PageCache 导致「热点」文件无法充分利用缓存，从而增大了性能开销，因此，这时应该使用直接 I/O。
+
+另外，由于直接 I/O 绕过了 PageCache，就无法享受内核的这两点的优化：
+
+- 内核的 I/O 调度算法会缓存尽可能多的 I/O 请求在 PageCache 中，最后「**合并**」成一个更大的 I/O 请求再发给磁盘，这样做是为了减少磁盘的寻址操作；
+- 内核也会「**预读**」后续的 I/O 请求放在 PageCache 中，一样是为了减少对磁盘的操作；

面经/问答/计网.md

@@ -301,3 +301,83 @@ HTTP 长连接通常使用 HTTP/1.1 中的 "keep-alive" 机制，在客户端和
 WebSocket 的长连接是指在服务器和客户端之间建立一条持久的双向通信通道，该通道会一直保持打开状态，直到显示地关闭。这种连接不仅可以支持即时通信和实时数据传输，而且可以使得服务器可以主动向客户端推送消息，从而实现更高效的数据传输和通信。此外，WebSocket 还有一个心跳检测机制（即“ping/pong”），用于检测连接是否处于活动状态，并保证连接的稳定性。
 
 综上所述，HTTP 的长连接主要是为了减少建立和断开连接所带来的性能开销，而 WebSocket 的长连接则是为了支持实时通信和实时数据传输，并保持连接的稳定性。此外，WebSocket 通过心跳检测机制可以更好地保护长连接，而 HTTP 长连接则没有这个机制。
+
+## HTTPS建立连接过程
+
+TCP握手之后HTTPS需要四次握手建立连接。
+
+*1. ClientHello*
+
+首先，由客户端向服务器发起加密通信请求，也就是 `ClientHello` 请求。
+
+在这一步，客户端主要向服务器发送以下信息：
+
+（1）客户端支持的 TLS 协议版本，如 TLS 1.2 版本。
+
+（2）客户端生产的随机数（`Client Random`），后面用于生成「会话秘钥」条件之一。
+
+（3）客户端支持的密码套件列表，如 RSA 加密算法。
+
+*2. SeverHello*
+
+服务器收到客户端请求后，向客户端发出响应，也就是 `SeverHello`。服务器回应的内容有如下内容：
+
+（1）确认 TLS 协议版本，如果浏览器不支持，则关闭加密通信。
+
+（2）服务器生产的随机数（`Server Random`），也是后面用于生产「会话秘钥」条件之一。
+
+（3）确认的密码套件列表，如 RSA 加密算法。
+
+（4）服务器的数字证书。
+
+*3.客户端回应*
+
+客户端收到服务器的回应之后，首先通过浏览器或者操作系统中的 CA 公钥，确认服务器的数字证书的真实性。
+
+如果证书没有问题，客户端会**从数字证书中取出服务器的公钥**，然后使用它加密报文，向服务器发送如下信息：
+
+（1）一个随机数（`pre-master key`）。该随机数会被服务器公钥加密。
+
+（2）加密通信算法改变通知，表示随后的信息都将用「会话秘钥」加密通信。
+
+（3）客户端握手结束通知，表示客户端的握手阶段已经结束。这一项同时把之前所有内容的发生的数据做个摘要，用来供服务端校验。
+
+上面第一项的随机数是整个握手阶段的第三个随机数，会发给服务端，所以这个随机数客户端和服务端都是一样的。
+
+**服务器和客户端有了这三个随机数（Client Random、Server Random、pre-master key），接着就用双方协商的加密算法，各自生成本次通信的「会话秘钥」**。
+
+*4. 服务器的最后回应*
+
+服务器收到客户端的第三个随机数（`pre-master key`）之后，通过协商的加密算法，计算出本次通信的「会话秘钥」。
+
+然后，向客户端发送最后的信息：
+
+（1）加密通信算法改变通知，表示随后的信息都将用「会话秘钥」加密通信。
+
+（2）服务器握手结束通知，表示服务器的握手阶段已经结束。这一项同时把之前所有内容的发生的数据做个摘要，用来供客户端校验。
+
+至此，整个 TLS 的握手阶段全部结束。接下来，客户端与服务器进入加密通信，就完全是使用普通的 HTTP 协议，只不过用「会话秘钥」加密内容。
+
+## HTTPS特点
+
+1. 优化stream
+
+![image-20230524110037552](assets/image-20230524110037552.png)
+
+* 一个TCP连接包含多个stream
+* 每个stream都是一个双向数据流包含多个message
+* message请求或者响应，包含多个frame
+* frame是https最小帧格式，包含头frame和数据frame
+
+2. 压缩头部
+
+* 静态字典
+* 动态字典
+* 哈夫曼编码
+
+## HTTP 与 HTTPS 有哪些区别？
+
+- HTTP 是超文本传输协议，信息是明文传输，存在安全风险的问题。HTTPS 则解决 HTTP 不安全的缺陷，在 TCP 和 HTTP 网络层之间加入了 SSL/TLS 安全协议，使得报文能够加密传输。
+- HTTP 连接建立相对简单， TCP 三次握手之后便可进行 HTTP 的报文传输。而 HTTPS 在 TCP 三次握手之后，还需进行 SSL/TLS 的握手过程，才可进入加密报文传输。
+- 两者的默认端口不一样，HTTP 默认端口号是 80，HTTPS 默认端口号是 443。
+- HTTPS 协议需要向 CA（证书权威机构）申请数字证书，来保证服务器的身份是可信的。

面经/项目/中煤项目.md

@@ -157,4 +157,22 @@ ack = 1
 
 流量分析项目是六所网络安全部门的一个子项目，它主要功能是识别工控设备异常流量，达到防范攻击和攻击溯源的功能，我主要负责的是工控协议流量解析部分，负责12种工控协议的解析。协议解析主要参照各个协议的官网进行翻译，主要通过策略模式进行协议选择，选择协议之后通过责任链模式进行协议判断，主要实现了对方提供的接口，输出的是解析成功的json格式的对象，通过SPI机制生成一个三方jar包。
 
-该项目的主要难点就是协议需要参照协议官网进行一个字节一个字节的分析解析，并对其错误情况进行判断。
+该项目的主要难点就是协议需要参照协议官网进行一个字节一个字节的分析解析，并对其错误情况进行判断。
+
+
+
+## 开场白
+
+面试官您好，我叫孙浩博，本科和硕士就读于西安电子科技大学，专业是计算机科学。在实验室在学校我主要参与中煤大数据煤矿项目，该项目主要承担煤矿数据的采集、传输、存储、分发、计算等任务。我主要负责其中的采集模块和kafka监控模块，在电子六所实习我主要从事软件开发实习生，主要跟进流量分析项目。
+
+在平常的工作和学习中，我会写博客来记录自己的经验和总结，以便能够更好地总结和交流。个人博客地址在我的简历上面
+
+
+
+good morning. I am glad to be here for this interview. First let me introduce myself. My name is Sun Haobo, 23 , come from xidian University where I major in Computer Science
+
+During my time at university, I actively participated in the CMDB (Coal Mine Big Data) project, which primarily focused on various tasks related to coal mine data, including data collection, transmission, storage, distribution, and computation. My primary responsibilities were in the data collection module and the Kafka monitoring module.
+
+During my internship at NCSE. I worked as a software development intern and actively contributed to the development of a traffic analysis project. Specifically, my responsibilities focused on the development of the parsing module.
+
+As part of my regular work and studies, I maintain a personal blog to document my experiences, enabling me to effectively summarize and share my knowledge. The link to my personal blog can be found on my resume.