java笔记整理路由算法：根据文档id或路由计算目标的分片id

#用户输入作为CSS内容，导致攻击者可以进行闭合绕过

XSS 攻击的分类
XSS分为：存储型 、反射型 、DOM型

存储型XSS：持久化，代码是存储在 服务器 中的，如在个人信息或发表文章等地方，插入代码，如果没有过滤或过滤不严，那么这些代码将储存到服务器中，用户访问该页面的时候触发代码执行。这种XSS比较危险，容易造成蠕虫，盗窃cookie。
反射型XSS：非持久化，需要欺骗用户自己去点击链接才能触发XSS代码（服务器中没有这样的页面和内容），一般容易出现在 web 页面。反射型XSS大多数是用来盗取用户的Cookie信息。
DOM型XSS：不经过后端，DOM-XSS漏洞是基于文档对象模型(Document Objeet Model，DOM)的一种漏洞，DOM-XSS是通过 url 传入参数去控制触发的，其实也属于反射型XSS。

反射型

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uge3h0uG-1666199870569)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018163747368.png)]

存储型

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rBCarc88-1666199870570)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018163807773.png)]

DOM型
如下图，我们在 URL 中传入参数的值，然后客户端页面通过 js 脚本利用 DOM 的方法获得 URL 中参数的值，再通过 DOM 方法赋值给选择列表，该过程没有经过后端，完全是在前端完成的。

所以，我们就可以在我们输入的参数上做手脚了。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aGqlVsy3-1666199870572)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018163908951.png)]

对 XSS 漏洞的简单攻击
反射型XSS
先放出源代码

//前端 1.html：

//后端 action.php：

这里有一个用户提交的页面，用户可以在此提交数据，数据提交之后给后台处理：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tn0lh3S2-1666199870574)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164029630.png)]

所以，我们可以在输入框中提交数据： ，看看会有什么反应：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eOgjtdwj-1666199870576)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164043262.png)]

页面直接弹出了 hack 的页面，可以看到，我们插入的语句已经被页面给执行了。

这就是最基本的 反射型的 XSS 漏洞 ，这种漏洞数据流向是： 前端–>后端–>前端

存储型XSS
先给出源代码

//前端：2.html

$sql="insert into xss value ($id,'$name')";
$result=mysql_query($sql);

?>
//供其他用户访问页面：show2.php

这里有一个用户提交的页面，数据提交给后端之后，后端存储在数据库中 （比反射型多了一个存储进数据库的过程，所以会存在存储型 XSS 漏洞）。然后当其他用户访问另一个页面的时候，后端调出该数据，显示给另一个用户，XSS 代码就被执行了。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GmiVRD3t-1666199870579)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164241844.png)]

我们输入 1 和 ，注意，这里的 hack 的单引号要进行转义，因为 sql 语句中的 $name 是单引号的，所以这里不转义的话就会闭合 sql 语句中的单引号。不然注入不进去。提交了之后，我们看看数据库：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mKCJGIuM-1666199870581)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164257262.png)]

可以看到，我们的 XSS 语句已经插入到数据库中了
然后当其他用户访问 show2.php 页面时，我们插入的 XSS 代码就执行了。
存储型 XSS 的数据流向是：前端–>后端–>数据库–>后端–>前端

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HGJbwEOK-1666199870586)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164317845.png)]

DOM型XSS
先放上源代码

// 前端3.html

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Z48QJgyM-1666199870587)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164415208.png)]

我们可以输入 <img src=1 οnerrοr=alert(‘hack’)> ，然后看看页面的变化：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NqjIv46T-1666199870588)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164424050.png)]

页面直接弹出了 hack 的页面，可以看到，我们插入的语句已经被页面给执行了。
这就是 DOM型XSS 漏洞，这种漏洞数据流向是： 前端–>浏览器

XSS 攻击过程
反射型 XSS 漏洞
Alice 经常浏览某个网站，此网站为 Bob 所拥有。Bob 的站点需要 Alice 使用用户名/密码进行登录，并存储了 Alice 敏感信息(比如银行帐户信息)。
Tom 发现 Bob 的站点存在反射性的 XSS 漏洞。
Tom 利用 Bob 网站的反射型 XSS 漏洞编写了一个exp，做成链接的形式，并利用各种手段诱使 Alice 点击。
Alice 在登录到 Bob 的站点后，浏览了 Tom 提供的恶意链接。
嵌入到恶意链接中的恶意脚本在 Alice 的浏览器中执行。此脚本 盗窃敏感信息 (cookie、帐号信息等信息)。然后在 Alice 完全不知情的情况下将这些信息发送给 Tom。
Tom 利用获取到的 cookie 就可以以 Alice 的身份登录 Bob 的站点，如果脚本的功更强大的话，Tom 还可以对 Alice 的浏览器做控制并进一步利用漏洞控制。

存储型XSS漏洞
Bob 拥有一个 Web 站点，该站点允许用户发布信息/浏览已发布的信息。
Tom 检测到 Bob 的站点存在存储型的 XSS 漏洞。
Tom 在 Bob 的网站上发布一个带有恶意脚本的热点信息，该热点信息存储在了 Bob 的服务器的数据库中，然后吸引其它用户来阅读该热点信息。
Bob 或者是任何的其他人如 Alice 浏览该信息之后，Tom 的恶意脚本就会执行。
Tom的恶意脚本执行后，Tom就可以对浏览器该页面的用户发动一起 XSS 攻击。
六、XSS漏洞的危害
从以上我们可以知道，存储型的XSS危害最大。因为他存储在服务器端，所以不需要我们和被攻击者有任何接触，只要被攻击者访问了该页面就会遭受攻击。而反射型和DOM型的XSS则需要我们去诱使用户点击我们构造的恶意的URL，需要我们和用户有直接或者间接的接触，比如利用社会工程学或者利用在其他网页挂马的方式。

那么，利用XSS漏洞可以干什么呢？

如果JS水平一般的话，还可以利用网上免费的 XSS 平台来构造代码实施攻击。

七、XSS 的防御
XSS防御的总体思路是：对 用户的输入(和URL参数) 进行 过滤 ，对 输出 进行 html编码。也就是对用户提交的所有内容进行过滤，对url中的参数进行过滤；然后对动态输出到页面的内容进行 html 编码，转换为 html 实体，使脚本无法在浏览器中执行。

对输入的内容进行过滤，可以分为黑名单过滤和白名单过滤。黑名单过滤虽然可以拦截大部分的 XSS 攻击，但是还是存在被绕过的风险。白名单过滤虽然可以基本杜绝 XSS 攻击，但是真实环境中一般是不能进行如此严格的白名单过滤的。

对输出进行 html 编码，就是通过函数，将用户的输入的数据进行 html 编码，使其不能作为脚本运行。
如下，是使用 php 中的 htmlspecialchars 函数对用户输入的 name 参数进行 html 编码，将其转换为 html 实体：

#使用htmlspecialchars函数对用户输入的name参数进行html编码，将其转换为html实体

#使用htmlspecialchars函数对用户输入的name参数进行html编码，将其转换为html实体
$name = htmlspecialchars( $_GET[ ‘name’ ] );
如下，图一是没有进行 html 编码的，图2是进行了 html 编码的。经过 html 编码后 script 标签被当成了 html 实体。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YpxRPrOM-1666199870591)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164517552.png)]

我们还可以服务端设置会话 Cookie 的 HTTP Only 属性，这样，客户端的 JS 脚本就不能获取Cookie 信息了。

3 csrf
cookie session token
我觉得在开始学习CSRF之前应该先学会区分这三种东西：cookie session token

cookie:
Cookie，有时也用其复数形式 Cookies。类型为“小型文本文件”，是某些网站为了辨别用户身份，进行Session跟踪而储存在用户本地终端上的数据（通常经过加密），由用户客户端计算机暂时或永久保存的信息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9jxO45Hp-1666199870593)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164826269.png)]

1 浏览器第一次访问服务端时，服务器此时肯定不知道他的身份，所以创建一个独特的身份标识数据，格式为key=value，放入到Set-Cookie字段里，随着响应报文发给浏览器。
2 浏览器看到有Set-Cookie字段以后就知道这是服务器给的身份标识，于是就保存起来，下次请求时会自动将此key=value值放入到Cookie字段中发给服务端。
3 服务端收到请求报文后，发现Cookie字段中有值，就能根据此值识别用户的身份然后提供个性化的服务。

session:
如果将账户的一些信息都存入Cookie中的话，一旦信息被拦截，那么我们所有的账户信息都会丢失掉。所以就出现了Session，在一次会话中将重要信息保存在Session中，浏览器只记录SessionId一个SessionId对应一次会话请求。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cic9zuUQ-1666199870598)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164925599.png)]

token:

Session是将要验证的信息存储在服务端，并以Session Id和数据进行对应，SessionId由客户端存储，在请求时将SessionId也带过去，因此实现了状态的对应。而Token是在服务端将用户信息经过Base64Url编码过后传给在客户端，每次用户请求的时候都会带上这一段信息，因此服务端拿到此信息进行解密后就知道此用户是谁了，这个方法叫做JWT(Json Web Token)。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-k5sh7Hy1-1666199870602)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164934470.png)]

概念：
跨站请求伪造（Cross-Site Request Forgery，简称CSRF）是指，攻击者可能利用网页中的恶意代码强迫受害者浏览器向被攻击的Web站点发送伪造的请求，篡夺受害者的认证Cookie等身份信息，从而假冒受害者对目标站点执行指定的操作。

攻击原理：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YwY2SpY2-1666199870603)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018164949045.png)]总结一下：要想实现这个攻击，需要满足：登录受信任网站A，并在本地生成Cookie。在不登出A的情况下，访问危险网站B。

1、客户端通过账户密码登录访问网站A。

2、网站A验证客户端的账号密码，成功则生成一个sessionlD，并返回给客户端存储在浏览器中。

3、该客户端Tab—个新页面访问了网站B。

4、网站B自动触发要求该客户端访问网站A。(即在网站B中有链接指向网站A)

5、客户端通过网站B中的链接访问网站A。(此时携带有合法的SessionID进行访问站A的)

6、此时网站A只需检验sessionIlD是否合法，合法则执行相应的操作。(因此具体啥工具就得看链接，以及网站B要求访问时携带的数据

csrf的两类：
一：Get类型的csrf

仅仅须要一个HTTP请求。就能够构造一次简单的CSRF

样例：

银行站点A：它以GET请求来完毕银行转账的操作，如：

http://www.mybank.com/Transfer.php?toBankId=11&money=1000
危险站点B：它里面有一段HTML的代码例如以下：

<img src=http://www.mybank.com/Transfer.php?toBankId=11&money=1000>
首先。你登录了银行站点A，然后访问危险站点B，噢，这时你会发现你的银行账户少了1000块。

为什么会这样呢？原因是银行站点A违反了HTTP规范，使用GET请求更新资源。

在访问危险站点B的之前。你已经登录了银行站点A，而B中的 一个合法的请求，但这里被不法分子利用了）。

所以你的浏览器会带上你的银行站点A的Cookie发出Get请求，去获取资源以GET的方式请求第三方资源（这里的第三方就是指银行站点了）

demo：dvwa:low level

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Jupbx7lI-1666199870604)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018165113424.png)]

这是一个修改密码的界面，然后我们点击右下角查看源码，

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-X4rzqf1G-1666199870608)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018165129423.png)]

首先分析一下源码，他这个首先就是通过get方式传进password_new和password_conf这两个参数，然后判断用户输入的这两个参数是否一样。没有什么防控csrf的措施，所以很容易受到CSRF的攻击。

于是构造url:

http://64336ea7-ad15-47e2-bf11-17a61a7c78e4.node4.buuoj.cn:81/vulnerabilities
/csrf/?password_new=123456&password_conf=123456&change=change

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dmIZbH3T-1666199870609)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018165148836.png)]

登陆成功。over

现实中，攻击者往往会先搭建一个站点，然后上传一个html文档，该文档中含有恶意的链接。让后将这个html文档的地址发送给用户，用户一旦点击将会自动加载恶意链接完成攻击。

POST类型的CSRF:

在普通用户的眼中，点击网页->打开试看视频->购买视频是一个很正常的一个流程。可是在攻击者的眼中可以算正常但又不正常的，当然不正常的情况下，是在开发者安全意识不足所造成的。攻击者在购买处抓到购买时候网站处理购买(扣除)用户余额的地址。

比如:

/coures/user/handler666buy.php
通过提交表单，buy.php处理购买的信息，这里的666为视频ID。那么攻击者现在构造一个链接，链接中包含以下内容。

CSRF漏洞的防御
一：请求地址中添加 token 并验证。

CSRF 攻击之所以能够成功，是因为黑客可以完全伪造用户的请求，该请求中所有的用户验证信息都是存在于 cookie 中，因此黑客可以在不知道这些验证信息的情况下直接利用用户自己的 cookie 来通过安全验证。要抵御 CSRF关键在于在请求中放入黑客所不能伪造的信息，并且该信息不存在于 cookie 之中。可以在 HTTP 请求中以参数的形式加入一个随机产生的 token，并在服务器端建立一个拦截器来验证这token，如果请求中没有 token 或者 token 内容不正确，则认为可能是 CSRF 攻击而拒绝该请求。

二：验证 HTTP Referer 字段

根据 HTTP 协议，在 HTTP 头中有一个字段叫Referer，它记录了该 HTTP 请求的来源地址。在通常情况下，访问一个安全受限页面的请求来自于同一个网站，比如需要访问 ：

Refer:http://bank.example/withdraw?account=bob&amount=1000000&for=Mallory
时，该转帐请求的 Referer 值就会是转账按钮所在的页面的 URL，通常是以 bank.example 域名开头的地址。而如果黑客要对银行网站实施 CSRF 攻击，他只能在他自己的网站构造请求，当用户通过黑客的网站发送请求到银行时，该请求的 Referer 是指向黑客自己的网站。因此，要防御 CSRF 攻击，银行网站只需要对于每一个转账请求验证其 Referer 值，如果是以 bank.example 开头的域名，则说明该请求是来自银行网站自己的请求，是合法的。如果 Referer 是其他网站的话，则有可能是黑客的 CSRF 攻击，拒绝该请求。

这种方法并非万无一失。Referer 的值是由浏览器提供的，虽然 HTTP 协议上有明确的要求，但是每个浏览器对于 Referer 的具体实现可能有差别，并不能保证浏览器自身没有安全漏洞。

demo：dvwa middle

用刚才的办法，发现不对，于是查看源码。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-V1338jcl-1666199870611)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221018165335148.png)]

其中，stripos(strs, str)函数：返回字符str在字符串strs中的位置 。 HTTP_REFERER表示数据包中的referer字段（表示数据包的来源链接），SERVER_NAME表示数据包中的host（要访问的主机地址），所以后端会检测看referer中是否有host，如果有才会通过。

直接将low级别制作的html文档名称改为[host].html，这样referer中就会存在host，用户一旦点击就会绕过检测攻击成功

二，Java
1 函数式编程
概念：
面向对象思想关注用什么对象完成什么事情。而函数式编程思想就类似数学中的函数。它关注的是对数据进行了说明操作。（类似把具体的操作代码通过参数的形式传递进去）

好处：

为了看懂公司大佬写的代码。
大数据量下处理集合效率高
代码可读性高
减少嵌套

Lambda表达式
04.Lambda表达式入门 P5 - 00:11

Lambda是JDK8中的语法糖，它可以对某些匿名内部类的写法进行简化。它是函数式编程思想的一个重要体现。让我们不用关注是什么对象，而是更关注我们对数据进行了什么操作。写Lambda记得看底层源码。

关注的是参数列表和具体的逻辑代码体

省略规则:

参数类型可以省略
方法体中只有一句代码时大括号return和唯一一句代码的分号可以省略
方法只有一个参数时小括号可以省略

Stream流

概述
1.1 为什么学？
能够看懂公司里的代码
大数量下处理集合效率高

代码可读性高

消灭嵌套地狱

//查询未成年作家的评分在70以上的书籍 由于洋流影响所以作家和书籍可能出现重复，需要进行去重
List bookList = new ArrayList<>();
Set uniqueBookValues = new HashSet<>();
Set uniqueAuthorValues = new HashSet<>();
for (Author author : authors) {
if (uniqueAuthorValues.add(author)) {
if (author.getAge() < 18) {
List books = author.getBooks();
for (Book book : books) {
if (book.getScore() > 70) {
if (uniqueBookValues.add(book)) {
bookList.add(book);
}
}
}
}
}
}
System.out.println(bookList);
List collect = authors.stream()
.distinct()
.filter(author -> author.getAge() < 18)
.map(author -> author.getBooks())
.flatMap(Collection::stream)
.filter(book -> book.getScore() > 70)
.distinct()
.collect(Collectors.toList());
System.out.println(collect);
1.2 函数式编程思想
1.2.1 概念

面向对象思想需要关注用什么对象完成什么事情。而函数式编程思想就类似于我们数学中的函数。它主要关注的是对数据进行了什么操作。

1.2.2 优点

代码简洁，开发快速

接近自然语言，易于理解

易于"并发编程"

Lambda表达式
2.1 概述
Lambda是JDK8中一个语法糖。他可以对某些匿名内部类的写法进行简化。它是函数式编程思想的一个重要体现。让我们不用关注是什么对象。而是更关注我们对数据进行了什么操作。
2.2 核心原则
可推导可省略

3 基本格式
(参数列表)->{代码}
例一
我们在创建线程并启动时可以使用匿名内部类的写法：

new Thread(new Runnable() {
@Override
public void run() {
System.out.println(“你知道吗 我比你想象的 更想在你身边”);
}
}).start();
可以使用Lambda的格式对其进行修改。修改后如下：

new Thread(()->{
System.out.println(“你知道吗 我比你想象的 更想在你身边”);
}).start();
例二:

现有方法定义如下，其中IntBinaryOperator是一个接口。先使用匿名内部类的写法调用该方法。

public static int calculateNum(IntBinaryOperator operator){
int a = 10;
int b = 20;
return operator.applyAsInt(a, b);
}

public static void main(String[] args) {
int i = calculateNum(new IntBinaryOperator() {
@Override
public int applyAsInt(int left, int right) {
return left + right;
}
});
System.out.println(i);
}
Lambda写法：

public static void main(String[] args) {
int i = calculateNum((int left, int right)->{
return left + right;
});
System.out.println(i);
}
例三：

现有方法定义如下，其中IntPredicate是一个接口。先使用匿名内部类的写法调用该方法。

public static void printNum(IntPredicate predicate){
int[] arr = {1,2,3,4,5,6,7,8,9,10};
for (int i : arr) {
if(predicate.test(i)){
System.out.println(i);
}
}
}
public static void main(String[] args) {
printNum(new IntPredicate() {
@Override
public boolean test(int value) {
return value%2==0;
}
});
}
Lambda写法：

public static void main(String[] args) {
printNum((int value)-> {
return value%2==0;
});
}
public static void printNum(IntPredicate predicate){
int[] arr = {1,2,3,4,5,6,7,8,9,10};
for (int i : arr) {
if(predicate.test(i)){
System.out.println(i);
}
}
}
例四：

现有方法定义如下，其中Function是一个接口。先使用匿名内部类的写法调用该方法。

public static R typeConver(Function<String,R> function){
String str = “1235”;
R result = function.apply(str);
return result;
}
public static void main(String[] args) {
Integer result = typeConver(new Function<String, Integer>() {
@Override
public Integer apply(String s) {
return Integer.valueOf(s);
}
});
System.out.println(result);
}
Lambda写法：

Integer result = typeConver((String s)->{
return Integer.valueOf(s);
});
System.out.println(result);
例五：

现有方法定义如下，其中IntConsumer是一个接口。先使用匿名内部类的写法调用该方法。

public static void foreachArr(IntConsumer consumer){
int[] arr = {1,2,3,4,5,6,7,8,9,10};
for (int i : arr) {
consumer.accept(i);
}
}
public static void main(String[] args) {
foreachArr(new IntConsumer() {
@Override
public void accept(int value) {
System.out.println(value);
}
});
}
Lambda写法：

public static void main(String[] args) {
foreachArr((int value)->{
System.out.println(value);
});
}
2.4 省略规则
参数类型可以省略

方法体只有一句代码时大括号return和唯一一句代码的分号可以省略

方法只有一个参数时小括号可以省略

以上这些规则都记不住也可以省略不记

Stream流
3.1 概述
Java8的Stream使用的是函数式编程模式，如同它的名字一样，它可以被用来对集合或数组进行链状流式的操作。可以更方便的让我们对集合或数组操作。
3.2 案例数据准备

org.projectlombok
lombok
1.18.16

@Data
@NoArgsConstructor
@AllArgsConstructor
@EqualsAndHashCode//用于后期的去重使用
public class Author {
//id
private Long id;
//姓名
private String name;
//年龄
private Integer age;
//简介
private String intro;
//作品
private List books;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
@EqualsAndHashCode//用于后期的去重使用
public class Book {
//id
private Long id;
//书名
private String name;

//分类
private String category;

//评分
private Integer score;

//简介
private String intro;

}
private static List getAuthors() {
//数据初始化
Author author = new Author(1L,“蒙多”,33,“一个从菜刀中明悟哲理的祖安人”,null);
Author author2 = new Author(2L,“亚拉索”,15,“狂风也追逐不上他的思考速度”,null);
Author author3 = new Author(3L,“易”,14,“是这个世界在限制他的思维”,null);
Author author4 = new Author(3L,“易”,14,“是这个世界在限制他的思维”,null);
//书籍列表
List books1 = new ArrayList<>();
List books2 = new ArrayList<>();
List books3 = new ArrayList<>();

books1.add(new Book(1L,“刀的两侧是光明与黑暗”,“哲学,爱情”,88,“用一把刀划分了爱恨”));
books1.add(new Book(2L,“一个人不能死在同一把刀下”,“个人成长,爱情”,99,“讲述如何从失败中明悟真理”));

books2.add(new Book(3L,“那风吹不到的地方”,“哲学”,85,“带你用思维去领略世界的尽头”));
books2.add(new Book(3L,“那风吹不到的地方”,“哲学”,85,“带你用思维去领略世界的尽头”));
books2.add(new Book(4L,“吹或不吹”,“爱情,个人传记”,56,“一个哲学家的恋爱观注定很难把他所在的时代理解”));

books3.add(new Book(5L,“你的剑就是我的剑”,“爱情”,56,“无法想象一个武者能对他的伴侣这么的宽容”));
books3.add(new Book(6L,“风与剑”,“个人传记”,100,“两个哲学家灵魂和肉体的碰撞会激起怎么样的火花呢？”));
books3.add(new Book(6L,“风与剑”,“个人传记”,100,“两个哲学家灵魂和肉体的碰撞会激起怎么样的火花呢？”));

author.setBooks(books1);
author2.setBooks(books2);
author3.setBooks(books3);
author4.setBooks(books3);

List authorList = new ArrayList<>(Arrays.asList(author,author2,author3,author4));
return authorList;
24
3.3 快速入门
3.3.1 需求

我们可以调用getAuthors方法获取到作家的集合。现在需要打印所有年龄小于18的作家的名字，并且要注意去重。

3.3.2 实现

//打印所有年龄小于18的作家的名字，并且要注意去重
List authors = getAuthors();
authors.
stream()//把集合转换成流
.distinct()//先去除重复的作家
.filter(author -> author.getAge()<18)//筛选年龄小于18的
.forEach(author -> System.out.println(author.getName()));//遍历打印名字
3.4 常用操作
3.4.1 创建流

单列集合： 集合对象.stream()

List authors = getAuthors();
Stream stream = authors.stream();
数组：Arrays.stream(数组)或者使用Stream.of来创建

Integer[] arr = {1,2,3,4,5};
Stream stream = Arrays.stream(arr);
Stream stream2 = Stream.of(arr);
双列集合：转换成单列集合后再创建

Map<String,Integer> map = new HashMap<>();
map.put(“蜡笔小新”,19);
map.put(“黑子”,17);
map.put(“日向翔阳”,16);
Stream<Map.Entry<String, Integer>> stream = map.entrySet().stream();
3.4.2 中间操作

filter
可以对流中的元素进行条件过滤，符合过滤条件的才能继续留在流中。

例如：

打印所有姓名长度大于1的作家的姓名

List authors = getAuthors();
authors.stream()
.filter(author -> author.getName().length()>1)
.forEach(author -> System.out.println(author.getName()));
map
可以把对流中的元素进行计算或转换。

例如：

打印所有作家的姓名

List authors = getAuthors();

authors
.stream()
.map(author -> author.getName())
.forEach(name->System.out.println(name));
// 打印所有作家的姓名
List authors = getAuthors();

authors.stream()
.map(author -> author.getAge())
.map(age->age+10)
.forEach(age-> System.out.println(age));
distinct
可以去除流中的重复元素。

例如：

打印所有作家的姓名，并且要求其中不能有重复元素。

List authors = getAuthors();
authors.stream()
.distinct()
.forEach(author -> System.out.println(author.getName()));
注意：distinct方法是依赖Object的equals方法来判断是否是相同对象的。所以需要注意重写equals方法。

sorted
可以对流中的元素进行排序。

例如：

对流中的元素按照年龄进行降序排序，并且要求不能有重复的元素。

List authors = getAuthors();
// 对流中的元素按照年龄进行降序排序，并且要求不能有重复的元素。
authors.stream()
.distinct()
.sorted()
.forEach(author -> System.out.println(author.getAge()));
List authors = getAuthors();
// 对流中的元素按照年龄进行降序排序，并且要求不能有重复的元素。
authors.stream()
.distinct()
.sorted((o1, o2) -> o2.getAge()-o1.getAge())
.forEach(author -> System.out.println(author.getAge()));
注意：如果调用空参的sorted()方法，需要流中的元素是实现了Comparable。

limit
可以设置流的最大长度，超出的部分将被抛弃。

例如：

对流中的元素按照年龄进行降序排序，并且要求不能有重复的元素,然后打印其中年龄最大的两个作家的姓名。

List authors = getAuthors();
authors.stream()
.distinct()
.sorted()
.limit(2)
.forEach(author -> System.out.println(author.getName()));
skip

跳过流中的前n个元素，返回剩下的元素

例如：

打印除了年龄最大的作家外的其他作家，要求不能有重复元素，并且按照年龄降序排序。

// 打印除了年龄最大的作家外的其他作家，要求不能有重复元素，并且按照年龄降序排序。
List authors = getAuthors();
authors.stream()
.distinct()
.sorted()
.skip(1)
.forEach(author -> System.out.println(author.getName()));
flatMap

map只能把一个对象转换成另一个对象来作为流中的元素。而flatMap可以把一个对象转换成多个对象作为流中的元素。

例一：

打印所有书籍的名字。要求对重复的元素进行去重。

// 打印所有书籍的名字。要求对重复的元素进行去重。
List authors = getAuthors();

authors.stream()
.flatMap(author -> author.getBooks().stream())
.distinct()
.forEach(book -> System.out.println(book.getName()));
例二：

打印现有数据的所有分类。要求对分类进行去重。不能出现这种格式：哲学,爱情

// 打印现有数据的所有分类。要求对分类进行去重。不能出现这种格式：哲学,爱情 爱情
List authors = getAuthors();
authors.stream()
.flatMap(author -> author.getBooks().stream())
.distinct()
.flatMap(book -> Arrays.stream(book.getCategory().split(“,”)))
.distinct()
.forEach(category-> System.out.println(category));
3.4.3 终结操作

forEach
对流中的元素进行遍历操作，我们通过传入的参数去指定对遍历到的元素进行什么具体操作。

例子：

输出所有作家的名字

// 输出所有作家的名字
List authors = getAuthors();

authors.stream()
.map(author -> author.getName())
.distinct()
.forEach(name-> System.out.println(name));

count
可以用来获取当前流中元素的个数。

例子：

打印这些作家的所出书籍的数目，注意删除重复元素。

// 打印这些作家的所出书籍的数目，注意删除重复元素。
List authors = getAuthors();

long count = authors.stream()
.flatMap(author -> author.getBooks().stream())
.distinct()
.count();
System.out.println(count);
max&min

可以用来或者流中的最值。

例子：

分别获取这些作家的所出书籍的最高分和最低分并打印。

// 分别获取这些作家的所出书籍的最高分和最低分并打印。
//Stream -> Stream ->Stream ->求值

List authors = getAuthors();
Optional max = authors.stream()
.flatMap(author -> author.getBooks().stream())
.map(book -> book.getScore())
.max((score1, score2) -> score1 - score2);

Optional min = authors.stream()
.flatMap(author -> author.getBooks().stream())
.map(book -> book.getScore())
.min((score1, score2) -> score1 - score2);
System.out.println(max.get());
System.out.println(min.get());

collect
把当前流转换成一个集合。

例子：

获取一个存放所有作者名字的List集合。

// 获取一个存放所有作者名字的List集合。
List authors = getAuthors();
List nameList = authors.stream()
.map(author -> author.getName())
.collect(Collectors.toList());
System.out.println(nameList);
获取一个所有书名的Set集合。

// 获取一个所有书名的Set集合。
List authors = getAuthors();
Set books = authors.stream()
.flatMap(author -> author.getBooks().stream())
.collect(Collectors.toSet());

System.out.println(books);
获取一个Map集合，map的key为作者名，value为List

// 获取一个Map集合，map的key为作者名，value为List
List authors = getAuthors();

Map<String, List> map = authors.stream()
.distinct()
.collect(Collectors.toMap(author -> author.getName(), author -> author.getBooks()));

System.out.println(map);
查找与匹配

anyMatch
可以用来判断是否有任意符合匹配条件的元素，结果为boolean类型。

例子：

判断是否有年龄在29以上的作家

// 判断是否有年龄在29以上的作家
List authors = getAuthors();
boolean flag = authors.stream()
.anyMatch(author -> author.getAge() > 29);
System.out.println(flag);

allMatch
可以用来判断是否都符合匹配条件，结果为boolean类型。如果都符合结果为true，否则结果为false。

例子：

判断是否所有的作家都是成年人

// 判断是否所有的作家都是成年人
List authors = getAuthors();
boolean flag = authors.stream()
.allMatch(author -> author.getAge() >= 18);
System.out.println(flag);
noneMatch

可以判断流中的元素是否都不符合匹配条件。如果都不符合结果为true，否则结果为false

例子：

判断作家是否都没有超过100岁的。

// 判断作家是否都没有超过100岁的。
List authors = getAuthors();

boolean b = authors.stream()
.noneMatch(author -> author.getAge() > 100);

System.out.println(b);

findAny
获取流中的任意一个元素。该方法没有办法保证获取的一定是流中的第一个元素。

例子：

获取任意一个年龄大于18的作家，如果存在就输出他的名字

// 获取任意一个年龄大于18的作家，如果存在就输出他的名字
List authors = getAuthors();
Optional optionalAuthor = authors.stream()
.filter(author -> author.getAge()>18)
.findAny();

optionalAuthor.ifPresent(author -> System.out.println(author.getName()));

findFirst
获取流中的第一个元素。

例子：

获取一个年龄最小的作家，并输出他的姓名。

// 获取一个年龄最小的作家，并输出他的姓名。
List authors = getAuthors();
Optional first = authors.stream()
.sorted((o1, o2) -> o1.getAge() - o2.getAge())
.findFirst();

first.ifPresent(author -> System.out.println(author.getName()));
reduce归并

对流中的数据按照你指定的计算方式计算出一个结果。（缩减操作）

reduce的作用是把stream中的元素给组合起来，我们可以传入一个初始值，它会按照我们的计算方式依次拿流中的元素和初始化值进行计算，计算结果再和后面的元素计算。

reduce两个参数的重载形式内部的计算方式如下：

T result = identity;
for (T element : this stream)
result = accumulator.apply(result, element)
return result;
其中identity就是我们可以通过方法参数传入的初始值，accumulator的apply具体进行什么计算也是我们通过方法参数来确定的。

例子：

使用reduce求所有作者年龄的和

// 使用reduce求所有作者年龄的和
List authors = getAuthors();
Integer sum = authors.stream()
.distinct()
.map(author -> author.getAge())
.reduce(0, (result, element) -> result + element);
System.out.println(sum);
使用reduce求所有作者中年龄的最大值

// 使用reduce求所有作者中年龄的最大值
List authors = getAuthors();
Integer max = authors.stream()
.map(author -> author.getAge())
.reduce(Integer.MIN_VALUE, (result, element) -> result < element ? element : result);

System.out.println(max);
使用reduce求所有作者中年龄的最小值

// 使用reduce求所有作者中年龄的最小值
List authors = getAuthors();
Integer min = authors.stream()
.map(author -> author.getAge())
.reduce(Integer.MAX_VALUE, (result, element) -> result > element ? element : result);
System.out.println(min);
reduce一个参数的重载形式内部的计算

boolean foundAny = false;
T result = null;
for (T element : this stream) {
if (!foundAny) {
foundAny = true;
result = element;
}
else
result = accumulator.apply(result, element);
}
return foundAny ? Optional.of(result) : Optional.empty();
如果用一个参数的重载方法去求最小值代码如下：

// 使用reduce求所有作者中年龄的最小值
List authors = getAuthors();
Optional minOptional = authors.stream()
.map(author -> author.getAge())
.reduce((result, element) -> result > element ? element : result);
minOptional.ifPresent(age-> System.out.println(age));
3.5 注意事项
惰性求值（如果没有终结操作，没有中间操作是不会得到执行的）

流是一次性的（一旦一个流对象经过一个终结操作后。这个流就不能再被使用）

不会影响原数据（我们在流中可以多数据做很多处理。但是正常情况下是不会影响原来集合中的元素的。这往往也是我们期望的）

Optional
4.1 概述
我们在编写代码的时候出现最多的就是空指针异常。所以在很多情况下我们需要做各种非空的判断。
例如：

Author author = getAuthor();
if(author!=null){
System.out.println(author.getName());
}
尤其是对象中的属性还是一个对象的情况下。这种判断会更多。

而过多的判断语句会让我们的代码显得臃肿不堪。

所以在JDK8中引入了Optional,养成使用Optional的习惯后你可以写出更优雅的代码来避免空指针异常。

并且在很多函数式编程相关的API中也都用到了Optional，如果不会使用Optional也会对函数式编程的学习造成影响。

4.2 使用
4.2.1 创建对象

Optional就好像是包装类，可以把我们的具体数据封装Optional对象内部。然后我们去使用Optional中封装好的方法操作封装进去的数据就可以非常优雅的避免空指针异常。

我们一般使用Optional的静态方法ofNullable来把数据封装成一个Optional对象。无论传入的参数是否为null都不会出现问题。

Author author = getAuthor();
Optional authorOptional = Optional.ofNullable(author);
你可能会觉得还要加一行代码来封装数据比较麻烦。但是如果改造下getAuthor方法，让其的返回值就是封装好的Optional的话，我们在使用时就会方便很多。

而且在实际开发中我们的数据很多是从数据库获取的。Mybatis从3.5版本可以也已经支持Optional了。我们可以直接把dao方法的返回值类型定义成Optional类型，MyBastis会自己把数据封装成Optional对象返回。封装的过程也不需要我们自己操作。

如果你确定一个对象不是空的则可以使用Optional的静态方法of来把数据封装成Optional对象。

Author author = new Author();
Optional authorOptional = Optional.of(author);
但是一定要注意，如果使用of的时候传入的参数必须不为null。（尝试下传入null会出现什么结果）

如果一个方法的返回值类型是Optional类型。而如果我们经判断发现某次计算得到的返回值为null，这个时候就需要把null封装成Optional对象返回。这时则可以使用Optional的静态方法empty来进行封装。

Optional.empty()
所以最后你觉得哪种方式会更方便呢？ofNullable

4.2.2 安全消费值

我们获取到一个Optional对象后肯定需要对其中的数据进行使用。这时候我们可以使用其ifPresent方法对来消费其中的值。

这个方法会判断其内封装的数据是否为空，不为空时才会执行具体的消费代码。这样使用起来就更加安全了。

例如,以下写法就优雅的避免了空指针异常。

Optional authorOptional = Optional.ofNullable(getAuthor());

authorOptional.ifPresent(author -> System.out.println(author.getName()));
4.2.3 获取值

如果我们想获取值自己进行处理可以使用get方法获取，但是不推荐。因为当Optional内部的数据为空的时候会出现异常。

4.2.4 安全获取值

如果我们期望安全的获取值。我们不推荐使用get方法，而是使用Optional提供的以下方法。

orElseGet

获取数据并且设置数据为空时的默认值。如果数据不为空就能获取到该数据。如果为空则根据你传入的参数来创建对象作为默认值返回。

Optional authorOptional = Optional.ofNullable(getAuthor());
Author author1 = authorOptional.orElseGet(() -> new Author());
orElseThrow

获取数据，如果数据不为空就能获取到该数据。如果为空则根据你传入的参数来创建异常抛出。

Optional authorOptional = Optional.ofNullable(getAuthor());
try {
Author author = authorOptional.orElseThrow((Supplier) () -> new RuntimeException(“author为空”));
System.out.println(author.getName());
} catch (Throwable throwable) {
throwable.printStackTrace();
}
4.2.5 过滤

我们可以使用filter方法对数据进行过滤。如果原本是有数据的，但是不符合判断，也会变成一个无数据的Optional对象。

Optional authorOptional = Optional.ofNullable(getAuthor());
authorOptional.filter(author -> author.getAge()>100).ifPresent(author -> System.out.println(author.getName()));
4.2.6 判断

我们可以使用isPresent方法进行是否存在数据的判断。如果为空返回值为false,如果不为空，返回值为true。但是这种方式并不能体现Optional的好处，更推荐使用ifPresent方法。

Optional authorOptional = Optional.ofNullable(getAuthor());

if (authorOptional.isPresent()) {
System.out.println(authorOptional.get().getName());
}
4.2.7 数据转换

Optional还提供了map可以让我们的对数据进行转换，并且转换得到的数据也还是被Optional包装好的，保证了我们的使用安全。

例如我们想获取作家的书籍集合。

private static void testMap() {
Optional authorOptional = getAuthorOptional();
Optional<List> optionalBooks = authorOptional.map(author -> author.getBooks());
optionalBooks.ifPresent(books -> System.out.println(books));
}
函数式接口
5.1 概述

只有一个抽象方法的接口我们称之为函数接口。
JDK的函数式接口都加上了@FunctionalInterface 注解进行标识。但是无论是否加上该注解只要接口中只有一个抽象方法，都是函数式接口。

5.2 常见函数式接口
Consumer 消费接口

根据其中抽象方法的参数列表和返回值类型知道，我们可以在方法中对传入的参数进行消费。

Function 计算转换接口

根据其中抽象方法的参数列表和返回值类型知道，我们可以在方法中对传入的参数计算或转换，把结果返回

Predicate 判断接口

根据其中抽象方法的参数列表和返回值类型知道，我们可以在方法中对传入的参数条件判断，返回判断结果

Supplier 生产型接口

根据其中抽象方法的参数列表和返回值类型知道，我们可以在方法中创建对象，把创建好的对象返回

5.3 常用的默认方法

and
我们在使用Predicate接口时候可能需要进行判断条件的拼接。而and方法相当于是使用&&来拼接两个判断条件

例如：

打印作家中年龄大于17并且姓名的长度大于1的作家。

List authors = getAuthors();
Stream authorStream = authors.stream();
authorStream.filter(new Predicate() {
@Override
public boolean test(Author author) {
return author.getAge()>17;
}
}.and(new Predicate() {
@Override
public boolean test(Author author) {
return author.getName().length()>1;
}
})).forEach(author -> System.out.println(author));
or

我们在使用Predicate接口时候可能需要进行判断条件的拼接。而or方法相当于是使用||来拼接两个判断条件。

例如：

打印作家中年龄大于17或者姓名的长度小于2的作家。

// 打印作家中年龄大于17或者姓名的长度小于2的作家。
List authors = getAuthors();
authors.stream()
.filter(new Predicate() {
@Override
public boolean test(Author author) {
return author.getAge()>17;
}
}.or(new Predicate() {
@Override
public boolean test(Author author) {
return author.getName().length()<2;
}
})).forEach(author -> System.out.println(author.getName()));

negate
Predicate接口中的方法。negate方法相当于是在判断添加前面加了个! 表示取反

例如：

打印作家中年龄不大于17的作家。

// 打印作家中年龄不大于17的作家。
List authors = getAuthors();
authors.stream()
.filter(new Predicate() {
@Override
public boolean test(Author author) {
return author.getAge()>17;
}
}.negate()).forEach(author -> System.out.println(author.getAge()));

方法引用
我们在使用lambda时，如果方法体中只有一个方法的调用的话（包括构造方法）,我们可以用方法引用进一步简化代码。

6.1 推荐用法
我们在使用lambda时不需要考虑什么时候用方法引用，用哪种方法引用，方法引用的格式是什么。我们只需要在写完lambda方法发现方法体只有一行代码，并且是方法的调用时使用快捷键尝试是否能够转换成方法引用即可。

当我们方法引用使用的多了慢慢的也可以直接写出方法引用。

6.2 基本格式
类名或者对象名::方法名

6.3 语法详解(了解)
6.3.1 引用类的静态方法

其实就是引用类的静态方法

格式

类名::方法名
使用前提

如果我们在重写方法的时候，方法体中只有一行代码，并且这行代码是调用了某个类的静态方法，并且我们把要重写的抽象方法中所有的参数都按照顺序传入了这个静态方法中，这个时候我们就可以引用类的静态方法。

例如：

如下代码就可以用方法引用进行简化

List authors = getAuthors();

Stream authorStream = authors.stream();

authorStream.map(author -> author.getAge())
.map(age->String.valueOf(age));
注意，如果我们所重写的方法是没有参数的，调用的方法也是没有参数的也相当于符合以上规则。

优化后如下：

List authors = getAuthors();

Stream authorStream = authors.stream();

authorStream.map(author -> author.getAge())
.map(String::valueOf);
6.3.2 引用对象的实例方法

格式

对象名::方法名
使用前提

如果我们在重写方法的时候，方法体中只有一行代码，并且这行代码是调用了某个对象的成员方法，并且我们把要重写的抽象方法中所有的参数都按照顺序传入了这个成员方法中，这个时候我们就可以引用对象的实例方法

例如：

List authors = getAuthors();

Stream authorStream = authors.stream();
StringBuilder sb = new StringBuilder();
authorStream.map(author -> author.getName())
.forEach(name->sb.append(name));
优化后：

List authors = getAuthors();

Stream authorStream = authors.stream();
StringBuilder sb = new StringBuilder();
authorStream.map(author -> author.getName())
.forEach(sb::append);
6.3.4 引用类的实例方法

格式

类名::方法名
使用前提

如果我们在重写方法的时候，方法体中只有一行代码，并且这行代码是调用了第一个参数的成员方法，并且我们把要重写的抽象方法中剩余的所有的参数都按照顺序传入了这个成员方法中，这个时候我们就可以引用类的实例方法。

例如：

interface UseString{
String use(String str,int start,int length);
}

public static String subAuthorName(String str, UseString useString){
int start = 0;
int length = 1;
return useString.use(str,start,length);
}
public static void main(String[] args) {

subAuthorName(“三更草堂”, new UseString() {
@Override
public String use(String str, int start, int length) {
return str.substring(start,length);
}
});
}
优化后如下：

public static void main(String[] args) {
subAuthorName(“三更草堂”, String::substring);
}
6.3.5 构造器引用

如果方法体中的一行代码是构造器的话就可以使用构造器引用。

格式

类名::new
使用前提

如果我们在重写方法的时候，方法体中只有一行代码，并且这行代码是调用了某个类的构造方法，并且我们把要重写的抽象方法中的所有的参数都按照顺序传入了这个构造方法中，这个时候我们就可以引用构造器。

例如：

List authors = getAuthors();
authors.stream()
.map(author -> author.getName())
.map(name->new StringBuilder(name))
.map(sb->sb.append(“-三更”).toString())
.forEach(str-> System.out.println(str));
优化后：

List authors = getAuthors();
authors.stream()
.map(author -> author.getName())
.map(StringBuilder::new)
.map(sb->sb.append(“-三更”).toString())
.forEach(str-> System.out.println(str));
高级用法
基本数据类型优化
我们之前用到的很多Stream的方法由于都使用了泛型。所以涉及到的参数和返回值都是引用数据类型。
即使我们操作的是整数小数，但是实际用的都是他们的包装类。JDK5中引入的自动装箱和自动拆箱让我们在使用对应的包装类时就好像使用基本数据类型一样方便。但是你一定要知道装箱和拆箱肯定是要消耗时间的。虽然这个时间消耗很下。但是在大量的数据不断的重复装箱拆箱的时候，你就不能无视这个时间损耗了。

所以为了让我们能够对这部分的时间消耗进行优化。Stream还提供了很多专门针对基本数据类型的方法。

例如：mapToInt,mapToLong,mapToDouble,flatMapToInt,flatMapToDouble等。

private static void test27() {
List authors = getAuthors();
authors.stream()
.map(author -> author.getAge())
.map(age -> age + 10)
.filter(age->age>18)
.map(age->age+2)
.forEach(System.out::println);

authors.stream()
.mapToInt(author -> author.getAge())
.map(age -> age + 10)
.filter(age->age>18)
.map(age->age+2)
.forEach(System.out::println);
}
并行流
当流中有大量元素时，我们可以使用并行流去提高操作的效率。其实并行流就是把任务分配给多个线程去完全。如果我们自己去用代码实现的话其实会非常的复杂，并且要求你对并发编程有足够的理解和认识。而如果我们使用Stream的话，我们只需要修改一个方法的调用就可以使用并行流来帮我们实现，从而提高效率。

parallel方法可以把串行流转换成并行流。

private static void test28() {
Stream stream = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Integer sum = stream.parallel()
.peek(new Consumer() {
@Override
public void accept(Integer num) {
System.out.println(num+Thread.currentThread().getName());
}
})
.filter(num -> num > 5)
.reduce((result, ele) -> result + ele)
.get();
System.out.println(sum);
}
也可以通过parallelStream直接获取并行流对象。

List authors = getAuthors();
authors.parallelStream()
.map(author -> author.getAge())
.map(age -> age + 10)
.filter(age->age>18)
.map(age->age+2)
.forEach(System.out::println);
2 并发编程
并发与并行

并发与并行的区别是什么
并行在多处理器系统中存在，而并发可以在单处理器和多处理器系统中都存在，并发能够在单处理器系统中存在是因为并发是并行的假象
如果系统只有一个 CPU，则它根本不可能真正同时进行一个以上的线程，它只能把 CPU 运行时间划分成若干个时间段，再将时间段分配给各个线程执行，在一个时间段的线程代码运行时,其它线程处于挂起状态.这种方式我们称之为并发（Concurrent）。
当系统有一个以上 CPU 时，则线程的操作有可能非并发。当一个 CPU 执行一个线程时，另一个 CPU 可以执行另一个线程，两个线程互不抢占 CPU 资源，可以同时进行，这种方式我们称之为并行（Parallel）。

线程
线程的状态
线程是一个动态执行的过程，它有一个从产生到死亡的过程，共五种状态：新建（newThread）、就绪（runnable）、运行（running）、死亡（dead）、堵塞（blocked）
img

创建线程的多种方式
继承Thread类实现多线程
覆写Runnable()接口实现多线程，而后同样覆写run()。
覆写Callable接口实现多线程（JDK1.5）：核心方法叫call()方法，有返回值

通过线程池启动多线程
FixThreadPool(int n); 固定大小的线程池
SingleThreadPoolExecutor :单线程池
CachedThreadPool(); 缓存线程池

继承Thread和实现Runnable接口的区别：
a.实现Runnable接口避免多继承局限；
b.实现Runable接口可以实现资源共享

什么是守护线程？
只要当前JVM实例中尚存在任何一个非守护线程没有结束，守护线程就全部工作；只有当最后一个非守护线程结束时，守护线程随着JVM一同结束工作。

Daemon的作用是为其他线程的运行提供便利服务，守护线程最典型的应用就是 GC (垃圾回收器)
Thread daemonTread = new Thread();
daemonThread.setDaemon(true);
线程和进程的区别是什么？
根本区别：进程是操作系统资源分配的基本单位，而线程是任务调度和执行的基本单位
开销方面：每个进程都有独立的代码和数据空间（程序上下文），程序之间的切换会有较大的开销；线程可以看做轻量级的进程，同一类线程共享代码和数据空间，每个线程都有自己独立的运行栈和程序计数器（PC），线程之间切换的开销小。
所处环境：在操作系统中能同时运行多个进程（程序）；而在同一个进程（程序）中有多个线程同时执行（通过CPU调度，在每个时间片中只有一个线程执行）
内存分配：系统在运行的时候会为每个进程分配不同的内存空间；而对线程而言，除了CPU外，系统不会为线程分配内存（线程所使用的资源来自其所属进程的资源），线程组之间只能共享资源。
包含关系：没有线程的进程可以看做是单线程的，如果一个进程内有多个线程，则执行过程不是一条线的，而是多条线（线程）共同完成的；线程是进程的一部分，所以线程也被称为轻权进程或者轻量级进程。
线程、进程、协程的区别是什么？
进程拥有自己独立的堆和栈，既不共享堆，亦不共享栈，进程由操作系统调度。

线程拥有自己独立的栈和共享的堆，共享堆，不共享栈，线程亦由操作系统调度(标准线程是这样的)。

协程和线程一样共享堆，不共享栈，协程由程序员在协程的代码里显示调度。

一个应用程序一般对应一个进程，一个进程一般有一个主线程，还有若干个辅助线程，线程之间是平行运行的，在线程里面可以开启协程，让程序在特定的时间内运行。

协程和线程的区别是：协程避免了无意义的调度，由此可以提高性能，但也因此，程序员必须自己承担调度的责任，同时，协程也失去了标准线程使用多CPU的能力。

线程池
线程池有什么作用？
提高效率：创建好一定数量的线程放在池中，等需要使用的时候就从池中拿一个，这要比需要的时候创建一个线程对象要快的多。
方便管理：可以编写线程池管理代码对池中的线程同一进行管理，比如说启动时有该程序创建100个线程，每当有请求的时候，就分配一个线程去工作，如果刚好并发有101个请求，那多出的这一个请求可以排队等候，避免因无休止的创建线程导致系统崩溃。
说说几种常见的线程池及使用场景

1、newSingleThreadExecutor
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue()));
}

2、newFixedThreadPool
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue());
}

3、newCachedThreadPool
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
}

4、newScheduledThreadPool
创建一个定长线程池，支持定时及周期性任务执行。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}

线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。 说明：Executors各个方法的弊端：
1）newFixedThreadPool和newSingleThreadExecutor:
  主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。
2）newCachedThreadPool和newScheduledThreadPool:
  主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。

线程池都有哪几种工作队列
ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。
LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列
SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

线程池中的几种重要的参数及流程说明
corePoolSize：核心池的大小，在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
maximumPoolSize：线程池最大线程数，它表示在线程池中最多能创建多少个线程；
keepAliveTime：表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0；
unit：参数keepAliveTime的时间单位
workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数有以下几种选择：
ArrayBlockingQueue
LinkedBlockingQueue
SynchronousQueue
PriorityBlockingQueue
hreadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程做些更有意义的事情，比如设置daemon和优先级等
handler：表示当拒绝处理任务时的策略，有以下四种取值：
1、AbortPolicy：直接抛出异常。
2、CallerRunsPolicy：只用调用者所在线程来运行任务。
3、DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
4、DiscardPolicy：不处理，丢弃掉。
5、根据应用场景需要实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务
怎么理解无界队列和有界队列

有界队列
初始的poolSize < corePoolSize，提交的runnable任务，会直接做为new一个Thread的参数，立马执行。
当提交的任务数超过了corePoolSize，会将当前的runable提交到一个block queue中。
有界队列满了之后，如果poolSize < maximumPoolsize时，会尝试new 一个Thread的进行救急处理，立马执行对应的runnable任务。
如果3中也无法处理了，就会走到第四步执行reject操作。

无界队列
与有界队列相比，除非系统资源耗尽，否则无界的任务队列不存在任务入队失败的情况。
当有新的任务到来，系统的线程数小于corePoolSize时，则新建线程执行任务。当达到corePoolSize后，就不会继续增加，若后续仍有新的任务加入，而没有空闲的线程资源，则任务直接进入队列等待。
若任务创建和处理的速度差异很大，无界队列会保持快速增长，直到耗尽系统内存。
当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略。
如何合理配置线程池的大小
大家认为线程池的大小经验值应该这样设置：（其中N为CPU的个数）
如果是CPU密集型应用，则线程池大小设置为N+1
如果是IO密集型应用，则线程池大小设置为2N+1
线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程。

计算密集型任务的特点是要进行大量的计算，消耗CPU资源，比如计算圆周率、对视频进行高清解码等等，全靠CPU的运算能力。这种计算密集型任务虽然也可以用多任务完成，但是任务越多，花在任务切换的时间就越多，CPU执行任务的效率就越低，所以，要最高效地利用CPU，计算密集型任务同时进行的数量应当等于CPU的核心数。

第二种任务的类型是IO密集型，涉及到网络、磁盘IO的任务都是IO密集型任务，这类任务的特点是CPU消耗很少，任务的大部分时间都在等待IO操作完成（因为IO的速度远远低于CPU和内存的速度）。对于IO密集型任务，任务越多，CPU效率越高，但也有一个限度。常见的大部分任务都是IO密集型任务，比如Web应用。

submit()和execute()
JDK5往后，任务分两类：一类是实现了Runnable接口的类，一类是实现了Callable接口的类。两者都可以被ExecutorService执行，它们的区别是：

execute(Runnable x) 没有返回值。可以执行任务，但无法判断任务是否成功完成。——实现Runnable接口
submit(Runnable x) 返回一个future。可以用这个future来判断任务是否成功完成。——实现Callable接口
线程池原理
线程池提供了两个钩子(beforeExecute，afterExecute)给我们，我们继承线程池，在执行任务前后做一些事情。

线程池由两个核心数据结构组成：
1）线程集合（workers）：存放执行任务的线程，是一个HashSet；
2）任务等待队列（workQueue）：存放等待线程池调度执行的任务，是一个阻塞式队列BlockingQueue；

任务执行流程
1）线程池中线程数量小于corePoolSize，此时任务不会进等待队列，线程池直接创建一个线程Worker执行提交的任务；
2）线程池中线程数量不小于corePoolSize并且等待队列未满，任务直接添加到等待队列，等待线程池调度执行；
3）线程池中线程数量不小于corePoolSize但是等待队列已满且线程数量小于maximumPoolSize，线程池会进行扩容新创建一个线程Worker执行提交的任务，新创建的Worker会被添加到线程集合workers中；
4）等待队列已满并且线程数量已达到maximumPoolSize，这种情况下线程池无法继续执行任务会拒绝任务，执行一个指定的拒接策略。

JDK内置的拒绝策略主要有下面几种：
1）调用线程执行（CallerRunsPolicy）：任务被线程池拒绝后，任务会被调用线程执行；
2）终止执行（AbortPolicy）：任务被拒绝时，抛出RejectedExecutionException异常报错
3）丢弃任务（DiscardPolicy）：任务被直接丢弃，不会抛异常报错；
4）丢失老任务（DiscardOldestPolicy）：把等待队列中最老的任务删除，删除后重新提交当前任务。

线程安全

死锁产生的4个必要条件？
互斥条件：进程要求对所分配的资源进行排它性控制，即在一段时间内某资源仅为一进程所占用。
请求和保持条件：当进程因请求资源而阻塞时，对已获得的资源保持不放。
不剥夺条件：进程已获得的资源在未使用完之前，不能剥夺，只能在使用完时由自己释放。
环路等待条件：在发生死锁时，必然存在一个进程–资源的环形链。
怎么预防死锁？
资源一次性分配：一次性分配所有资源，这样就不会再有请求了：（破坏请求条件）
只要有一个资源得不到分配，也不给这个进程分配其他的资源：（破坏请保持条件）
可剥夺资源：即当某进程获得了部分资源，但得不到其它资源，则释放已占有的资源（破坏不可剥夺条件）
资源有序分配法：系统给每类资源赋予一个编号，每一个进程按编号递增的顺序请求资源，释放则相反（破坏环路等待条件）
总结：
1、以确定的顺序获得锁
2、超时放弃
img

Java各种锁
CAS
CAS操作的就是乐观锁，每次不加锁而是假设没有冲突而去完成某项操作，如果因为冲突失败就重试，直到成功为止。
CAS是英文单词Compare And Swap的缩写，翻译过来就是比较并替换。
CAS机制当中使用了3个基本操作数：内存地址V，旧的预期值A，要修改的新值B。更新一个变量的时候，只有当变量的预期值A和内存地址V当中的实际值相同时，才会将内存地址V对应的值修改为B。
CAS的缺点：

CPU开销较大：在并发量比较高的情况下，如果许多线程反复尝试更新某一个变量，却又一直更新不成功，循环往复，会给CPU带来很大的压力。
不能保证代码块的原子性：CAS机制所保证的只是一个变量的原子性操作，而不能保证整个代码块的原子性。比如需要保证3个变量共同进行原子性的更新，就不得不使用Synchronized了。
ABA问题：

CAS可以有效的提升并发的效率，但同时也会引入ABA问题。
比如线程1从内存X中取出A，这时候另一个线程2也从内存X中取出A，并且线程2进行了一些操作将内存X中的值变成了B，然后线程2又将内存X中的数据变成A，这时候线程1进行CAS操作发现内存X中仍然是A，然后线程1操作成功。虽然线程1的CAS操作成功，但是整个过程就是有问题的。
比如链表的头在变化了两次后恢复了原值，但是不代表链表就没有变化。
所以JAVA中提供了AtomicStampedReference/AtomicMarkableReference来处理会发生ABA问题的场景，主要是在对象中额外再增加一个标记来标识对象是否有过变更。
乐观锁常见的实现方式
版本号机制：一般是在数据表中加上一个数据版本号version字段，表示数据被修改的次数，当数据被修改时，version值会加一。当线程A要更新数据值时，在读取数据的同时也会读取version值，在提交更新时，若刚才读取到的version值为当前数据库中的version值相等时才更新，否则重试更新操作，直到更新成功。
CAS算法：即compare and swap（比较与交换），是一种有名的无锁算法。无锁编程，即不使用锁的情况下实现多线程之间的变量同步，也就是在没有线程被阻塞的情况下实现变量的同步，所以也叫非阻塞同步（Non-blocking Synchronization）。CAS算法涉及到三个操作数：需要读写的内存值 V；进行比较的值 A；拟写入的新值 B。当且仅当 V 的值等于 A时，CAS通过原子方式用新值B来更新V的值，否则不会执行任何操作（比较和替换是一个原子操作）。一般情况下是一个自旋操作，即不断的重试。
简单的来说CAS适用于写比较少的情况下（多读场景，冲突一般较少），synchronized适用于写比较多的情况下（多写场景，冲突一般较多）

自旋锁和适应性自旋锁
Why：切换线程的开销大，在许多场景中，同步资源的锁定时间很短，为了这一小段时间去切换线程，线程挂起和恢复现场的花费可能会让系统得不偿失。如果物理机器有多个处理器，能够让两个或以上的线程同时并行执行，我们就可以让后面那个请求锁的线程不放弃CPU的执行时间，看看持有锁的线程是否很快就会释放锁。
What：为了让当前线程“稍等一下”，我们需让当前线程进行自旋，如果在自旋完成后前面锁定同步资源的线程已经释放了锁，那么当前线程就可以不必阻塞而是直接获取同步资源，从而避免切换线程的开销。这就是自旋锁。

How：

当前线程竞争锁失败时，打算阻塞自己
不直接阻塞自己，而是自旋一会（空等待，比如一个空的有限for循环
在自旋的同时重新竞争锁
如果自旋结束前获得了锁，那么锁获取成功；否则，自旋结束后阻塞自己
如果在自旋的时间内，锁就被旧owner释放了，那么当前线程就不需要阻塞自己（也不需要在未来锁释放时恢复），减少了一次线程切换。
自旋锁的缺点：自旋等待虽然避免了线程切换的开销，但它要占用处理器时间。如果锁被占用的时间很短，自旋等待的效果就会非常好。反之，如果锁被占用的时间很长，那么自旋的线程只会白浪费处理器资源。所以，自旋等待的时间必须要有一定的限度，如果自旋超过了限定次数（默认是10次，可以使用-XX:PreBlockSpin来更改）没有成功获得锁，就应当挂起线程。
实现原理：自旋锁的实现原理同样也是CAS，AtomicInteger中调用unsafe进行自增操作的源码中的do-while循环就是一个自旋操作，如果修改数值失败则通过循环来执行自旋，直至修改成功。

适应性自旋锁：
自旋锁在JDK1.4.2中引入，使用-XX:+UseSpinning来开启。JDK6中变为默认开启，并且引入了自适应的自旋锁（适应性自旋锁）。
自适应意味着自旋的时间（次数）不再固定，而是由前一次在同一个锁上的自旋时间及锁的拥有者的状态来决定。
如果在同一个锁对象上，自旋等待刚刚成功获得过锁，并且持有锁的线程正在运行中，那么虚拟机就会认为这次自旋也是很有可能再次成功，进而它将允许自旋等待持续相对更长的时间。
如果对于某个锁，自旋很少成功获得过，那在以后尝试获取这个锁时将可能省略掉自旋过程，直接阻塞线程，避免浪费处理器资源。
无锁 VS 偏向锁 VS 轻量级锁 VS 重量级锁
这四种锁是指锁的状态，专门针对synchronized的。偏斜锁、轻量级锁、重量级锁的代码实现，并不在核心类库部分，而是在JVM的代码中。
synchronized是悲观锁，在操作同步资源之前需要给同步资源先加锁，这把锁就是存在Java对象头里的，以Hotspot虚拟机为例，对象头主要包括两部分数据：Mark Word（标记字段）、Klass Pointer（类型指针）。
锁一共有4种状态，级别从低到高依次是：无锁、偏向锁、轻量级锁和重量级锁。锁状态只能升级不能降级。
四种锁状态对应的的Mark Word内容：
img

无锁:
无锁没有对资源进行锁定，所有的线程都能访问并修改同一个资源，但同时只有一个线程能修改成功。CAS原理及应用即是无锁的实现。

偏向锁:
偏向锁是指一段同步代码一直被一个线程所访问，那么该线程会自动获取锁，降低获取锁的代价。
当一个线程访问同步代码块并获取锁时，会在Mark Word里存储锁偏向的线程ID。在线程进入和退出同步块时不再通过CAS操作来加锁和解锁，而是检测Mark Word里是否存储着指向当前线程的偏向锁。
偏向锁只有遇到其他线程尝试竞争偏向锁时，持有偏向锁的线程才会释放锁，线程不会主动释放偏向锁。偏向锁的撤销，需要等待全局安全点（在这个时间点上没有字节码正在执行），它会首先暂停拥有偏向锁的线程，判断锁对象是否处于被锁定状态。撤销偏向锁后恢复到无锁（标志位为“01”）或轻量级锁（标志位为“00”）的状态。
偏向锁在JDK 6及以后的JVM里是默认启用的。可以通过JVM参数关闭偏向锁：-XX:-UseBiasedLocking=false，关闭之后程序默认会进入轻量级锁状态。

轻量级锁:
是指当锁是偏向锁的时候，被另外的线程所访问，偏向锁就会升级为轻量级锁，其他线程会通过自旋的形式尝试获取锁，不会阻塞，从而提高性能。
在代码进入同步块的时候，如果同步对象锁状态为无锁状态（锁标志位为“01”状态，是否为偏向锁为“0”），虚拟机首先将在当前线程的栈帧中建立一个名为锁记录（Lock Record）的空间，用于存储锁对象目前的Mark Word的拷贝，然后拷贝对象头中的Mark Word复制到锁记录中。拷贝成功后，虚拟机将使用CAS操作尝试将对象的Mark Word更新为指向Lock Record的指针，并将Lock Record里的owner指针指向对象的Mark Word。如果这个更新动作成功了，那么这个线程就拥有了该对象的锁，并且对象Mark Word的锁标志位设置为“00”，表示此对象处于轻量级锁定状态。
如果轻量级锁的更新操作失败了，虚拟机首先会检查对象的Mark Word是否指向当前线程的栈帧，如果是就说明当前线程已经拥有了这个对象的锁，那就可以直接进入同步块继续执行，否则说明多个线程竞争锁。
若当前只有一个等待线程，则该线程通过自旋进行等待。但是当自旋超过一定的次数，或者一个线程在持有锁，一个在自旋，又有第三个来访时，轻量级锁升级为重量级锁。

重量级锁
升级为重量级锁时，锁标志的状态值变为“10”，此时Mark Word中存储的是指向重量级锁的指针，此时等待锁的线程都会进入阻塞状态。

公平锁和非公平锁
公平锁是指多个线程按照申请锁的顺序来获取锁，线程直接进入队列中排队，队列中的第一个线程才能获得锁。公平锁的优点是等待锁的线程不会饿死。缺点是整体吞吐效率相对非公平锁要低，等待队列中除第一个线程以外的所有线程都会阻塞，CPU唤醒阻塞线程的开销比非公平锁大。
非公平锁是多个线程加锁时直接尝试获取锁，获取不到才会到等待队列的队尾等待。但如果此时锁刚好可用，那么这个线程可以无需阻塞直接获取到锁，所以非公平锁有可能出现后申请锁的线程先获取锁的场景。非公平锁的优点是可以减少唤起线程的开销，整体的吞吐效率高，因为线程有几率不阻塞直接获得锁，CPU不必唤醒所有线程。缺点是处于等待队列中的线程可能会饿死，或者等很久才会获得锁。
ReentrantLock里面有一个内部类Sync，Sync继承AQS（AbstractQueuedSynchronizer），添加锁和释放锁的大部分操作实际上都是在Sync中实现的。它有公平锁FairSync和非公平锁NonfairSync两个子类。ReentrantLock默认使用非公平锁，也可以通过构造器来显示的指定使用公平锁。
可重入锁和非可重入锁
可重入锁又名递归锁，是指在同一个线程在外层方法获取锁的时候，再进入该线程的内层方法会自动获取锁（前提锁对象得是同一个对象或者class），不会因为之前已经获取过还没释放而阻塞。
Java中ReentrantLock和synchronized都是可重入锁，可重入锁的一个优点是可一定程度避免死锁。NonReentrantLock是非可重入锁。

public class Widget {
public synchronized void doSomething() {
System.out.println(“方法1执行…”);
doOthers();
}

public synchronized void doOthers() {
System.out.println("方法2执行...");
}

}
类中的两个方法都是被内置锁synchronized修饰的，doSomething()方法中调用doOthers()方法。因为内置锁是可重入的，所以同一个线程在调用doOthers()时可以直接获得当前对象的锁，进入doOthers()进行操作。
如果是一个不可重入锁，那么当前线程在调用doOthers()之前需要将执行doSomething()时获取当前对象的锁释放掉，实际上该对象锁已被当前线程所持有，且无法释放。所以此时会出现死锁。

独享锁和共享锁
独享锁也叫排他锁，是指该锁一次只能被一个线程所持有。如果线程T对数据A加上排它锁后，则其他线程不能再对A加任何类型的锁。获得排它锁的线程即能读数据又能修改数据。JDK中的synchronized和JUC中Lock的实现类就是互斥锁。
共享锁是指该锁可被多个线程所持有。如果线程T对数据A加上共享锁后，则其他线程只能对A再加共享锁，不能加排它锁。获得共享锁的线程只能读数据，不能修改数据。
独享锁与共享锁都是通过AQS来实现的，通过实现不同的方法，来实现独享或者共享。

JMM
什么是JMM
JMM并不像JVM内存结构一样是真实存在的。他只是一个抽象的概念。
屏蔽了各种硬件和操作系统的访问差异的，保证了Java程序在各种平台下对内存的访问都能得到一致效果的机制及规范。
目的是解决由于多线程通过共享内存进行通信时，存在的原子性、可见性（缓存一致性）以及有序性问题。
Java内存模型规定所有的变量都是存在主存当中（类似于前面说的物理内存），每个线程都有自己的工作内存（类似于前面的高速缓存）。线程对变量的所有操作都必须在工作内存中进行，而不能直接对主存进行操作。并且每个线程不能访问其他线程的工作内存。
原子性、可见性、有序性
原子性：即一个操作或者多个操作 要么全部执行并且执行的过程不会被任何因素打断，要么就都不执行。基本数据类型的访问都具备原子性，例外就是 long 和 double，虚拟机将没有被 volatile 修饰的 64 位
数据操作划分为两次 32 位操作。
可见性：是指当多个线程访问同一个变量时，一个线程修改了这个变量的值，其他线程能够立即看得到修改的值。除了 volatile 外，synchronized 和 final 也可以保证可见性。
有序性：即程序执行的顺序按照代码的先后顺序执行。Java 提供 volatile 和 synchronized 保证有序性

as-if-serial
as-if-serial语义的意思是：不管怎么重排序，单线程程序的执行结果不能被改变。编译器、runtime和处理器都必须遵守as-if-serial语义。所以编译器和处理器不会对存在数据依赖关系的操作做重排序，因为这种重排序会改变执行结果。但是，如果操作之间不存在数据依赖关系，这些操作就可能被编译器和处理器重排序。

happens-before Why？
因为jvm会对代码进行编译优化，指令会出现重排序的情况，为了避免编译优化对并发编程安全性的影响，需要happens-before规则定义一些禁止编译优化的场景，保证并发编程的正确性。
在程序运行过程中，所有的变更会先在寄存器或本地cache中完成，然后才会被拷贝到主存以跨越内存栅栏（本地或工作内存到主存之间的拷贝动作），此种跨越序列或顺序称为happens-before。happens-before本质是顺序，重点是跨越内存栅栏。
What？
happens-before 在Java内存模型中意味着：前一个操作的结果可以被后续的操作获取。例如前面一个操作把变量a赋值为1，那后面的操作肯定能知道a已经变成了1。

程序次序规则：一个线程内写在前面的操作先行发生于后面的。
管程锁定规则： unlock 操作先行发生于后面对同一个锁的 lock 操作。
volatile 规则：对 volatile 变量的写操作先行发生于后面的读操作。
线程启动规则：线程的 start 方法先行发生于线程的每个动作。
线程终止规则：线程中所有操作先行发生于对线程的终止检测。
对象终结规则：对象的初始化先行发生于 finalize 方法。
as-if-serial 和 happens-before 的区别
as-if-serial 保证单线程程序的执行结果不变，happens-before 保证正确同步的多线程程序的执行结果不变。
这两种语义的目的都是为了在不改变程序执行结果的前提下尽可能提高程序执行并行度。

指令重排序
为了提高性能，编译器和处理器通常会对指令进行重排序，重排序指从源代码到指令序列的重排序，分为三种：
① 编译器优化的重排序，编译器在不改变单线程程序语义的前提下可以重排语句的执行顺序。
② 指令级并行的重排序，如果不存在数据依赖性，处理器可以改变语句对应机器指令的执行顺序。
③内存系统的重排序。

volatile
三个特性：

可见性：即当一个线程修改了声明为volatile变量的值，新值对于其他要读该变量的线程来说是立即可见的。而普通变量是不能做到这一点的，普通变量的值在线程间传递需要通过主内存来完成。
有序性：volatile变量的所谓有序性也就是被声明为volatile的变量的临界区代码的执行是有顺序的，即禁止指令重排序。
受限原子性：这里volatile变量的原子性与synchronized的原子性是不同的，synchronized的原子性是指只要声明为synchronized的方法或代码块儿在执行上就是原子操作的。而volatile是不修饰方法或代码块儿的，它用来修饰变量，对于单个volatile变量的读/写操作都具有原子性，但类似于volatile++这种复合操作不具有原子性。所以volatile的原子性是受限制的。并且在多线程环境中，volatile并不能保证原子性。
volatile写-读的内存语义：volatile写的内存语义：当写线程写一个volatile变量时，JMM会把该线程对应的本地内存中的共享变量值刷新到主内存。
volatile读的内存语义：当读线程读一个volatile变量时，JMM会把该线程对应的本地内存置为无效，线程接下来将从主内存读取共享变量。

实现原理：
volatile可见性的实现就是借助了CPU的lock指令，通过在写volatile的机器指令前加上lock前缀，使写volatile具有以下两个原则：
写volatile时处理器会将缓存写回到主内存。
一个处理器的缓存写回到内存会导致其他处理器的缓存失效。
volatile有序性的保证就是通过禁止指令重排序来实现的。
synchronized
synchronized是如何实现的
synchronized是由一对儿monitorentry/monitorexit指令实现的，Monitor对象是同步的基本实现单元。
在Java 6之前，Monitor的实现完全是依靠操作系统内部的互斥锁，因为需要进行用户态到内核态的切换，所以同步操作是一个无差别的重量级操作。
现代的（Oracle）JDK中，JVM对此进行了大刀阔斧地改进，提供了三种不同的Monitor实现，也就是常说的三种不同的锁：偏斜锁（Biased Locking）、轻量级锁和重量级锁，大大改进了其性能。
synchronized和lock的区别
来源： lock是一个接口，而synchronized是java的一个关键字，synchronized是内置的语言实现；
异常是否释放锁： synchronized在发生异常时候会自动释放占有的锁，因此不会出现死锁；而lock发生异常时候，不会主动释放占有的锁，必须手动unlock来释放锁，可能引起死锁的发生。（所以最好将同步代码块用try catch包起来，finally中写入unlock，避免死锁的发生。）
是否响应中断：lock等待锁过程中可以用interrupt来中断等待，而synchronized只能等待锁的释放，不能响应中断；
是否知道获取锁：Lock可以通过trylock来知道有没有获取锁，而synchronized不能；
Lock可以提高多个线程进行读操作的效率。（可以通过readwritelock实现读写分离）在性能上来说，如果竞争资源不激烈，两者的性能是差不多的，而当竞争资源非常激烈时（即有大量线程同时竞争），此时Lock的性能要远远优于synchronized。
synchronized使用Object对象本身的wait 、notify、notifyAll调度机制，而Lock可以使用Condition进行线程之间的调度
synchronized和volatile的区别
区别

volatile本质是在告诉jvm当前变量在寄存器（工作内存）中的值是不确定的，需要从主存中读取；synchronized则是锁定当前变量，只有当前线程可以访问该变量，其他线程被阻塞住。
volatile仅能使用在变量级别；synchronized则可以使用在变量、方法、和类级别的
volatile仅能实现变量的修改可见性，不能保证原子性；而synchronized则可以保证变量的修改可见性和原子性
volatile不会造成线程的阻塞；synchronized可能会造成线程的阻塞
volatile标记的变量不会被编译器优化；synchronized标记的变量可以被编译器优化
JUC

AQS是什么
AQS 队列同步器是用来构建锁或其他同步组件的基础框架，它使用一个 volatile int state变量作为共享资源，如果线程获取资源失败，则进入同步队列等待；如果获取成功就执行临界区代码，释放资源时会通知同步队列中的等待线程。
同步器面向锁的实现者，简化了锁的实现方式，屏蔽了同步状态管理、线程排队、等待与唤醒等底层操作。

AQS 两种模式
独占模式表示锁只会被一个线程占用，其他线程必须等到持有锁的线程释放锁后才能获取锁，同一时间只能有一个线程获取到锁。
共享模式表示多个线程获取同一个锁有可能成功，ReadLock 就采用共享模式。
独占模式通过 acquire 和 release 方法获取和释放锁，共享模式通过 acquireShared 和 releaseShared方法获取和释放锁。
原子类
JDK 5 提供了 java.util.concurrent.atomic 包，这个包中的原子操作类提供了一种用法简单、性能高效、线程安全地更新一个变量的方式。到 JDK 8 该包共有17个类，依据作用分为四种：原子更新基本类型类、原子更新数组类、原子更新引用类以及原子更新字段类，atomic 包里的类基本都是使用 Unsafe实现的包装类。

AtomicIntger 实现原子更新的原理
getAndIncrement 以原子方式将当前的值加 1，首先在 for 死循环中取得 AtomicInteger 里存储的数值，第二步对 AtomicInteger 当前的值加 1 ，第三步调用 compareAndSet 方法进行原子更新，先检查当前数值是否等于 expect，如果等于则说明当前值没有被其他线程修改，则将值更新为 next，否则会更新失败返回 false，程序会进入 for 循环重新进行 compareAndSet 操作。

CountDownLatch
CountDownLatch 是基于执行时间的同步类，允许一个或多个线程等待其他线程完成操作，构造方法接收一个 int 参数作为计数器，如果要等待 n 个点就传入 n。每次调用 countDown 方法时计数器减 1，await 方法会阻塞当前线程直到计数器变为0，由于 countDown 方法可用在任何地方，所以 n 个点既可以是 n 个线程也可以是一个线程里的 n 个执行步骤。

CyclicBarrier
循环屏障是基于同步到达某个点的信号量触发机制，作用是让一组线程到达一个屏障时被阻塞，直到最后一个线程到达屏障才会解除。构造方法中的参数表示拦截线程数量，每个线程调用 await 方法告诉CyclicBarrier 自己已到达屏障，然后被阻塞。还支持在构造方法中传入一个 Runnable 任务，当线程到达屏障时会优先执行该任务。适用于多线程计算数据，最后合并计算结果的应用场景。

CountDownLacth 的计数器只能用一次，而 CyclicBarrier 的计数器可使用 reset 方法重置，所以CyclicBarrier 能处理更为复杂的业务场景，例如计算错误时可用重置计数器重新计算。

Semaphore
信号量用来控制同时访问特定资源的线程数量，通过协调各个线程以保证合理使用公共资源。信号量可以用于流量控制，特别是公共资源有限的应用场景，比如数据库连接。

Semaphore 的构造方法参数接收一个 int 值，表示可用的许可数量即最大并发数。使用 acquire 方法获得一个许可证，使用 release 方法归还许可，还可以用 tryAcquire 尝试获得许可。

Exchanger
交换者是用于线程间协作的工具类，用于进行线程间的数据交换。它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。

两个线程通过 exchange 方法交换数据，第一个线程执行 exchange 方法后会阻塞等待第二个线程执行该方法，当两个线程都到达同步点时这两个线程就可以交换数据，将本线程生产出的数据传递给对方。

应用场景包括遗传算法、校对工作等。

其它 sleep wait
属于不同的两个类，sleep()方法是线程类（Thread）的静态方法，wait()方法是Object类里的方法。
sleep()方法不会释放锁，wait()方法释放对象锁。
sleep()方法可以在任何地方使用，wait()方法则只能在同步方法或同步块中使用。
sleep()使线程进入阻塞状态（线程睡眠），wait()方法使线程进入等待队列（线程挂起），也就是阻塞类别不同。
wait notify notifAll
wait()、notify/notifyAll() 方法是Object的本地final方法，无法被重写。
wait() 需要被try catch包围，以便发生异常中断也可以使wait等待的线程唤醒。
notify 和wait 的顺序不能错，如果A线程先执行notify方法，B线程再执行wait方法，那么B线程是无法被唤醒的。
notify方法只唤醒一个等待线程并使该线程开始执行。notifyAll 会唤醒所有等待线程。
一般在synchronized 同步代码块里使用 wait()、notify/notifyAll() 方法。

由于 wait()、notify/notifyAll() 在synchronized 代码块执行，说明当前线程一定是获取了锁的。当线程执行wait()方法时候，会释放当前的锁，然后让出CPU，进入等待状态。
只有当 notify/notifyAll() 被执行时候，才会唤醒一个或多个正处于等待状态的线程，然后继续往下执行，直到执行完synchronized 代码块的代码或是中途遇到wait() ，再次释放锁。
也就是说，notify/notifyAll() 的执行只是唤醒沉睡的线程，而不会立即释放锁，锁的释放要看代码块的具体执行情况。所以在编程中，尽量在使用了notify/notifyAll() 后立即退出临界区，以唤醒其他线程让其获得锁
ThreadLocal
ThreadLocal提供了线程内存储变量的能力，这些变量不同之处在于每一个线程读取的变量是对应的互相独立的。通过get和set方法就可以得到当前线程对应的值。ThreadLocal相当于维护了一个map，key就是当前的线程，value就是需要存储的对象。实际上是ThreadLocal的静态内部类ThreadLocalMap为每个Thread都维护了一个数组table，ThreadLocal确定了一个数组下标，而这个下标就是value存储的对应位置。
Thread类中有一个私有的成员变量threadLocals，Thread并没有提供成员变量threadLocals的设置与访问的方法，ThreadLocal是线程Thread中属性threadLocals的管理者。对于ThreadLocal的get,
set，remove的操作结果都是针对当前线程Thread实例的threadLocals存，取，删除操作。
存在的问题

线程复用会产生脏数据，由于线程池会重用 Thread 对象，因此与 Thread 绑定的ThreadLocal 也会被重用。如果没有调用remove 清理与线程相关的 ThreadLocal 信息，那么假如下一个线程没有调用 set设置初始值就可能 get到重用的线程信息。
ThreadLocal 还存在内存泄漏的问题，由于 ThreadLocal 是弱引用，但 Entry 的 value 是强引用，因此当 ThreadLocal 被垃圾回收后，value 依旧不会被释放。因此需要及时调用 remove 方法进行清理操作。

3 jvm
1、内存模型以及分区，需要详细到每个区放什么。
JVM 分为堆区和栈区，还有方法区，初始化的对象放在堆里面，引用放在栈里面， class 类信息常量池（static 常量和 static 变量）等放在方法区 new:

方法区：主要是存储类信息，常量池（static 常量和 static 变量），编译后的代码（字节码）等数据堆：初始化的对象，成员变量 （那种非 static 的变量），所有的对象实例和数组都要在堆上分配栈：栈的结构是栈帧组成的，调用一个方法就压入一帧，帧上面存储局部变量表，操作数栈，方法出口等信息，局部变量表存放的是 8大基础类型加上一个应用类型，所以还是一个指向地址的指针本地方法栈：主要为 Native 方法服务程序计数器：记录当前线程执行的行号
堆里面的分区：Eden，survival （from+ to），老年代，各自的特点。

堆里面分为新生代和老生代（java8 取消了永久代，采用了 Metaspace），新生代包 含 Eden+Survivor 区，survivor 区里面分为 from 和 to 区，内存回收时，如果用的是复制算法，从 from 复制到 to，当经过一次或者多次 GC 之后，存活下来的对象会被移动到老年区，当 JVM 内存不够用的时候，会触发 Full GC，清理 JVM 老年区当新生区满了之后会触发 YGC,先把存活的对象放到其中一个 Survice 区，然后进行垃圾清理。因为如果仅仅清理需要删除的对象，这样会导致内存碎片，因此一般会把 Eden 进行完全的清理，然后整理内存。那么下次 GC 的时候， 就会使用下一个 Survive，这样循环使用。如果有特别大的对象，新生代放不下，就会使用老年代的担保，直接放到老年代里面。因为 JVM 认为，一般大对象的存活时间一般比较久远。
1

GC 的两种判定方法
引用计数法：指的是如果某个地方引用了这个对象就+1，如果失效了就-1，当为 0 就会回收但是 JVM没有用这种方式，因为无法判定相互循环引用（A 引用 B,B 引用 A） 的情况。引用链法： 通过一种 GC ROOT 的对象（方法区中静态变量引用的对象等-static 变量）来判断，如果有一条链能够到达 GCROOT 就说明，不能到达 GC ROOT 就说明可以回收。
1

Minor GC 与 Full GC 分别在什么时候发生？
首先区分一下Minor GC和Full GC。
Minor GC是新生代GC，指的是发生在新生代的垃圾收集动作。由于java对象大都是朝生夕死的，所以Minor GC非常平凡，一般回收速度也比较i快。

Major GC/Full GC 是老年代GC，指的是发生在老年代的GC，出现Major GC一般经常会伴有Minor GC，Major GC的速度比Minor GC慢的多。

何时发生？
(1)Minor GC发生：当jvm无法为新的对象分配空间的时候就会发生Minor gc，所以分配对象的频率越高，也就越容易发生Minor gc。

(2)Full GC：发生GC有两种情况：
①当老年代无法分配内存的时候，会导致MinorGC。
②当发生Minor GC的时候可能触发Full GC，由于老年代要对年轻代进行担保，由于进行一次垃圾回收之前是无法确定有多少对象存活，因此老年代并不能清除自己要担保多少空间，因此采取采用动态估算的方法：也就是上一次回收发送时晋升到老年代的对象容量的平均值作为经验值，这样就会有一个问题，当发生一次Minor GC以后，存活的对象剧增（假设小对象），此时老年代并没有满，但是此时平均值增加了，会造成发生Full GC。

类加载的几个过程：
加载、验证、准备、解析、初始化。然后是使用和卸载了

通过全限定名来加载生成 class 对象到内存中，然后进行验证这个 class 文件，包括文件格式校验、元数据验证，字节码校验等。准备是对这个对象分配内存。解析是将符号引用转化为直接引用（指针引用），初始化就是开始执行构造器的代码
1

6.JVM 内存分哪几个区，每个区的作用是什么
java 虚拟机主要分为以下几个区:
方法区：

有时候也成为永久代，在该区内很少发生垃圾回收，但是并不代表不发生 GC，在这里进行的 GC 主要是对方法区里的常量池和对类型的卸载方法区主要用来存储已被虚拟机加载的类的信息、常量、静态变量和即时编译器编译后的代码等数据。该区域是被线程共享的。方法区里有一个运行时常量池，用于存放静态编译产生的字面量和符号引用。该常量池具有动态性，也就是说常量并不一定是编译时确定，运行时生成的常量也会存在这个常量池中。
1
虚拟机栈:

虚拟机栈也就是我们平常所称的栈内存,它为 java方法服务，每个方法在执行的时候都会创建一个栈帧，用于存储局部变量表、操作数栈、动态链接和方法出口等信息。虚拟机栈是线程私有的，它的生命周期与线程相同。局部变量表里存储的是基本数据类型、returnAddress类型（指向一条字节码指令的地址）和对象引用，这个对象引用有可能是指向对象起始地址的一个指针，也有可能是代表对象的句柄或者与对象相关联的位置。局部变量所需的内存空间在编译器间确定操作数栈的作用主要用来存储运算结果以及运算的操作数，它不同于局部变量表通过索引来访问，而是压栈和出栈的方式每个栈帧都包含一个指向运行时常量池中该栈帧所属方法的引用，持有这个引用是为了支持方法调用过程中的动态连接.动态链接就是将常量池中的符号引用在运行期转化为直接引用。
1
本地方法栈：

本地方法栈和虚拟机栈类似，只不过本地方法栈为 Native 方法服务。

堆
java 堆是所有线程所共享的一块内存，在虚拟机启动时创建，几乎所有的对象实例都在这里创建，因此该区域经常发生垃圾回收操作。

程序计数器

内存空间小，字节码解释器工作时通过改变这个计数值可以选取下一条需要执行的字节码，指令，分支、循环、跳转、异常处理和线程恢复等功能都需要依赖这个计数器完成。该内存区域是唯一一个 java 虚拟机规范没有规定任何 OOM 情况的区域。
7.如和判断一个对象是否存活?(或者 GC 对象的判定方法）
判断一个对象是否存活有两种方法:
1.引用计数法

所谓引用计数法就是给每一个对象设置一个引用计数器，每当有一个地方引用这个对象时，就将计数器加一，引用失效时，计数器就减一。当一个对象的引用计数器为零时，说明此对象没有被引用，也就是“死对象”,将会被垃圾回收.
引用计数法有一个缺陷就是无法解决循环引用问题，也就是说当对象 A 引用对象 B，对象B 又引用者对象 A，那么此时 A,B 对象的引用计数器都不为零，也就造成无法完成垃圾回收，所以主流的虚拟机都没有采用这种算法。
2.可达性算法(引用链法)

该算法的思想是：从一个被称为 GC Roots的对象开始向下搜索，如果一个对象到 GCRoots 没有任何引用链相连时，则说明此对象不可用。
在 java 中可以作为 GC Roots 的对象有以下几种:

虚拟机栈中引用的对象方法区类静态属性引用的对象方法区常量池引用的对象本地方法栈 JNI 引用的对象

虽然这些算法可以判定一个对象是否能被回收，但是当满足上述条件时，一个对象比不一定会被回收。当一个对象不可达 GC Root 时，这个对象并不会立马被回收，而是出于一个死缓的阶段，若要被真正的回收需要经历两次标记

如果对象在可达性分析中没有与 GC Root 的引用链，那么此时就会被第一次标记并且进行一次筛选，筛选的条件是是否有必要执行 finalize()方法。当对象没有覆盖 finalize()方法或者已被虚拟机调用过，那么就认为是没必要的。

如果该对象有必要执行 finalize()方法，那么这个对象将会放在一个称为 F-Queue 的对队列中，虚拟机会触发一个 Finalize()线程去执行，此线程是低优先级的，并且虚拟机不会承

诺一直等待它运行完，这是因为如果 finalize()执行缓慢或者发生了死锁，那么就会造成 FQueue 队列一直等待，造成了内存回收系统的崩溃。GC 对处于 F-Queue 中的对象进行第二次被标记，这时，该对象将被移除”即将回收”集合，等待回收。
8.java 中垃圾收集的方法有哪些?
标记-清除:
这是垃圾收集算法中最基础的，根据名字就可以知道，它的思想就是标记哪些要被回收的对象，然后统一回收。这种方法很简单，但是会有两个主要问题：
1.效率不高，标记和清除的效率都很低；
2.会产生大量不连续的内存碎片，导致以后程序在分配较大的对象时，由于没有充足的连续内存而提前触发一次 GC 动作。

复制算法:
为了解决效率问题，复制算法将可用内存按容量划分为相等的两部分，然后每次只使用其中的一块，当一块内存用完时，就将还存活的对象复制到第二块内存上，然后一次性清楚完第一块内存，再将第二块上的对象复制到第一块。但是这种方式，内存的代价太高，每次基本上都要浪费一般的内存。

于是将该算法进行了改进，内存区域不再是按照 1：1 去划分，而是将内存划分为8:1:1 三部分，较大那份内存交 Eden 区，其余是两块较小的内存区叫 Survior 区。

每次都会优先使用 Eden 区，若 Eden 区满，就将对象复制到第二块内存区上，然后清除 Eden 区，如果此时存活的对象太多，以至于 Survivor 不够时，会将这些对象通过分配担保机制复制到老年代中。(java 堆又分为新生代和老年代)

标记-整理：
该算法主要是为了解决标记-清除，产生大量内存碎片的问题；当对象存活率较高时，也解决了复制算法的效率问题。它的不同之处就是在清除对象的时候现将可回收对象移动到一端，然后清除掉端边界以外的对象，这样就不会产生内存碎片了。

分代收集：
现在的虚拟机垃圾收集大多采用这种方式，它根据对象的生存周期，将堆分为新生代和老年代。在新生代中，由于对象生存期短，每次回收都会有大量对象死去，那么这时就采用复制算法。老年代里的对象存活率较高，没有额外的空间进行分配担保，所以可以使用标记-整理或者 标记-清除。

9.什么是类加载器，类加载器有哪些?
实现通过类的权限定名获取该类的二进制字节流的代码块叫做类加载器。

主要有一下四种类加载器:

启动类加载器(Bootstrap ClassLoader)用来加载 java 核心类库，无法被 java 程序直接引用。扩展类加载器(extensions class loader):它用来加载 Java 的扩展库。Java虚拟机的实现会提供一个扩展库目录。该类加载器在此目录里面查找并加载 Java 类。系统类加载器（system class loader）：它根据 Java 应用的类路径（CLASSPATH） 来加载 Java类。一般来说，Java应用的类都是由它来完成加载的。可以通过ClassLoader.getSystemClassLoader()来获取它。用户自定义类加载器，通过继承 java.lang.ClassLoader 类的方式实现。
类加载器双亲委派模型机制？
当一个类收到了类加载请求时，不会自己先去加载这个类，而是将其委派给父类，由父类去加载，如果此时父类不能加载，反馈给子类，由子类去完成类的加载。
1
11.什么情况下会发生栈内存溢出？
1、栈是线程私有的，栈的生命周期和线程一样，每个方法在执行的时候就会创建一个栈帧，它包含局部变量表、操作数栈、动态链接、方法出口等信息，局部变量表又包括基本数据类型和对象的引用；
2、当线程请求的栈深度超过了虚拟机允许的最大深度时，会抛出StackOverFlowError异常，方法递归调用肯可能会出现该问题；
3、调整参数-xss去调整jvm栈的大小

12.怎么打破双亲委派模型？
自定义类加载器，继承ClassLoader类，重写loadClass方法和findClass方法；

13.强引用、软应用、弱引用、虚引用的区别？
强引用：
强引用是我们使用最广泛的引用，如果一个对象具有强引用，那么垃圾回收期绝对不会回收它，当内存空间不足时，垃圾回收器宁愿抛出OutOfMemoryError，也不会回收具有强引用的对象；我们可以通过显示的将强引用对象置为null，让gc认为该对象不存在引用，从而来回收它；

软引用：
软应用是用来描述一些有用但不是必须的对象，在java中用SoftReference来表示，当一个对象只有软应用时，只有当内存不足时，才会回收它；
软引用可以和引用队列联合使用，如果软引用所引用的对象被垃圾回收器所回收了，虚拟机会把这个软引用加入到与之对应的引用队列中；

弱引用：
弱引用是用来描述一些可有可无的对象，在java中用WeakReference来表示，在垃圾回收时，一旦发现一个对象只具有软引用的时候，无论当前内存空间是否充足，都会回收掉该对象；
弱引用可以和引用队列联合使用，如果弱引用所引用的对象被垃圾回收了，虚拟机会将该对象的引用加入到与之关联的引用队列中；

虚引用：
虚引用就是一种可有可无的引用，无法用来表示对象的生命周期，任何时候都可能被回收，虚引用主要使用来跟踪对象被垃圾回收的活动，虚引用和软引用与弱引用的区别在于：虚引用必须和引用队列联合使用；在进行垃圾回收的时候，如果发现一个对象只有虚引用，那么就会将这个对象的引用加入到与之关联的引用队列中，程序可以通过发现一个引用队列中是否已经加入了虚引用，来了解被引用的对象是否需要被进行垃圾回收；

总结
JVM在一些互联网大厂是面试必问的一个技术点，所以在面试时一定要注重重点，想一些高并发高可用的技术。面试时要掌握节奏，说一些让面试官眼前一亮的技术，有些基础的东西能少说就少说，毕竟面试官面了这么多早就听够了，越是稀少的越是能激发面试官的兴趣，然后掌握在自己的节奏中。

4 开源组件
前端：
1、hui
http://www.h-ui.net

依赖：bootstrap 国产开源里面感觉比较好的一个，同时有通用和后台两套组件库，并且还有付费版本。算是开源独立作者里面能够有个商业模式的了。希望越做越好。这里的是前端框架。

img

2、elementUI
https://element-plus.gitee.io/zh-CN/

基于vue2.x vue3.x

饿了么团队推出的ui组件库

img

3、iviewui View UI Pro
https://pro.iviewui.com/pro/introduce

iviewui推出的ui组件加强版，哈哈，这次有个不一样的

img

4、Ant Design
https://ant.design/index-cn

基于react，当然也有vue的版本

阿里巴巴蚂蚁金服ui团队推出的通用型ui组件库

img

5、AT-UI
https://at-ui.github.io/at-ui/#/zh

京东凹凸实验室推出的通用型ui组件库

img

6、vant
https://youzan.github.io/vant/#/zh-CN

基于vue有赞团队推出的通用型ui组件库，官网的介绍是基于vue的统一风格的组件库，希望能够快速搭建使用。多快呢？试试看吧。

img

7、MuseUI
https://muse-ui.org/#/zh-CN

基于vue2.0 扁平化，这个ui相当的扁平化，主打的风格还有官网都是扁平化的，很有特色。

img

8 uView UI
官网：https://www.uviewui.com/

支持APP/H5/各小程序平台多端发布的通用 UI 框架，知名多端开发框架 uni-app 生态里优秀的UI框架，一次编写，多端发布。

image-20221018202016332

uView UI 是一个用于 uni-app 多端开发的优质 UI 组件库，由第三方爱好者的团队编写。介绍 uView UI 之前，先简单介绍一下 uni-app

全栈

5 服务切分
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5eOGtvyf-1666199870619)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019102942496.png)]

单体架构的优势：
1、便于开发
2、易于测试
3、易于部署

单体架构的不足：
1、复杂性高
2、交付效率低：构建和部署耗时长
3、伸缩性差：只能按整体横向扩展，无法分模块垂直扩展，IO密集型模块和CPU密集型模块无法独立升级和扩容
4、可靠性差：一个BUG可能引起整个项目的运行
5、阻碍技术创新

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-r4ARd682-1666199870623)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019104029242.png)]

微服务架构的优势：
1、易于开发和维护
2、独立部署
3、伸缩性强
4、与组织结构相匹配
5、技术异构性

微服务面临的挑战：
1、服务拆分：
(1)、微服务拆分原则：领域模型、组织结构、康威定律、单一职责
(2)、微服务拥有独立数据库
(3)、微服务之间确定服务边界
2、数据一致性
(1)、可靠性事件模式
(2)、补偿模式-sagas模式
3、服务通信
(1)、通信技术方案：RPC、REST、异步消息
(2)、服务注册和发现
(3)、负载均衡
4、服务网关：
(1)、API Gateway
(2)、为前端服务的后端
(3)、身份认证、路由服务、流量控制、日志统计
5、高可观察
(1)、健康检测、集中控制
(2)、日志聚合及检索
(3)、分布式追踪
6、可靠性(客户端实现)：
(1)、流量控制、超时控制
(2)、舱壁隔离(线程隔离)，熔断机制
(3)、服务降级，幂等重试

微服务拆分原则：
1、单一职责、高内聚低耦合
2、微服务粒度适中
3、考虑团队结构
4、以业务模型切入
5、演进式拆分
6、避免环形依赖与双向依赖
7、DDD

微服务拆分步骤：
1、分析业务模型：
(1)、弱耦合在一起
(2)、高内聚力
2、确定服务边界：
(1)、服务应包含单一的界限上下文
3、微服务数据库拆分

微服务数据一致性：
1、分布式事务不适用微服务
(1)、2PC会有单点故障
(2)、由于锁的原因降低吞吐量
(3)、Nosql数据库并不支持
2、采用最终一致性来实现数据一致性
(1)、可靠性事件模式：消息队列(支付宝转余额宝)
(2)、补偿模式-sagas模型：一些列的有序事务(每一个事物都有补偿子事务)

技术选型的三要素：
1、技术选型的广度和深度：
2、把握和分析技术选项的优缺点
3、紧密结合项目和团队的情况

Eureka简介：
1、支持跨机房的高可用
2、数据一致性是数据最终一致性
3、Eureka Client会对服务注册表进行缓存，降低Eureka的压力，进一步增强了它的高可用

借助logbook输出HTTP日志
1、pom添加logbook依赖
2、在服务提供者工程添加logbook filter以输出日志
3、在服务消费者工程httpclient添加logbook拦截器

JWT介绍：
1、基于token的进行身份验证的方案
2、jwt设计一个字符串由header、payload、signature组成
3、具备安全、自包含、紧凑等特点

JWT优点：
1、安全性高，防止token被伪造和篡改
2、自包含，减少存储开销
3、跨语言，支持多种语言的实现
4、支持过期，发布者等校验

JWT注意事项：
1、消息体是可以被base64解密成铭文
2、jwt不适合存放大量信息
3、无法作废未过期的jwt(可以借用Redis实现)

Redis的score机制：
可以做商品或房屋的热门商品，每点击一个商品的详情，就往Redis(zset)中添加一个，并只留排名前10的；

级联故障解决方案：
1、舱壁隔离(线程隔离)
2、超时控制
3、服务降级
4、熔断机制

Spring Cloud Sleuth原理：使用的是ThreadLocal，使用的是异步线程
1、疑问：(spring.factories)
(1)、追踪数据是如何生成的
(2)、追踪数据是如何再进程内和进程间传递的
(3)、如何解决跨线程池问题的

进程内根据ThreadLocal进行数据的传递
Hystrix是跨线程池的，业务线程和调用线程是隔离的

用户请求—》TraceFilter—》Trace拦截器—》Controller—》HystrixCallable

—》TraceRestTemplate—》TraceAspect—(可以通过MQ，也可以通过Http请求上报)—》Zipkin Server(ES、Mysql做数据存储)

日志检索方案：
1、ELK介绍
(1)、Elasticsearch(日志存储)
(2)、LogStash(负责日志收集)
(3)、Kibana(进行日志图形化展示)

查看日志信息：less info.log

本地缓存：
private final Cache<String, String> registerCache =
CacheBuilder.newBuilder().maximumSize(100).expireAfterAccess(15, TimeUnit.MINUTES)
.removalListener(new RemovalListener<String, String>() {

@Override
public void onRemoval(RemovalNotification<String, String> notification) {
  String email = notification.getValue();
  User user = new User();
  user.setEmail(email);
  List<User> targetUser = userMapper.selectUsersByQuery(user);
  if (!targetUser.isEmpty() && Objects.equal(targetUser.get(0).getEnable(), 0)) {
    userMapper.delete(email);// 代码优化: 在删除前首先判断用户是否已经被激活，对于未激活的用户进行移除操作
  }

}

}).build();
private final Cache<String, String> resetCache = CacheBuilder.newBuilder().maximumSize(100).expireAfterAccess(15, TimeUnit.MINUTES).build();

微服务的消费模式：
1、服务直连模式(RestTemplate)特点：简洁明了、平台语言无关性、无法保证服务的可用性、生产环境比较少用
2、客户端发现模式：
(1)、服务实例启动后，将自己的位置信息提交到服务注册表
(2)、客户端从服务注册表进行查询，来获取可用的服务实例
(3)、客户端自行使用负载均衡算法从多个服务实例中选择出一个
3、服务端发现模式

微服务的消费者：
1、HttpClient(RestTemplateBuilder)
2、Ribbon(基于客户端的负载均衡器(加权、随机、轮询算法))(RestTemplateBuilder+配置)
3、Feign：

使用API网关的意义：
1、API网关的意义：
(1)、集合多个API
(2)、统一API入口
2、常见API网关的实现方式：nginx、zuul、getaway

API网关带来的好处：
1、避免将内部信息泄露给外部
2、能给API添加额外的安全层
3、可以降低API调用的复杂度
4、微服务模拟与虚拟化

zuul简介：
1、功能：认证、压力测试、动态路由、负载削减、安全、静态响应处理、主动交换管理等

服务熔断：
1、断路器
2、断路器模式

熔断器的意义：
1、好处：
(1)、系统稳定
(2)、减少性能损耗
(3)、及时响应
(4)、阀值可定制

熔断器的功能：
1、异常处理
2、日志记录
3、测试失败的操作
4、手动复位
5、加速断路
7、重试失败请求

微服务的高级主题----自动扩展
1、什么是自动扩展
(1)、垂直扩展：就是升级(双核变四核)
(2)、水平扩展：就是数量变多(1台主机增加到4台主机)
2、自我注册和自我发现(服务注册表(Eureka)、客户端、微服务实例)
3、自动扩展的意义：
(1)、提高了高可用性和容错能力
(2)、增加了可伸缩性
(3)、具有最佳使用率，并节约成本
(4)、优先考虑某些服务或服务组

自动扩展的常见模式：
1、自动扩展的不同级别：应用程序级别、基础架构级别

如何实现微服务的自动扩展：
1、要思考的问题：
(1)、如何管理数千个容器
(2)、如何监控他们
(3)、在部署工作时如何应用规则和约束？
(4)、如何利用容器来获得资源效率？
(5)、如何确保至少有一定数量的最小实例正在运行？
(6)、如何确保依赖服务正常运行？
(7)、如何进行滚动的升级和优雅的迁移？
(8)、如何回滚错误的部署？
2、所需功能：依赖两个关键功能
(1)、一个容器抽象层，在许多物理或虚拟机上提供统一的抽象
(2)、容器编排和初始化系统在集群抽象之上只能管理部署
3、容器编排：
(1)、容器编排工具提供了一个抽象层来处理大规模的集装箱部署
(2)、具备发现、资源管理、监控和部署等功能
3、容器编排工作职能：
(1)、集群管理
(2)、自动部署
(3)、可伸缩性
(4)、运行状况监控
(5)、基础架构抽象
(6)、资源优化
(7)、资源分配
(8)、服务可用性
(9)、敏捷性
(10)、隔离

资源分配常用算法：
1、常用算法：
(1)、传播：将负载平均分配到各个主机上
(2)、装箱：负载先把第一台主机用完了，在用其它主机，按需付费
(3)、随机：负载随机选择主机

常见容器编排技术：
(1)、Docker Swarm
(2)、Kubernetes
(3)、Apache Mesos

6 应用框架
Spring
这是其他Java框架中的绝对领导者。 掌握Spring是Java开发人员职位最普遍的要求之一。 造成这种情况的原因很多，但主要的原因是普遍性。

Spring是一个功能强大，轻量级且最受欢迎的Java EE框架。 正如开发人员自己所说：Spring使Java变得简单，现代，高效，可响应，可用于云。 它以依赖注入和面向方面的编程功能而闻名。 实际上，它是框架的容器，使您可以执行任何复杂的任务-从使用数据库到测试过程。

开发人员更有可能选择Spring MVC和Spring Boot。 这些框架的最大优点是能够分离其他模块并由于控制反转（IoC）而专注于一个模块。

优点：

· 使用POJO（普通Java对象）可导致更简单，更灵活的代码库；

· 支持模块化（具有许多软件包和类）；

· 向后兼容和易于测试；

· 庞大的生态系统（Spring Boot，Spring Cloud）和社区；

· 广泛的文档和多个Spring教程。

缺点：

· 不太容易配置

· 陡峭的学习曲线

为了检查每个框架的流行程度，使用了Google趋势。

以下是最近5年Spring受欢迎程度的统计数据：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lmvT8kb0-1666199870624)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019105651769.png)]

Spring受欢迎程度有所下降，但总体情况表明，它正在逐年增长。

如前所述，Spring在Github上拥有非常活跃的社区和37K星。

SpringBoot中好用的框架
http://doc.ruoyi.vip/ruoyi-vue/

RuoYi-Vue 是一个 Java EE 企业级快速开发平台，基于经典技术组合（Spring Boot、Spring Security、MyBatis、Jwt、Vue），内置模块如：部门管理、角色用户、菜单及按钮授权、数据权限、系统参数、日志管理、代码生成等。在线定时任务配置；支持集群，支持多数据源，支持分布式事务。

在线体验

若依官网：http://ruoyi.vip(opens new window)

演示地址：http://vue.ruoyi.vip(opens new window)

代码下载：https://gitee.com/y_project/RuoYi-Vue

Hibernate
在谈论最佳的Java Web框架时，不能忽视Hibernate。

Hibernate是一个ORM（对象/关系映射）框架。 它允许您不使用SQL而是使用Java将查询写到数据库服务器，这通常会改变数据库的常规外观。

尽管Hibernate并不是一个成熟的框架，但它使您可以轻松地转换各种数据库的信息。 无论应用程序大小和用户数量如何，此功能还可以简化扩展。 通常，此框架可以描述为快速，强大，易于扩展和可定制的。

它是在GNU Lesser General的公共2.1许可下分发的免费软件。

优点：

· Hibernate使您可以通过在代码中进行微小的更改来与任何数据库进行通信。

· MySQL，Db2或Oracle，Hibernate与数据库无关；

· 缓存工具以查询相同的错误目录；

· N + 1或缓慢的加载支持；

· 数据丢失风险低，并且需要的功率更少。

缺点：

· 如果电源关闭，您可能会丢失所有数据。

· 重新启动可能非常慢。

查看下面的图表，我们发现在这5年中，Hibernate的普及率一直在下降：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Fq0si1gM-1666199870627)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019105237597.png)]

GitHub星级：4,3K。

MyBatis
MyBatis是用于Java编程的映射框架。 它简化了将Java应用程序与SQL数据库链接的过程：它充当它们之间的中间件。

通常，您将需要Java数据库连接API才能将应用程序连接到关系数据库。 MyBatis简化了过程。 它使开发人员仅使用几行代码即可执行基本的SQL操作。

MyBatis可以与Hibernate框架进行比较。 它们都代表了应用程序和数据库之间的一种桥梁。 唯一的区别是MyBatis不会将Java对象映射到关系数据库。

优点：

· 简便快捷的发展；

· XML标记，支持动态SQL语句编写；

· 非常适合编写纯SQL。

缺点：

· SQL可能绑定到特定的数据库供应商。

· 数据库可移植性差。

根据Google的说法，对该框架的兴趣正逐渐增加：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-49wAnWTY-1666199870627)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019105224745.png)]

Github星级：13.6K。

其余：Struts，Vaadin，JavaServer Faces（JSF）

7 设计模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-c8lR6EP1-1666199870632)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019142732133.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xdcjyIQ9-1666199870633)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019142759178.png)]

一、工厂模式
1.1、场景解析
定义一个对象，根据子类唯一标识，获取对应的实现进行业务操作

1.2、使用示例
以下策略模式、责任链模式、模板方法模式所用到的对象都会交给spring管理，并定义工厂类对外服务。

二、策略模式
2.1、场景解析
当if…else条件一多，可以封装起来替换。

2.2、解决问题
如果分支变多，代码就会变得臃肿，难以维护，可读性低。

如果你需要新增一种类型，那只能在原有代码上修改。

2.3、专业术语
违背了面向对象的开闭原则和单一责任原则

开闭原则（对于扩展是开放的，但是对于修改是封闭的）：增加或者删除某个逻辑，都需要修改到原来代码。

单一原则（规定一个类应该只有一个发生变化的原因）：修改任何类型的分支逻辑代码，都需要改动当前类的代码。

2.4、使用示例
2.4.1 逻辑分析
定义一个接口或者抽象类，里面存放两部分：一部分是唯一类型指定，另一部分是该类型相关业务逻辑实现。

2.4.2 场景解析
每个人在社会某个时间段某个领域都有相应的角色以及相对应的工作；如学生和程序员，学生需要读书，而程序需要打码。

2.4.3 代码实现
策略接口对象

public interface IPersonStrategy {
// 类型
PersonRoleEnum type();
// 业务逻辑
void work();
}
实现

@Slf4j
@Component
public class StudentStrategy implements IPersonStrategy {

@Override
public PersonRoleEnum type() {
    return PersonRoleEnum.STUDENT;
}

@Override
public void work() {
    log.info("学生要努力学习");
}

}

@Slf4j
@Component
public class ProgrammerStrategy implements IPersonStrategy{
@Override
public PersonRoleEnum type() {
return PersonRoleEnum.PROGRAMMER;
}

@Override
public void work() {
    log.info("程序员要努力打码");
}

}
工厂

@Component
public class PersonStrategyFactory {

private Map<PersonRoleEnum, IPersonStrategy> iPersonStrategyMap = Maps.newConcurrentMap();

public void work(PersonRoleEnum personRoleEnum) {
    Optional.of(iPersonStrategyMap.get(personRoleEnum)).ifPresent(bean -> bean.work());
}

@PostConstruct
public void init() {
    Map<String, IPersonStrategy> beansOfType = SpringUtil.getBeansOfType(IPersonStrategy.class);
    beansOfType.values().forEach(bean -> iPersonStrategyMap.put(bean.type(), bean));
}

}
使用

@SpringBootTest
public class PersonStrategyUsage {

@Autowired
private PersonStrategyFactory personStrategyFactory;

@Test
public void test() {
    personStrategyFactory.work(PersonRoleEnum.STUDENT);
    personStrategyFactory.work(PersonRoleEnum.PROGRAMMER);
}

}
输出

2022-08-11 15:50:14,728 c.j.s.s.i.model.cl.StudentStrategy- 学生要努力学习
2022-08-11 15:50:14,728 c.j.s.s.i.m.cl.ProgrammerStrategy- 程序员要努力打码
三、责任链模式
3.1、场景解析
多个参与者依序处理一个东西。

3.2、解决问题
请求者和接收者完全解耦：在责任链模式中，请求者只需发送请求，无须知道具体的处理逻辑；而每一个接收者只处理自己的部分，无须知道请求者和其它接收者。

动态组合职责：在使用时可以灵活的组合接收对象，也可以根据需要拓展接收对象。

3.3、使用示例
3.3.1 逻辑分析
定义一个接口或者抽象类

实现继承对象差异化处理

处理对象集合初始化（随意组合）

3.3.2 场景解析
OA申请：在平时的工作中，难免遇到突发事情需要请假，当请假天数在[1,2]范围时，由部门经理审核；当请假天数在[3,5]范围时，由总经理审核；
为了简便没行政主管啥事，可自行拓展组合。

3.3.3 代码实现
责任抽象对象

@Data
public abstract class AbstractOAHandler {

/**
 * 是否需要传递给下个处理者处理
 */
private AbstractOAHandler nextHandler;

/**
 * 处理请求，并返回处理结果
 *
 * @param user 请求者
 * @param day  请假天数
 * @return 处理结果
 */
public abstract String handlerRequest(String user, Integer day);

}
实现

@Slf4j
@Order(1) // 数字越小排越前,为了测试方便
@Component
public class DepartmentManagerOAHandler extends AbstractOAHandler{
@Override
public String handlerRequest(String user, Integer day) {
log.info(“经过了部门经理”);
if (day > 0 && day < 3) {
return String.format(“【部门经理】通过了【%s】%d天请假申请！”, user, day);
}else {
if (Objects.nonNull(this.getNextHandler())) {
return this.getNextHandler().handlerRequest(user, day);
}
}
return String.format(“不通过【%s】%d天请假申请！”, user, day);
}
}

@Slf4j
@Order(0)
@Component
public class GeneralManagerOAHandler extends AbstractOAHandler{
@Override
public String handlerRequest(String user, Integer day) {
log.info(“经过了总经理”);
if (day >= 3 && day <= 5) {
return String.format(“【总经理】通过了【%s】%d天请假申请！”, user, day);
}else {
if (Objects.nonNull(this.getNextHandler())) {
return this.getNextHandler().handlerRequest(user, day);
}
}
return String.format(“不通过【%s】%d天请假申请！”, user, day);
}
}
工厂

@Component
public class OAHandlerFactory {
@Autowired
private List handlers;
private AbstractOAHandler handler;

@PostConstruct
public void init() {
    for (int i = 0; i < handlers.size(); i++) {
        if (i == 0) {
            handler = handlers.get(0);
        }else {
            handler.setNextHandler(handlers.get(i));
        }
    }
}
public String request(String user, Integer day) {
    return handler.handlerRequest(user, day);
}

}
使用

@SpringBootTest
public class OAHandlerUsage {

@Autowired
private OAHandlerFactory oaHandler;

@Test
public void test() {
    System.out.println(oaHandler.request("小明", 0));
    System.out.println(oaHandler.request("小明", 1));
    System.out.println(oaHandler.request("小明", 3));
}

}
输出

2022-08-11 15:45:48,550 c.j.s.s.i.m.z.GeneralManagerOAHandler- 经过了总经理
2022-08-11 15:45:48,550 c.j.s.s.i.m.z.DepartmentManagerOAHandler- 经过了部门经理
不通过【小明】0天请假申请！
2022-08-11 15:45:48,550 c.j.s.s.i.m.z.GeneralManagerOAHandler- 经过了总经理
2022-08-11 15:45:48,550 c.j.s.s.i.m.z.DepartmentManagerOAHandler- 经过了部门经理
【部门经理】通过了【小明】1天请假申请！
2022-08-11 15:45:48,550 c.j.s.s.i.m.z.GeneralManagerOAHandler- 经过了总经理
【总经理】通过了【小明】3天请假申请！
四、模板方法
4.1、场景解析
父类定义骨架，子类可以差异化实现某些细节。

4.2、解决问题
把不变的行为写在父类，去除子类重复代码，提高代码的复用性，符合开闭原则

4.3、使用示例
4.3.1 逻辑分析
定义一个抽象类

继承对象差异化实现抽象方法

4.3.2 场景解析
在我们学习爬取网站商品入库分析时，假如以爬取A|B网站为例，通常A|B网站的商品字段以及请求接口数据是不相同的（授权鉴权等操作忽略），但是我们的整个爬取步骤是相同的。
如：1、请求接口，拿到商品数据 2、将数据处理成我们需要的格式 3、持久化到数据库。

4.3.3 代码实现
模板方法抽象对象

@Slf4j
public abstract class AbstractCrawlTemplate {

public final void crawl() {

    // 1、请求接口，拿到商品数据
    List<T> list = request();
    log.info("接口请求完成，数据量：{}", list.size());

    // 2、将数据处理成我们需要的格式t
    List<GoodsInfo> goodsInfos = list.stream().map(item -> convert(item)).collect(Collectors.toList());
    log.info("数据转换成功");

    // 3、持久化到数据库
    log.info("持久化成功");
}

protected abstract List<T> request();
protected abstract GoodsInfo convert(T t);

}
实现

@Component(“A”)
public class ACrawl extends AbstractCrawlTemplate {
@Override
protected List request() {
return new ArrayList() {{
add(AGoodsInfo.builder().id(1L).name(“牛奶”).price(BigDecimal.TEN).stock(100).build());
add(AGoodsInfo.builder().id(1L).name(“糖”).price(BigDecimal.TEN).stock(100).build());
}};
}

@Override
protected GoodsInfo convert(AGoodsInfo aGoodsInfo) {
    GoodsInfo goodsInfo = new GoodsInfo();
    goodsInfo.setId("a_" + aGoodsInfo.getId());
    goodsInfo.setName(aGoodsInfo.getName());
    goodsInfo.setPrice(aGoodsInfo.getPrice());
    goodsInfo.setStock(BigDecimal.valueOf(aGoodsInfo.getStock()));
    return goodsInfo;
}

}

@Component(“B”)
public class BCrawl extends AbstractCrawlTemplate {
@Override
protected List request() {
return new ArrayList() {{
add(BGoodsInfo.builder().bh(“1”).mc(“牛奶”).jg(12d).kc(10d).build());
}};
}

@Override
protected GoodsInfo convert(BGoodsInfo bGoodsInfo) {
    GoodsInfo goodsInfo = new GoodsInfo();
    goodsInfo.setId("b_" + bGoodsInfo.getBh());
    goodsInfo.setName(bGoodsInfo.getMc());
    goodsInfo.setPrice(BigDecimal.valueOf(bGoodsInfo.getJg()));
    goodsInfo.setStock(BigDecimal.valueOf(bGoodsInfo.getKc()));
    return goodsInfo;
}

}
工厂

@Component
public class CrawlTemplateFactory {
@Resource
private Map<String, AbstractCrawlTemplate> crawlMap;

public void exec(WebsiteEnum websiteEnum) {
    crawlMap.get(websiteEnum.name()).crawl();
}

}
使用

@SpringBootTest
public class CrawlTemplateUsage {

@Autowired
private CrawlTemplateFactory crawlTemplateFactory;

@Test
public void test() {
    crawlTemplateFactory.exec(WebsiteEnum.A);
}

}
输出

2022-08-13 16:42:02,069 c.j.s.s.i.m.m.AbstractCrawlTemplate- 接口请求完成，数据量：2
2022-08-13 16:42:02,070 c.j.s.s.i.m.m.AbstractCrawlTemplate- 数据转换成功
2022-08-13 16:42:02,070 c.j.s.s.i.m.m.AbstractCrawlTemplate- 持久化成功
五、单例模式
5.1、场景解析
保证一个类仅有一个实例，并提供一个访问它的全局访问点。

5.2 常见单列模式
懒汉模式：实例在需要用的时候才会去创建，存在线程安全问题。

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class LanHanSingleton {
private static LanHanSingleton instance;
public static LanHanSingleton getInstance() {
if (instance == null) {
instance = new LanHanSingleton();
}
return instance;
}
}
饿汉模式：实例在初始化的时候已经创建好了。没有线程安全问题，但是浪费内存。

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class EHanSingleton {
private static EHanSingleton instance = new EHanSingleton();
public static EHanSingleton getInstance() {
return instance;
}
}
双重校验锁：综合了懒汉模式和饿汉模式的优缺点，通过在synchronized关键字外加多一层if条件判断，既保证了线程安全也提高了效率、节省了内存。

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class DoubleCheckSingleton {
private volatile static DoubleCheckSingleton instance;
public static DoubleCheckSingleton getInstance() {
if (instance == null) {
synchronized (DoubleCheckSingleton.class) {
if (instance == null) {
instance = new DoubleCheckSingleton();
}
}
}
return instance;
}
}
枚举：代码简洁，线程安全

public enum EnumSingleton {
INSTANCE;
public EnumSingleton getInstance(){
return INSTANCE;
}
}
link:https://www.jianshu.com/p/1cce5de4ff36

三 ，分布式
1 消息队列
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3MQbhpPK-1666199870635)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019143258519.png)]
多个mq如何选型？
MQ 描述
RabbitMQ erlang开发，对消息堆积的支持并不好，当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降。每秒钟可以处理几万到十几万条消息。
RocketMQ java开发，面向互联网集群化功能丰富，对在线业务的响应时延做了很多的优化，大多数情况下可以做到毫秒级的响应，每秒钟大概能处理几十万条消息。
Kafka Scala开发，面向日志功能丰富，性能最高。当你的业务场景中，每秒钟消息数量没有那么多的时候，Kafka 的时延反而会比较高。所以，Kafka 不太适合在线业务场景。
ActiveMQ java开发，简单，稳定，性能不如前面三个。小型系统用也ok，但是不推荐。推荐用互联网主流的。
为什么要使用MQ？
因为项目比较大，做了分布式系统，所有远程服务调用请求都是同步执行经常出问题，所以引入了mq

作用 描述
解耦 系统耦合度降低，没有强依赖关系
异步 不需要同步执行的远程调用可以有效提高响应时间
削峰 请求达到峰值后，后端service还可以保持固定消费速率消费，不会被压垮
RocketMQ由哪些角色组成，每个角色作用和特点是什么？
角色 作用
Nameserver 无状态，动态列表；这也是和zookeeper的重要区别之一。zookeeper是有状态的。
Producer 消息生产者，负责发消息到Broker。
Broker 就是MQ本身，负责收发消息、持久化消息等。
Consumer 消息消费者，负责从Broker上拉取消息进行消费，消费完进行ack。
RocketMQ中的Topic和JMS的queue有什么区别？
queue就是来源于数据结构的FIFO队列。而Topic是个抽象的概念，每个Topic底层对应N个queue，而数据也真实存在queue上的。

RocketMQ Broker中的消息被消费后会立即删除吗？
不会，每条消息都会持久化到CommitLog中，每个Consumer连接到Broker后会维持消费进度信息，当有消息消费后只是当前Consumer的消费进度（CommitLog的offset）更新了。

追问：那么消息会堆积吗？什么时候清理过期消息？
4.6版本默认48小时后会删除不再使用的CommitLog文件

检查这个文件最后访问时间

判断是否大于过期时间

指定时间删除，默认凌晨4点

源码如下：

/**

• {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#isTimeToDelete()}
  */
  private boolean isTimeToDelete() {
  // when = “04”;
  String when = DefaultMessageStore.this.getMessageStoreConfig().getDeleteWhen();
  // 是04点，就返回true
  if (UtilAll.isItTimeToDo(when)) {
  return true;
  }
  // 不是04点，返回false
  return false;
  }

/**

• {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles()}
  */
  private void deleteExpiredFiles() {
  // isTimeToDelete()这个方法是判断是不是凌晨四点，是的话就执行删除逻辑。
  if (isTimeToDelete()) {
  // 默认是72，但是broker配置文件默认改成了48，所以新版本都是48。
  long fileReservedTime = 48 * 60 * 60 * 1000;
  deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(72 * 60 * 60 * 1000, xx, xx, xx);
  }
  }

/**

• {@link org.apache.rocketmq.store.CommitLog#deleteExpiredFile()}
  */
  public int deleteExpiredFile(xxx) {
  // 这个方法的主逻辑就是遍历查找最后更改时间+过期时间，小于当前系统时间的话就删了（也就是小于48小时）。
  return this.mappedFileQueue.deleteExpiredFileByTime(72 * 60 * 60 * 1000, xx, xx, xx);
  }

12345678910111213141516171819202122232425262728293031323334
RocketMQ消费模式有几种？
消费模型由Consumer决定，消费维度为Topic。

集群消费

1.一条消息只会被同Group中的一个Consumer消费

2.多个Group同时消费一个Topic时，每个Group都会有一个Consumer消费到数据

广播消费

消息将对一 个Consumer Group 下的各个 Consumer 实例都消费一遍。即即使这些 Consumer 属于同一个Consumer Group ，消息也会被 Consumer Group 中的每个 Consumer 都消费一次。

消费消息是push还是pull？
RocketMQ没有真正意义的push，都是pull，虽然有push类，但实际底层实现采用的是长轮询机制，即拉取方式

broker端属性 longPollingEnable 标记是否开启长轮询。默认开启

源码如下：

// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
// 看到没，这是一只披着羊皮的狼，名字叫PushConsumerImpl，实际干的确是pull的活。

// 拉取消息，结果放到pullCallback里
this.pullAPIWrapper.pullKernelImpl(pullCallback);
12345
追问：为什么要主动拉取消息而不使用事件监听方式？
事件驱动方式是建立好长连接，由事件（发送数据）的方式来实时推送。

如果broker主动推送消息的话有可能push速度快，消费速度慢的情况，那么就会造成消息在consumer端堆积过多，同时又不能被其他consumer消费的情况。而pull的方式可以根据当前自身情况来pull，不会造成过多的压力而造成瓶颈。所以采取了pull的方式。

broker如何处理拉取请求的？
Consumer首次请求Broker

Broker中是否有符合条件的消息

有 ->

响应Consumer

等待下次Consumer的请求

没有

挂起consumer的请求，即不断开连接，也不返回数据

使用consumer的offset，

DefaultMessageStore#ReputMessageService#run方法

每隔1ms检查commitLog中是否有新消息，有的话写入到pullRequestTable

当有新消息的时候返回请求

PullRequestHoldService 来Hold连接，每个5s执行一次检查pullRequestTable有没有消息，有的话立即推送

RocketMQ如何做负载均衡？
通过Topic在多Broker中分布式存储实现。

producer端
发送端指定message queue发送消息到相应的broker，来达到写入时的负载均衡

提升写入吞吐量，当多个producer同时向一个broker写入数据的时候，性能会下降

消息分布在多broker中，为负载消费做准备

默认策略是随机选择：

producer维护一个index

每次取节点会自增

index向所有broker个数取余

自带容错策略

其他实现：

SelectMessageQueueByHash

hash的是传入的args

SelectMessageQueueByRandom

SelectMessageQueueByMachineRoom 没有实现

也可以自定义实现MessageQueueSelector接口中的select方法

MessageQueue select(final List mqs, final Message msg, final Object arg);
1
consumer端
采用的是平均分配算法来进行负载均衡。

其他负载均衡算法

平均分配策略(默认)(AllocateMessageQueueAveragely)
环形分配策略(AllocateMessageQueueAveragelyByCircle)
手动配置分配策略(AllocateMessageQueueByConfig)
机房分配策略(AllocateMessageQueueByMachineRoom)
一致性哈希分配策略(AllocateMessageQueueConsistentHash)
靠近机房策略(AllocateMachineRoomNearby)

追问：当消费负载均衡consumer和queue不对等的时候会发生什么？
Consumer和queue会优先平均分配，如果Consumer少于queue的个数，则会存在部分Consumer消费多个queue的情况，如果Consumer等于queue的个数，那就是一个Consumer消费一个queue，如果Consumer个数大于queue的个数，那么会有部分Consumer空余出来，白白的浪费了。

消息重复消费
影响消息正常发送和消费的重要原因是网络的不确定性。

引起重复消费的原因

ACK

正常情况下在consumer真正消费完消息后应该发送ack，通知broker该消息已正常消费，从queue中剔除

当ack因为网络原因无法发送到broker，broker会认为词条消息没有被消费，此后会开启消息重投机制把消息再次投递到consumer

消费模式

在CLUSTERING模式下，消息在broker中会保证相同group的consumer消费一次，但是针对不同group的consumer会推送多次

解决方案

数据库表

处理消息前，使用消息主键在表中带有约束的字段中insert

Map

单机时可以使用map ConcurrentHashMap -> putIfAbsent guava cache

Redis

分布式锁搞起来。

如何让RocketMQ保证消息的顺序消费
你们线上业务用消息中间件的时候，是否需要保证消息的顺序性?

如果不需要保证消息顺序，为什么不需要?假如我有一个场景要保证消息的顺序，你们应该如何保证?

首先多个queue只能保证单个queue里的顺序，queue是典型的FIFO，天然顺序。多个queue同时消费是无法绝对保证消息的有序性的。所以总结如下：

同一topic，同一个QUEUE，发消息的时候一个线程去发送消息，消费的时候 一个线程去消费一个queue里的消息。

追问：怎么保证消息发到同一个queue？
Rocket MQ给我们提供了MessageQueueSelector接口，可以自己重写里面的接口，实现自己的算法，举个最简单的例子：判断i % 2 == 0，那就都放到queue1里，否则放到queue2里。

for (int i = 0; i < 5; i++) {
Message message = new Message(“orderTopic”, (“hello!” + i).getBytes());
producer.send(
// 要发的那条消息
message,
// queue 选择器 ，向 topic中的哪个queue去写消息
new MessageQueueSelector() {
// 手动 选择一个queue
@Override
public MessageQueue select(
// 当前topic 里面包含的所有queue
List mqs,
// 具体要发的那条消息
Message msg,
// 对应到 send（） 里的 args，也就是2000前面的那个0
Object arg) {
// 向固定的一个queue里写消息，比如这里就是向第一个queue里写消息
if (Integer.parseInt(arg.toString()) % 2 == 0) {
return mqs.get(0);
} else {
return mqs.get(1);
}
}
},
// 自定义参数：0
// 2000代表2000毫秒超时时间
i, 2000);
}
12345678910111213141516171819202122232425262728
RocketMQ如何保证消息不丢失
首先在如下三个部分都可能会出现丢失消息的情况：

Producer端

Broker端

Consumer端

1、Producer端如何保证消息不丢失
采取send()同步发消息，发送结果是同步感知的。

发送失败后可以重试，设置重试次数。默认3次。

producer.setRetryTimesWhenSendFailed(10);

集群部署，比如发送失败了的原因可能是当前Broker宕机了，重试的时候会发送到其他Broker上。

2、Broker端如何保证消息不丢失

修改刷盘策略为同步刷盘。默认情况下是异步刷盘的。

flushDiskType = SYNC_FLUSH

集群部署，主从模式，高可用。

3、Consumer端如何保证消息不丢失
完全消费正常后在进行手动ack确认。

rocketMQ的消息堆积如何处理
下游消费系统如果宕机了，导致几百万条消息在消息中间件里积压，此时怎么处理?

你们线上是否遇到过消息积压的生产故障?如果没遇到过，你考虑一下如何应对?

首先要找到是什么原因导致的消息堆积，是Producer太多了，Consumer太少了导致的还是说其他情况，总之先定位问题。

然后看下消息消费速度是否正常，正常的话，可以通过上线更多consumer临时解决消息堆积问题

追问：如果Consumer和Queue不对等，上线了多台也在短时间内无法消费完堆积的消息怎么办？
准备一个临时的topic

queue的数量是堆积的几倍

queue分布到多Broker中

上线一台Consumer做消息的搬运工，把原来Topic中的消息挪到新的Topic里，不做业务逻辑处理，只是挪过去

上线N台Consumer同时消费临时Topic中的数据

改bug

恢复原来的Consumer，继续消费之前的Topic

追问：堆积时间过长消息超时了？
RocketMQ中的消息只会在commitLog被删除的时候才会消失，不会超时。也就是说未被消费的消息不会存在超时删除这情况。

追问：堆积的消息会不会进死信队列？
不会，消息在消费失败后会进入重试队列（%RETRY%+ConsumerGroup），16次（默认16次）才会进入死信队列（%DLQ%+ConsumerGroup）。

源码如下：

public class SubscriptionGroupConfig {
private int retryMaxTimes = 16;
}

// {@link org.apache.rocketmq.broker.processor.SendMessageProcessor#asyncConsumerSendMsgBack}
// maxReconsumeTimes = 16
int maxReconsumeTimes = subscriptionGroupConfig.getRetryMaxTimes();
// 如果重试次数大于等于16，则创建死信队列
if (msgExt.getReconsumeTimes() >= maxReconsumeTimes || delayLevel < 0) {
// MixAll.getDLQTopic()就是给原有groupname拼上DLQ，死信队列
newTopic = MixAll.getDLQTopic(requestHeader.getGroup());
// 创建死信队列
topicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(xxx)
}
1234567891011121314
扩展：每次重试的时间间隔：

public class MessageStoreConfig {
// 每隔如下时间会进行重试，到最后一次时间重试失败的话就进入死信队列了。
private String messageDelayLevel = “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”;
}
1234
12345
看到这个源码你可能蒙蔽了，这不是18个时间间隔嘛。怎么是16次？继续看下面代码，我TM也懵了。

/**

• {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#sendMessageBack()}
• sendMessageBack()这个方法是消费失败后会请求他，意思是把消息重新放到队列，进行重试。
  */
  public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName) {
  Message newMsg = new Message();
  // ！！！我TM，真相了，3 + xxx。他是从第三个开始的。也就是舍弃了前两个时间间隔，18 - 2 = 16。也就是说第一次重试是在10s，第二次30s。
  // TMD！！！
  // TMD！！！
  // TMD！！！
  newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
  this.mQClientFactory.getDefaultMQProducer().send(newMsg);
  }
  1234567891011121314
  RocketMQ在分布式事务支持这块机制的底层原理?
  你们用的是RocketMQ?RocketMQ很大的一个特点是对分布式事务的支持，你说说他在分布式事务支持这块机制的底层原理?

分布式系统中的事务可以使用TCC（Try、Confirm、Cancel）、2pc来解决分布式系统中的消息原子性

RocketMQ 4.3+提供分布事务功能，通过 RocketMQ 事务消息能达到分布式事务的最终一致

RocketMQ实现方式：

Half Message：预处理消息，当broker收到此类消息后，会存储到RMQ_SYS_TRANS_HALF_TOPIC的消息消费队列中

检查事务状态：Broker会开启一个定时任务，消费RMQ_SYS_TRANS_HALF_TOPIC队列中的消息，每次执行任务会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知，Broker会定时去回调在重新检查。

超时：如果超过回查次数，默认回滚消息。

也就是他并未真正进入Topic的queue，而是用了临时queue来放所谓的half message，等提交事务后才会真正的将half message转移到topic下的queue。

如果让你来动手实现一个分布式消息中间件，整体架构你会如何设计实现?
我个人觉得从以下几个点回答吧：

需要考虑能快速扩容、天然支持集群

持久化的姿势

高可用性

数据0丢失的考虑

服务端部署简单、client端使用简单

看过RocketMQ 的源码没有。如果看过，说说你对RocketMQ 源码的理解?
要真让我说，我会吐槽蛮烂的，首先没任何注释，可能是之前阿里巴巴写了中文注释，捐赠给apache后，apache觉得中文注释不能留，自己又懒得写英文注释，就都给删了。里面比较典型的设计模式有单例、工厂、策略、门面模式。单例工厂无处不在，策略印象深刻比如发消息和消费消息的时候queue的负载均衡就是N个策略算法类，有随机、hash等，这也是能够快速扩容天然支持集群的必要原因之一。持久化做的也比较完善，采取的CommitLog来落盘，同步异步两种方式。

高吞吐量下如何优化生产者和消费者的性能?
开发
同一group下，多机部署，并行消费

单个Consumer提高消费线程个数

批量消费

消息批量拉取

业务逻辑批量处理

运维
网卡调优

jvm调优

多线程与cpu调优

Page Cache

再说说RocketMQ 是如何保证数据的高容错性的?
在不开启容错的情况下，轮询队列进行发送，如果失败了，重试的时候过滤失败的Broker

如果开启了容错策略，会通过RocketMQ的预测机制来预测一个Broker是否可用

如果上次失败的Broker可用那么还是会选择该Broker的队列

如果上述情况失败，则随机选择一个进行发送

在发送消息的时候会记录一下调用的时间与是否报错，根据该时间去预测broker的可用时间

其实就是send消息的时候queue的选择。源码在如下：

org.apache.rocketmq.client.latency.MQFaultStrategy#selectOneMessageQueue()
1
任何一台Broker突然宕机了怎么办？
Broker主从架构以及多副本策略。Master收到消息后会同步给Slave，这样一条消息就不止一份了，Master宕机了还有slave中的消息可用，保证了MQ的可靠性和高可用性。而且Rocket MQ4.5.0开始就支持了Dlegder模式，基于raft的，做到了真正意义的HA。

Broker把自己的信息注册到哪个NameServer上？
这么问明显在坑你，因为Broker会向所有的NameServer上注册自己的信息，而不是某一个，是每一个，全部！

安装及使用方法
单节点快速安装
环境要求：>=jdk1.8即可

这里下载4.7.1的版本

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Vdx4UW2d-1666199870639)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019145222616.png)]

1.上传
rz
2.解压
unzip rocketmq-all-4.7.1-bin-release.zip
3.创建启动脚本
cd /data1/rocketmq-all-4.7.1-bin-release
vim start.sh
nohup sh bin/mqnamesrv &
nohup sh bin/mqbroker -n 本机ip地址:9876 -c conf/broker.conf &
#创建完成之后修改文件权限
chmod 744 start.sh
4.启动
sh start.sh
5.查看是否启动成功
jps |grep Start
6.验证
#先添加环境变量

vim /etc/profile
export NAMESRV_ADDR=本机ip地址:9876
#环境变量生效

source /etc/profile
#生产消息

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
#消费消息

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
7.创建停止脚本
vim stop.sh
nohup sh bin/mqshutdown broker &
nohup sh bin/mqshutdown namesrv &
#修改文件权限

chmod 744 stop.sh
8.查看日志
tail -f ~/logs/rocketmqlogs/namesrv.log
tail -f ~/logs/rocketmqlogs/broker.log
9.如果需要修改内存
#修改bin/runserver.sh

JAVA_OPT=“${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m”
#修改bin/runbroker.sh

JAVA_OPT=“${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m”
多master模式
1.优缺点
一个集群无Slave，全是Master，例如2个Master或者3个Master

1.优点
配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；

2.缺点
单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。

2.安装
下面以A,B2台机器为例，namesrv和broker在同一台机器上

1.上传安装包
AB2台机器分别上传安装包

rz
2.解压
unzip rocketmq-all-4.7.1-bin-release.zip
3.创建namesrv启动脚本
cd /data1/rocketmq-all-4.7.1-bin-release
vim startnamesrv.sh
nohup sh bin/mqnamesrv &
#创建完成之后修改文件权限

chmod 744 startnamesrv.sh
4.启动namesrv
sh startnamesrv.sh
5.修改broker配置
A修改conf/2m-noslave/broker-a.properties

B修改conf/2m-noslave/broker-b.properties
添加的下面内容

#添加nameserver

namesrvAddr=A机器ip地址:9876;B机器ip地址:9876
#添加自动创建topic=false

autoCreateTopicEnable=false
6.创建brokersrv启动脚本
如果是B机器，改为conf/2m-noslave/broker-b.properties

cd /data1/rocketmq-all-4.7.1-bin-release
vim startbrokersrv.sh
nohup sh bin/mqbroker -c conf/2m-noslave/broker-a.properties &
#创建完成之后修改文件权限

chmod 744 startbrokersrv.sh
7.启动brokersrv
sh startbrokersrv.sh
8.查看是否启动成功
jps |grep Start
9.查看日志
tail -f ~/logs/rocketmqlogs/namesrv.log
tail -f ~/logs/rocketmqlogs/broker.log
10.如果需要修改内存
#修改bin/runserver.sh

JAVA_OPT=“${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m”
#修改bin/runbroker.sh

JAVA_OPT=“${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m”
多Master多Slave模式-异步复制
1.优缺点
每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级）

1.优点
即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；

2.缺点
Master宕机，磁盘损坏情况下会丢失少量消息。

2.安装

下面以6台机器为例，2n2m2s

A,B namesrv

C,D broker master

E,F broker slave 其中E是C的slave F是D的slave

1.上传安装包
6台机器分别上传安装包

rz
2.解压
unzip rocketmq-all-4.7.1-bin-release.zip
3.创建namesrv启动脚本
cd /data1/rocketmq-all-4.7.1-bin-release
vim startnamesrv.sh
nohup sh bin/mqnamesrv &
#创建完成之后修改文件权限

chmod 744 startnamesrv.sh
4.启动namesrv
sh startnamesrv.sh
5.修改broker配置
C修改conf/2m-2s-async/broker-a.properties

D修改conf/2m-2s-async/broker-b.properties

E修改conf/2m-2s-async/broker-a-s.properties

F修改conf/2m-2s-async/broker-b-s.properties
添加的下面内容

#添加nameserver

namesrvAddr=A机器ip地址:9876;B机器ip地址:9876
#添加自动创建topic=false

autoCreateTopicEnable=false
6.创建brokersrv启动脚本
D机器，改为conf/2m-2s-async/broker-b.properties

E机器，改为conf/2m-2s-async/broker-a-s.properties

F机器，改为conf/2m-2s-async/broker-b-s.properties

cd /data1/rocketmq-all-4.7.1-bin-release
vim startbrokersrv.sh
nohup sh bin/mqbroker -c conf/2m-2s-async/broker-a.properties &
#创建完成之后修改文件权限

chmod 744 startbrokersrv.sh
7.启动brokersrv
sh startbrokersrv.sh
8.查看是否启动成功
jps |grep Start
9.查看日志
tail -f ~/logs/rocketmqlogs/namesrv.log
tail -f ~/logs/rocketmqlogs/broker.log
10.如果需要修改内存
#修改bin/runserver.sh

JAVA_OPT=“${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m”
#修改bin/runbroker.sh

JAVA_OPT=“${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m”
多Master多Slave模式-同步双写
1.优缺点
每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

1.优点
数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；

2.缺点
性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

2.安装
参照多Master多Slave模式-异步复制

只需要启动broker的时候使用conf/2m-2s-sync下面的配置文件即可

mqadmin管理工具
1.使用文档
地址：rocketmq/operation.md at master · apache/rocketmq · GitHub

2.常用命令
注意：

执行命令方法：./mqadmin {command} {args} 几乎所有命令都需要配置-n表示NameServer地址，格式为ip:port 几乎所有命令都可以通过-h获取帮助 如果既有Broker地址（-b）配置项又有clusterName（-c）配置项，则优先以Broker地址执行命令，如果不配置Broker地址，则对集群中所有主机执行命令，只支持一个Broker地址。-b格式为ip:port，port默认是10911

1.创建更新topic配置
./mqadmin updateTopic -n ip:9876 -c DefaultCluster -t TopicTest
2.查看topic列表信息
./mqadmin topicList -n ip:9876 -c
3.删除topic
./mqadmin deleteTopic -n ip:9876 -c DefaultCluster -t TopicTest
4.查看topic路由信息
./mqadmin topicRoute -n ip:9876 -t TopicTest
5.查看topic消息队列offset
./mqadmin topicStatus -n ip:9876 -t TopicTest
6.计算负载
以平均负载算法计算消费者列表负载消息队列的负载结果

./mqadmin allocateMQ -n ip:9876 -i ip1,ip2 -t TopicTest
7.查看topic的tps
打印Topic订阅关系、TPS、积累量、24h读写总量等信息

./mqadmin statsAll -n ip:9876 -t TopicTest
8.查看集群信息
./mqadmin clusterList -n ip:9876 -m
9.查看broker信息
./mqadmin brokerStatus -n ip:9876 -b ip:10911
10.根据id查询msg
./mqadmin queryMsgById -n ip:9876 -i msgId
11.发送消息
./mqadmin sendMessage -n ip:9876 -t TopicTest -p ‘this is a body content’ -c mqtest
12.消费消息
./mqadmin consumeMessage -n ip:9876 -t TopicTest
安装dashboard
提供了出色的监控功能。客户端和应用程序的各种事件、性能和系统信息的图表和统计明显地提供给用户

1.git地址
原来叫做rocketmq-console地址在https://github.com/apache/rocketmq-externals下的rocketmq-console

后改名叫rocketmq-dashboard新地址GitHub - apache/rocketmq-dashboard: The state-of-the-art Dashboard of Apache RoccketMQ provides excellent monitoring capability. Various graphs and statistics of events, performance and system information of clients and application is evidently made available to the user.

2.安装

1.下载源码
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5QJhXAgv-1666199870641)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019150414189.png)]

2.上传到服务器
rz -be
3.解压
unzip fileName
4.修改配置文件
进入rocketmq-dashboard-master/src/main/resources
vim application.properties
server.port=你要的端口号
namesrvAddr:你的namesrv地址
5.编译打包
mvn clean package -Dmaven.test.skip=true
#编译完成之后会在target目录生成jar包

6.创建启动脚本
vim start.sh
nohup java -jar rocketmq-dashboard-1.0.1-SNAPSHOT.jar >> rocketmq-dashboard.log &
#保存之后修改文件权限

chmod 744 start.sh
7.验证访问页面
http://ip:port
10.java客户端使用
使用例子可以在官网找

Simple Message Example - Apache RocketMQ
1.添加依赖

org.apache.rocketmq
rocketmq-client
4.9.3

2.生产消息
生产消息有3种模式

Send Messages Synchronously
Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS notification, SMS marketing system, etc…

Send Messages Asynchronously
Asynchronous transmission is generally used in response time sensitive business scenarios.

Send Messages in One-way Mode
One-way transmission is used for cases requiring moderate reliability, such as log collection.

这里以Send Messages Synchronously为例

private static DefaultMQProducer producer = null;

public SendResult producer(String topic, String tag, String messgae) {
// TODO Auto-generated method stub
SendResult sendResult = new SendResult();
try {
if (null == producer) {
producer = new DefaultMQProducer(“producer”);
// 设置NameServer的地址
producer.setNamesrvAddr(namesrvAddr);
// 启动Producer实例
producer.start();
}

    // 创建消息，并指定Topic，Tag和消息体
    Message msg = new Message(topic, tag, messgae.getBytes(RemotingHelper.DEFAULT_CHARSET));
    // 发送消息到一个Broker
    sendResult = producer.send(msg);
    LOGGER.info(sendResult);
} catch (Exception e) {
    // TODO: handle exception
    LOGGER.error("producer", e);
}
return sendResult;

}
3.消费消息
private void pushConsumer(String topic, String tag) {
​ // 实例化消费者
​ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(tag);
​ /**

 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
          * 如果非第一次启动，那么按照上次消费的位置继续消费
          */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置NameServer的地址
        consumer.setNamesrvAddr(namesrvAddr);
// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
try {
    consumer.subscribe(topic, tag);
    // 注册回调实现类来处理从broker拉取回来的消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
            try {
                for (MessageExt msg : msgs) {

                    String body = new String(msg.getBody());
                    LOGGER.info("这里可以加上你要拿消费到的body干什么用");
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            } catch (Exception e) {
                // TODO: handle exception
                LOGGER.error("", e);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
    });
    // 启动消费者实例
    consumer.start();
} catch (MQClientException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
}
}

2 分布式缓存
我们再做缓存之前需要把数据先分好类

按变化频率：

静态数据：一般不变的，类似于字典表

准静态数据：变化频率很低，部门结构设置，全国行政区划数据

中间状态数据：一些计算的可复用中间数据，变量副本，配置中心的本地副本

按使用频率：

热数据：使用频率高的

读写比大的：读的频率远大于写的频率

这些数据就比较适合使用缓存。

缓存无处不在。内存可以看作是cpu和磁盘之间的缓存。cpu与内存的处理速度也不一致，所以出现了L1&L2 Cache

缓存的本质：系统各级之间处理速度不匹配，利用空间换时间。

缓存加载时间

1. 启动时全量加载

2. 懒加载

2.1. 同步使用加载
先看缓存里是否有数据，没有的话从数据库读取。读取的数据，先放到内存，然后返回给调用方。

2.2. 延迟异步加载
从缓存里获取数据，不管有没有都直接返回。

策略1：如果缓存为空的话，则发起一个异步线程负责加载。

策略2：异步线程负责维护缓存的数据，定期或根据条件触发更新。

缓存过期策略

按FIFO或LRU

固定时间过期

根据业务进行时间的加权。

1、本地缓存
1.Map 缓存
public static final Map<String,Object> CACHE=new` `HashMap();CACHE.put("key","value");
2.Guava缓存
Cache<String,String> cache = CacheBuilder.newBuilder() .maximumSize(1024) .expireAfterWrite(60,TimeUnit.SECONDS) .weakValues() .build();cache.put(“word”,“Hello Guava Cache”);System.out.println(cache.getIfPresent("word"));
3.Spring Cache
基于注解和AOP，使用方便

可以配置Condition和SPEL，非常灵活

需要注意：绕过Spring的话，注解无效

核心功能：@Cacheable、@CachePut、@CacheEvict

本地缓存的缺点：
在集群环境中，如果每个节点都保存一份缓存，导致占用内存变大

在JVM中长期存在，会影响GC

缓存数据的调度处理，影响业务线程，争夺资源

2、远程缓存
1.Redis
Redis是一个开源的使用ANSI C语言编写的，基于内存也可以持久化的key-value数据库，并提供多种语言的API

1.什么是Redis？它主要用来什么的？
Redis，英文全称是Remote Dictionary Server（远程字典服务），是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。

与MySQL数据库不同的是，Redis的数据是存在内存中的。它的读写速度非常快，每秒可以处理超过10万次读写操作。因此redis被广泛应用于缓存，另外，Redis也经常用来做分布式锁。除此之外，Redis支持事务、持久化、LUA 脚本、LRU 驱动事件、多种集群方案。

2.说说Redis的基本数据结构类型
大多数小伙伴都知道，Redis有以下这五种基本类型：

String（字符串）
Hash（哈希）
List（列表）
Set（集合）
zset（有序集合）
它还有三种特殊的数据结构类型

Geospatial
Hyperloglog
Bitmap
2.1 Redis 的五种基本数据类型

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-pLTWoGQo-1666199870642)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019161630491.png)]

String（字符串）
简介:String是Redis最基础的数据结构类型，它是二进制安全的，可以存储图片或者序列化的对象，值最大存储为512M
简单使用举例: set key value、get key等
应用场景：共享session、分布式锁，计数器、限流。
内部编码有3种，int（8字节长整型）/embstr（小于等于39字节字符串）/raw（大于39个字节字符串）
C语言的字符串是char[]实现的，而Redis使用SDS（simple dynamic string） 封装，sds源码如下：

struct sdshdr{ unsigned int len; // 标记buf的长度 unsigned int free; //标记buf中未使用的元素个数 char buf[]; // 存放元素的坑 }

SDS 结构图如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cHM5SXdB-1666199870648)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019161644840.png)]

Redis为什么选择SDS结构，而C语言原生的char[]不香吗？

举例其中一点，SDS中，O(1)时间复杂度，就可以获取字符串长度；而C 字符串，需要遍历整个字符串，时间复杂度为O(n)
Hash（哈希）
简介：在Redis中，哈希类型是指v（值）本身又是一个键值对（k-v）结构
简单使用举例：hset key field value 、hget key field
内部编码：ziplist（压缩列表） 、hashtable（哈希表）
应用场景：缓存用户信息等。
注意点：如果开发使用hgetall，哈希元素比较多的话，可能导致Redis阻塞，可以使用hscan。而如果只是获取部分field，建议使用hmget。
字符串和哈希类型对比如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qXFPCWQG-1666199870649)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019161816953.png)]

List（列表）
简介：列表（list）类型是用来存储多个有序的字符串，一个列表最多可以存储2^32-1个元素。
简单实用举例：lpush key value [value …] 、lrange key start end
内部编码：ziplist（压缩列表）、linkedlist（链表）
应用场景：消息队列，文章列表,
一图看懂list类型的插入与弹出：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mC7TPsrw-1666199870654)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019161833183.png)]

list应用场景参考以下：

lpush+lpop=Stack（栈）
lpush+rpop=Queue（队列）
lpsh+ltrim=Capped Collection（有限集合）
lpush+brpop=Message Queue（消息队列）

Set（集合）
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FMXKMzIW-1666199870654)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019161847293.png)]

简介：集合（set）类型也是用来保存多个的字符串元素，但是不允许重复元素
简单使用举例：sadd key element [element …]、smembers key
内部编码：intset（整数集合）、hashtable（哈希表）
注意点：smembers和lrange、hgetall都属于比较重的命令，如果元素过多存在阻塞Redis的可能性，可以使用sscan来完成。
应用场景：用户标签,生成随机数抽奖、社交需求。

有序集合（zset）
简介：已排序的字符串集合，同时元素不能重复
简单格式举例：zadd key score member [score member …]，zrank key member
底层内部编码：ziplist（压缩列表）、skiplist（跳跃表）
应用场景：排行榜，社交需求（如用户点赞）。
2.2 Redis 的三种特殊数据类型
Geo：Redis3.2推出的，地理位置定位，用于存储地理位置信息，并对存储的信息进行操作。
HyperLogLog：用来做基数统计算法的数据结构，如统计网站的UV。
Bitmaps ：用一个比特位来映射某个元素的状态，在Redis中，它的底层是基于字符串类型实现的，可以把bitmaps成作一个以比特位为单位的数组

3.Redis为什么这么快？
Redis为什么这么快

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YWbEDYH6-1666199870660)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019161959445.png)]

3.1 基于内存存储实现
我们都知道内存读写是比在磁盘快很多的，Redis基于内存存储实现的数据库，相对于数据存在磁盘的MySQL数据库，省去磁盘I/O的消耗。

3.2 高效的数据结构
我们知道，Mysql索引为了提高效率，选择了B+树的数据结构。其实合理的数据结构，就是可以让你的应用/程序更快。先看下Redis的数据结构&内部编码图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PE9Tvpqv-1666199870661)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019162100789.png)]

SDS简单动态字符串

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bKeHbcKj-1666199870666)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019162111601.png)]

字符串长度处理：Redis获取字符串长度，时间复杂度为O(1)，而C语言中，需要从头开始遍历，复杂度为O（n）;
空间预分配：字符串修改越频繁的话，内存分配越频繁，就会消耗性能，而SDS修改和空间扩充，会额外分配未使用的空间，减少性能损耗。
惰性空间释放：SDS 缩短时，不是回收多余的内存空间，而是free记录下多余的空间，后续有变更，直接使用free中记录的空间，减少分配。
二进制安全：Redis可以存储一些二进制数据，在C语言中字符串遇到’’会结束，而 SDS中标志字符串结束的是len属性。
字典

Redis 作为 K-V 型内存数据库，所有的键值就是用字典来存储。字典就是哈希表，比如HashMap，通过key就可以直接获取到对应的value。而哈希表的特性，在O（1）时间复杂度就可以获得对应的值。

跳跃表

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zTjBrULx-1666199870669)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019162124296.png)]

跳跃表是Redis特有的数据结构，就是在链表的基础上，增加多级索引提升查找效率。
跳跃表支持平均 O（logN）,最坏 O（N）复杂度的节点查找，还可以通过顺序性操作批量处理节点。

3.3 合理的数据编码
Redis 支持多种数据数据类型，每种基本类型，可能对多种数据结构。什么时候,使用什么样数据结构，使用什么样编码，是redis设计者总结优化的结果。

String：如果存储数字的话，是用int类型的编码;如果存储非数字，小于等于39字节的字符串，是embstr；大于39个字节，则是raw编码。
List：如果列表的元素个数小于512个，列表每个元素的值都小于64字节（默认），使用ziplist编码，否则使用linkedlist编码
Hash：哈希类型元素个数小于512个，所有值小于64字节的话，使用ziplist编码,否则使用hashtable编码。
Set：如果集合中的元素都是整数且元素个数小于512个，使用intset编码，否则使用hashtable编码。
Zset：当有序集合的元素个数小于128个，每个元素的值小于64字节时，使用ziplist编码，否则使用skiplist（跳跃表）编码
3.4 合理的线程模型
I/O 多路复用

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0iLVrdFS-1666199870677)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019162226915.png)]

I/O 多路复用

多路I/O复用技术可以让单个线程高效的处理多个连接请求，而Redis使用用epoll作为I/O多路复用技术的实现。并且，Redis自身的事件处理模型将epoll中的连接、读写、关闭都转换为事件，不在网络I/O上浪费过多的时间。

什么是I/O多路复用？

I/O ：网络 I/O
多路 ：多个网络连接
复用：复用同一个线程。
IO多路复用其实就是一种同步IO模型，它实现了一个线程可以监视多个文件句柄；一旦某个文件句柄就绪，就能够通知应用程序进行相应的读写操作；而没有文件句柄就绪时,就会阻塞应用程序，交出cpu。
单线程模型
Redis是单线程模型

Redis是单线程模型的，而单线程避免了CPU不必要的上下文切换和竞争锁的消耗。也正因为是单线程，如果某个命令执行过长（如hgetall命令），会造成阻塞。Redis是面向快速执行场景的数据库。，所以要慎用如smembers和lrange、hgetall等命令。
Redis 6.0 引入了多线程提速，它的执行命令操作内存的仍然是个单线程。

3.5 虚拟内存机制
Redis直接自己构建了VM机制 ，不会像一般的系统会调用系统函数处理，会浪费一定的时间去移动和请求。

Redis的虚拟内存机制是啥呢？

虚拟内存机制就是暂时把不经常访问的数据(冷数据)从内存交换到磁盘中，从而腾出宝贵的内存空间用于其它需要访问的数据(热数据)。通过VM功能可以实现冷热数据分离，使热数据仍在内存中、冷数据保存到磁盘。这样就可以避免因为内存不足而造成访问速度下降的问题。

4.什么是缓存击穿、缓存穿透、缓存雪崩？
5.什么是热Key问题，如何解决热key问题
什么是热Key呢？在Redis中，我们把访问频率高的key，称为热点key。

如果某一热点key的请求到服务器主机时，由于请求量特别大，可能会导致主机资源不足，甚至宕机，从而影响正常的服务。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5j50hXm9-1666199870685)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019162832439.png)]

而热点Key是怎么产生的呢？主要原因有两个：
用户消费的数据远大于生产的数据，如秒杀、热点新闻等读多写少的场景。

请求分片集中，超过单Redi服务器的性能，比如固定名称key，Hash落入同一台服务器，瞬间访问量极大，超过机器瓶颈，产生热点Key问题。

那么在日常开发中，如何识别到热点key呢？

凭经验判断哪些是热Key；

客户端统计上报；

服务代理层上报

如何解决热key问题？
Redis集群扩容：增加分片副本，均衡读流量；
将热key分散到不同的服务器中；
使用二级缓存，即JVM本地缓存,减少Redis的读请求。
6.Redis 过期策略和内存淘汰策略
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Lq6Xc1zM-1666199870686)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019162845827.png)]

6.1 Redis的过期策略
我们在set key的时候，可以给它设置一个过期时间，比如expire key 60。指定这key60s后过期，60s后，redis是如何处理的嘛？我们先来介绍几种过期策略：

定时过期

每个设置过期时间的key都需要创建一个定时器，到过期时间就会立即对key进行清除。该策略可以立即清除过期的数据，对内存很友好；但是会占用大量的CPU资源去处理过期的数据，从而影响缓存的响应时间和吞吐量。
惰性过期

只有当访问一个key时，才会判断该key是否已过期，过期则清除。该策略可以最大化地节省CPU资源，却对内存非常不友好。极端情况可能出现大量的过期key没有再次被访问，从而不会被清除，占用大量内存。
定期过期

每隔一定的时间，会扫描一定数量的数据库的expires字典中一定数量的key，并清除其中已过期的key。该策略是前两者的一个折中方案。通过调整定时扫描的时间间隔和每次扫描的限定耗时，可以在不同情况下使得CPU和内存资源达到最优的平衡效果。
expires字典会保存所有设置了过期时间的key的过期时间数据，其中，key是指向键空间中的某个键的指针，value是该键的毫秒精度的UNIX时间戳表示的过期时间。键空间是指该Redis集群中保存的所有键。
Redis中同时使用了惰性过期和定期过期两种过期策略。
假设Redis当前存放30万个key，并且都设置了过期时间，如果你每隔100ms就去检查这全部的key，CPU负载会特别高，最后可能会挂掉。
因此，redis采取的是定期过期，每隔100ms就随机抽取一定数量的key来检查和删除的。
但是呢，最后可能会有很多已经过期的key没被删除。这时候，redis采用惰性删除。在你获取某个key的时候，redis会检查一下，这个key如果设置了过期时间并且已经过期了，此时就会删除。
但是呀，如果定期删除漏掉了很多过期的key，然后也没走惰性删除。就会有很多过期key积在内存内存，直接会导致内存爆的。或者有些时候，业务量大起来了，redis的key被大量使用，内存直接不够了，运维小哥哥也忘记加大内存了。难道redis直接这样挂掉？不会的！Redis用8种内存淘汰策略保护自己~

6.2 Redis 内存淘汰策略
volatile-lru：当内存不足以容纳新写入数据时，从设置了过期时间的key中使用LRU（最近最少使用）算法进行淘汰；

allkeys-lru：当内存不足以容纳新写入数据时，从所有key中使用LRU（最近最少使用）算法进行淘汰。

volatile-lfu：4.0版本新增，当内存不足以容纳新写入数据时，在过期的key中，使用LFU算法进行删除key。

allkeys-lfu：4.0版本新增，当内存不足以容纳新写入数据时，从所有key中使用LFU算法进行淘汰；

volatile-random：当内存不足以容纳新写入数据时，从设置了过期时间的key中，随机淘汰数据；。

allkeys-random：当内存不足以容纳新写入数据时，从所有key中随机淘汰数据。

volatile-ttl：当内存不足以容纳新写入数据时，在设置了过期时间的key中，根据过期时间进行淘汰，越早过期的优先被淘汰；

noeviction：默认策略，当内存不足以容纳新写入数据时，新写入操作会报错。

7.说说Redis的常用应用场景
缓存

排行榜

计数器应用

共享Session

分布式锁

社交网络

消息队列

位操作

7.1 缓存
我们一提到redis，自然而然就想到缓存，国内外中大型的网站都离不开缓存。合理的利用缓存，比如缓存热点数据，不仅可以提升网站的访问速度，还可以降低数据库DB的压力。并且，Redis相比于memcached，还提供了丰富的数据结构，并且提供RDB和AOF等持久化机制，强的一批。

7.2 排行榜
当今互联网应用，有各种各样的排行榜，如电商网站的月度销量排行榜、社交APP的礼物排行榜、小程序的投票排行榜等等。Redis提供的zset数据类型能够实现这些复杂的排行榜。

比如，用户每天上传视频，获得点赞的排行榜可以这样设计：

1.用户Jay上传一个视频，获得6个赞，可以酱紫：

zadd user:ranking:2021-03-03 Jay 3
2.过了一段时间，再获得一个赞，可以这样：

zincrby user:ranking:2021-03-03 Jay 1
3.如果某个用户John作弊，需要删除该用户：

zrem user:ranking:2021-03-03 John
4.展示获取赞数最多的3个用户

zrevrangebyrank user:ranking:2021-03-03 0 2
7.3 计数器应用
各大网站、APP应用经常需要计数器的功能，如短视频的播放数、电商网站的浏览数。这些播放数、浏览数一般要求实时的，每一次播放和浏览都要做加1的操作，如果并发量很大对于传统关系型数据的性能是一种挑战。Redis天然支持计数功能而且计数的性能也非常好，可以说是计数器系统的重要选择。

7.4 共享Session
如果一个分布式Web服务将用户的Session信息保存在各自服务器，用户刷新一次可能就需要重新登录了，这样显然有问题。实际上，可以使用Redis将用户的Session进行集中管理，每次用户更新或者查询登录信息都直接从Redis中集中获取。

7.5 分布式锁
几乎每个互联网公司中都使用了分布式部署，分布式服务下，就会遇到对同一个资源的并发访问的技术难题，如秒杀、下单减库存等场景。

用synchronize或者reentrantlock本地锁肯定是不行的。
如果是并发量不大话，使用数据库的悲观锁、乐观锁来实现没啥问题。
但是在并发量高的场合中，利用数据库锁来控制资源的并发访问，会影响数据库的性能。
实际上，可以用Redis的setnx来实现分布式的锁。

7.6 社交网络
赞/踩、粉丝、共同好友/喜好、推送、下拉刷新等是社交网站的必备功能，由于社交网站访问量通常比较大，而且传统的关系型数据不太适保存 这种类型的数据，Redis提供的数据结构可以相对比较容易地实现这些功能。

7.7 消息队列
消息队列是大型网站必用中间件，如ActiveMQ、RabbitMQ、Kafka等流行的消息队列中间件，主要用于业务解耦、流量削峰及异步处理实时性低的业务。Redis提供了发布/订阅及阻塞队列功能，能实现一个简单的消息队列系统。另外，这个不能和专业的消息中间件相比。

7.8 位操作
用于数据量上亿的场景下，例如几亿用户系统的签到，去重登录次数统计，某用户是否在线状态等等。腾讯10亿用户，要几个毫秒内查询到某个用户是否在线，能怎么做？千万别说给每个用户建立一个key，然后挨个记（你可以算一下需要的内存会很恐怖，而且这种类似的需求很多。这里要用到位操作——使用setbit、getbit、bitcount命令。原理是：redis内构建一个足够长的数组，每个数组元素只能是0和1两个值，然后这个数组的下标index用来表示用户id（必须是数字哈），那么很显然，这个几亿长的大数组就能通过下标和元素值（0和1）来构建一个记忆系统。

8.Redis 的持久化机制有哪些？优缺点说说
Redis是基于内存的非关系型K-V数据库，既然它是基于内存的，如果Redis服务器挂了，数据就会丢失。为了避免数据丢失了，Redis提供了持久化，即把数据保存到磁盘。

Redis提供了RDB和AOF两种持久化机制，它持久化文件加载流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dgyvHYjz-1666199870687)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163406962.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SpWPRbWW-1666199870690)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163421972.png)]

8.1 RDB
RDB，就是把内存数据以快照的形式保存到磁盘上。

什么是快照?可以这样理解，给当前时刻的数据，拍一张照片，然后保存下来。

RDB持久化，是指在指定的时间间隔内，执行指定次数的写操作，将内存中的数据集快照写入磁盘中，它是Redis默认的持久化方式。执行完操作后，在指定目录下会生成一个dump.rdb文件，Redis 重启的时候，通过加载dump.rdb文件来恢复数据。RDB触发机制主要有以下几种：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-WhV8Q4kf-1666199870692)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163440194.png)]

RDB 的优点
适合大规模的数据恢复场景，如备份，全量复制等

RDB缺点
没办法做到实时持久化/秒级持久化。
新老版本存在RDB格式兼容问题

AOF
AOF（append only file） 持久化，采用日志的形式来记录每个写操作，追加到文件中，重启时再重新执行AOF文件中的命令来恢复数据。它主要解决数据持久化的实时性问题。默认是不开启的。

AOF的工作流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qnrrYzqj-1666199870698)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163503752.png)]

AOF的优点
数据的一致性和完整性更高

AOF的缺点
AOF记录的内容越多，文件越大，数据恢复变慢。

9.怎么实现Redis的高可用？
我们在项目中使用Redis，肯定不会是单点部署Redis服务的。因为，单点部署一旦宕机，就不可用了。为了实现高可用，通常的做法是，将数据库复制多个副本以部署在不同的服务器上，其中一台挂了也可以继续提供服务。Redis 实现高可用有三种部署模式：主从模式，哨兵模式，集群模式。

9.1 主从模式
主从模式中，Redis部署了多台机器，有主节点，负责读写操作，有从节点，只负责读操作。从节点的数据来自主节点，实现原理就是主从复制机制

主从复制包括全量复制，增量复制两种。一般当slave第一次启动连接master，或者认为是第一次连接，就采用全量复制，全量复制流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QqPhrTS5-1666199870702)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163633769.png)]

1.slave发送sync命令到master。
2.master接收到SYNC命令后，执行bgsave命令，生成RDB全量文件。
3.master使用缓冲区，记录RDB快照生成期间的所有写命令。
4.master执行完bgsave后，向所有slave发送RDB快照文件。
5.slave收到RDB快照文件后，载入、解析收到的快照。
6.master使用缓冲区，记录RDB同步期间生成的所有写的命令。
7.master快照发送完毕后，开始向slave发送缓冲区中的写命令;
8.salve接受命令请求，并执行来自master缓冲区的写命令
redis2.8版本之后，已经使用psync来替代sync，因为sync命令非常消耗系统资源，psync的效率更高。
slave与master全量同步之后，master上的数据，如果再次发生更新，就会触发增量复制。

当master节点发生数据增减时，就会触发replicationFeedSalves()函数，接下来在 Master节点上调用的每一个命令会使用replicationFeedSlaves()来同步到Slave节点。执行此函数之前呢，master节点会判断用户执行的命令是否有数据更新，如果有数据更新的话，并且slave节点不为空，就会执行此函数。这个函数作用就是：把用户执行的命令发送到所有的slave节点，让slave节点执行。流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sumsxLQU-1666199870705)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163805143.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PGeKHbDi-1666199870708)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163836988.png)]

9.2 哨兵模式
主从模式中，一旦主节点由于故障不能提供服务，需要人工将从节点晋升为主节点，同时还要通知应用方更新主节点地址。显然，多数业务场景都不能接受这种故障处理方式。Redis从2.8开始正式提供了Redis Sentinel（哨兵）架构来解决这个问题。

哨兵模式，由一个或多个Sentinel实例组成的Sentinel系统，它可以监视所有的Redis主节点和从节点，并在被监视的主节点进入下线状态时，自动将下线主服务器属下的某个从节点升级为新的主节点。但是呢，一个哨兵进程对Redis节点进行监控，就可能会出现问题（单点问题），因此，可以使用多个哨兵来进行监控Redis节点，并且各个哨兵之间还会进行监控。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7UD5fZvG-1666199870709)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019163930254.png)]

Sentinel哨兵模式

简单来说，哨兵模式就三个作用：

发送命令，等待Redis服务器（包括主服务器和从服务器）返回监控其运行状态；

哨兵监测到主节点宕机，会自动将从节点切换成主节点，然后通过发布订阅模式通知其他的从节点，修改配置文件，让它们切换主机；

哨兵之间还会相互监控，从而达到高可用。

故障切换的过程是怎样的呢

假设主服务器宕机，哨兵1先检测到这个结果，系统并不会马上进行 failover 过程，仅仅是哨兵1主观的认为主服务器不可用，这个现象成为主观下线。当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行 failover 操作。切换成功后，就会通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为客观下线。这样对于客户端而言，一切都是透明的。
哨兵的工作模式如下：

每个Sentinel以每秒钟一次的频率向它所知的Master，Slave以及其他Sentinel实例发送一个 PING命令。

如果一个实例（instance）距离最后一次有效回复 PING 命令的时间超过 down-after-milliseconds 选项所指定的值， 则这个实例会被 Sentinel标记为主观下线。

如果一个Master被标记为主观下线，则正在监视这个Master的所有 Sentinel 要以每秒一次的频率确认Master的确进入了主观下线状态。

当有足够数量的 Sentinel（大于等于配置文件指定的值）在指定的时间范围内确认Master的确进入了主观下线状态， 则Master会被标记为客观下线。

在一般情况下， 每个 Sentinel 会以每10秒一次的频率向它已知的所有Master，Slave发送 INFO 命令。

当Master被 Sentinel 标记为客观下线时，Sentinel 向下线的 Master 的所有 Slave 发送 INFO 命令的频率会从 10 秒一次改为每秒一次

若没有足够数量的 Sentinel同意Master已经下线， Master的客观下线状态就会被移除；若Master 重新向 Sentinel 的 PING 命令返回有效回复， Master 的主观下线状态就会被移除。

9.3 Cluster集群模式
哨兵模式基于主从模式，实现读写分离，它还可以自动切换，系统可用性更高。但是它每个节点存储的数据是一样的，浪费内存，并且不好在线扩容。因此，Cluster集群应运而生，它在Redis3.0加入的，实现了Redis的分布式存储。对数据进行分片，也就是说每台Redis节点上存储不同的内容，来解决在线扩容的问题。并且，它也提供复制和故障转移的功能。

Cluster集群节点的通讯

一个Redis集群由多个节点组成，各个节点之间是怎么通信的呢？通过Gossip协议！

Redis Cluster集群通过Gossip协议进行通信，节点之前不断交换信息，交换的信息内容包括节点出现故障、新节点加入、主从节点变更信息、slot信息等等。常用的Gossip消息分为4种，分别是：ping、pong、meet、fail。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DrfK9DTy-1666199870711)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019164259061.png)]

meet消息：通知新节点加入。消息发送者通知接收者加入到当前集群，meet消息通信正常完成后，接收节点会加入到集群中并进行周期性的ping、pong消息交换。

ping消息：集群内交换最频繁的消息，集群内每个节点每秒向多个其他节点发送ping消息，用于检测节点是否在线和交换彼此状态信息。

pong消息：当接收到ping、meet消息时，作为响应消息回复给发送方确认消息正常通信。pong消息内部封装了自身状态数据。节点也可以向集群内广播自身的pong消息来通知整个集群对自身状态进行更新。

fail消息：当节点判定集群内另一个节点下线时，会向集群内广播一个fail消息，其他节点接收到fail消息之后把对应节点更新为下线状态。

特别的，每个节点是通过集群总线(cluster bus) 与其他的节点进行通信的。通讯时，使用特殊的端口号，即对外服务端口号加10000。例如如果某个node的端口号是6379，那么它与其它nodes通信的端口号是 16379。nodes 之间的通信采用特殊的二进制协议。

Hash Slot插槽算法

既然是分布式存储，Cluster集群使用的分布式算法是一致性Hash嘛？并不是，而是Hash Slot插槽算法。

插槽算法把整个数据库被分为16384个slot（槽），每个进入Redis的键值对，根据key进行散列，分配到这16384插槽中的一个。使用的哈希映射也比较简单，用CRC16算法计算出一个16 位的值，再对16384取模。数据库中的每个键都属于这16384个槽的其中一个，集群中的每个节点都可以处理这16384个槽。

集群中的每个节点负责一部分的hash槽，比如当前集群有A、B、C个节点，每个节点上的哈希槽数 =16384/3，那么就有：

节点A负责0~5460号哈希槽
节点B负责5461~10922号哈希槽
节点C负责10923~16383号哈希槽

Redis Cluster集群

Redis Cluster集群中，需要确保16384个槽对应的node都正常工作，如果某个node出现故障，它负责的slot也会失效，整个集群将不能工作。

因此为了保证高可用，Cluster集群引入了主从复制，一个主节点对应一个或者多个从节点。当其它主节点 ping 一个主节点 A 时，如果半数以上的主节点与 A 通信超时，那么认为主节点 A 宕机了。如果主节点宕机时，就会启用从节点。

在Redis的每一个节点上，都有两个玩意，一个是插槽（slot），它的取值范围是0~16383。另外一个是cluster，可以理解为一个集群管理的插件。当我们存取的key到达时，Redis 会根据CRC16算法得出一个16 bit的值，然后把结果对16384取模。酱紫每个key都会对应一个编号在 0~16383 之间的哈希槽，通过这个值，去找到对应的插槽所对应的节点，然后直接自动跳转到这个对应的节点上进行存取操作。

虽然数据是分开存储在不同节点上的，但是对客户端来说，整个集群Cluster，被看做一个整体。客户端端连接任意一个node，看起来跟操作单实例的Redis一样。当客户端操作的key没有被分配到正确的node节点时，Redis会返回转向指令，最后指向正确的node，这就有点像浏览器页面的302 重定向跳转。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xtuCAgcV-1666199870726)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019164548896.png)]

故障转移

Redis集群实现了高可用，当集群内节点出现故障时，通过故障转移，以保证集群正常对外提供服务。

redis集群通过ping/pong消息，实现故障发现。这个环境包括主观下线和客观下线。

主观下线： 某个节点认为另一个节点不可用，即下线状态，这个状态并不是最终的故障判定，只能代表一个节点的意见，可能存在误判情况。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zucBDBHg-1666199870732)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019164602125.png)]

主观下线

客观下线： 指标记一个节点真正的下线，集群内多个节点都认为该节点不可用，从而达成共识的结果。如果是持有槽的主节点故障，需要为该节点进行故障转移。

假如节点A标记节 点B为主观下线，一段时间后，节点A通过消息把节点B的状态发到其它节点，当节点C接受到消息并解析出消息体时，如果发现节点B的pfail状态时，会触发客观下线流程；
当下线为主节点时，此时Redis Cluster集群为统计持有槽的主节点投票，看投票数是否达到一半，当下线报告统计数大于一半时，被标记为客观下线状态。
流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QkTd2gKk-1666199870734)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019164615203.png)]

客观下线

故障恢复：故障发现后，如果下线节点的是主节点，则需要在它的从节点中选一个替换它，以保证集群的高可用。流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6XWsNAIP-1666199870738)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019164626673.png)]

资格检查：检查从节点是否具备替换故障主节点的条件。

准备选举时间：资格检查通过后，更新触发故障选举时间。

发起选举：到了故障选举时间，进行选举。

选举投票：只有持有槽的主节点才有票，从节点收集到足够的选票（大于一半），触发替换主节点操作

10.使用过Redis分布式锁嘛？有哪些注意点呢？
分布式锁，是控制分布式系统不同进程共同访问共享资源的一种锁的实现。秒杀下单、抢红包等等业务场景，都需要用到分布式锁，我们项目中经常使用Redis作为分布式锁。

选了Redis分布式锁的几种实现方法，大家来讨论下，看有没有啥问题哈。

命令setnx + expire分开写

setnx + value值是过期时间

set的扩展命令（set ex px nx）

set ex px nx + 校验唯一随机值,再删除

10.1 命令setnx + expire分开写
if（jedis.setnx(key,lock_value) == 1）{ //加锁 expire（key，100）; //设置过期时间 try { do something //业务请求 }catch(){ } finally { jedis.del(key); //释放锁 } }

如果执行完setnx加锁，正要执行expire设置过期时间时，进程crash掉或者要重启维护了，那这个锁就“长生不老”了，别的线程永远获取不到锁啦，所以分布式锁不能这么实现。

10.2 setnx + value值是过期时间
long expires = System.currentTimeMillis() + expireTime; //系统时间+设置的过期时间 String expiresStr = String.valueOf(expires); // 如果当前锁不存在，返回加锁成功 if (jedis.setnx(key, expiresStr) == 1) { return true; } // 如果锁已经存在，获取锁的过期时间 String currentValueStr = jedis.get(key); // 如果获取到的过期时间，小于系统当前时间，表示已经过期 if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) { // 锁已过期，获取上一个锁的过期时间，并设置现在锁的过期时间（不了解redis的getSet命令的小伙伴，可以去官网看下哈） String oldValueStr = jedis.getSet(key_resource_id, expiresStr); if (oldValueStr != null && oldValueStr.equals(currentValueStr)) { // 考虑多线程并发的情况，只有一个线程的设置值和当前值相同，它才可以加锁 return true; } } //其他情况，均返回加锁失败 return false; }

笔者看过有开发小伙伴是这么实现分布式锁的，但是这种方案也有这些缺点：

过期时间是客户端自己生成的，分布式环境下，每个客户端的时间必须同步。
没有保存持有者的唯一标识，可能被别的客户端释放/解锁。
锁过期的时候，并发多个客户端同时请求过来，都执行了jedis.getSet()，最终只能有一个客户端加锁成功，但是该客户端锁的过期时间，可能被别的客户端覆盖。

10.3：set的扩展命令（set ex px nx）（注意可能存在的问题）
if（jedis.set(key, lock_value, “NX”, “EX”, 100s) == 1）{ //加锁 try { do something //业务处理 }catch(){ } finally { jedis.del(key); //释放锁 } }

这个方案可能存在这样的问题：

锁过期释放了，业务还没执行完。
锁被别的线程误删。
10.4 set ex px nx + 校验唯一随机值,再删除
if（jedis.set(key, uni_request_id, “NX”, “EX”, 100s) == 1）{ //加锁 try { do something //业务处理 }catch(){ } finally { //判断是不是当前线程加的锁,是才释放 if (uni_request_id.equals(jedis.get(key))) { jedis.del(key); //释放锁 } } }

在这里，判断当前线程加的锁和释放锁是不是一个原子操作。如果调用jedis.del()释放锁的时候，可能这把锁已经不属于当前客户端，会解除他人加的锁。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Ed4laCd7-1666199870742)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019164932834.png)]

一般也是用lua脚本代替。lua脚本如下：

if redis.call(‘get’,KEYS[1]) == ARGV[1] then return redis.call(‘del’,KEYS[1]) else return 0 end;

这种方式比较不错了，一般情况下，已经可以使用这种实现方式。但是存在锁过期释放了，业务还没执行完的问题（实际上，估算个业务处理的时间，一般没啥问题了）。

11.使用过Redisson嘛？说说它的原理
分布式锁可能存在锁过期释放，业务没执行完的问题。有些小伙伴认为，稍微把锁过期时间设置长一些就可以啦。其实我们设想一下，是否可以给获得锁的线程，开启一个定时守护线程，每隔一段时间检查锁是否还存在，存在则对锁的过期时间延长，防止锁过期提前释放。

当前开源框架Redisson就解决了这个分布式锁问题。我们一起来看下Redisson底层原理是怎样的吧：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-69jvD40e-1666199870748)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165048408.png)]

只要线程一加锁成功，就会启动一个watch dog看门狗，它是一个后台线程，会每隔10秒检查一下，如果线程1还持有锁，那么就会不断的延长锁key的生存时间。因此，Redisson就是使用Redisson解决了锁过期释放，业务没执行完问题。

12.什么是Redlock算法
Redis一般都是集群部署的，假设数据在主从同步过程，主节点挂了，Redis分布式锁可能会有哪些问题呢？一起来看些这个流程图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-q8VsIplX-1666199870765)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165059765.png)]

如果线程一在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时，master节点发生故障，一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦，但线程一也已经拿到锁了，锁的安全性就没了。

为了解决这个问题，Redis作者 antirez提出一种高级的分布式锁算法：Redlock。Redlock核心思想是这样的：

搞多个Redis master部署，以保证它们不会同时宕掉。并且这些master节点是完全相互独立的，相互之间不存在数据同步。同时，需要确保在这多个master实例上，是与在Redis单实例，使用相同方法来获取和释放锁。

我们假设当前有5个Redis master节点，在5台服务器上面运行这些Redis实例。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-K9rPdknY-1666199870768)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165111523.png)]

RedLock的实现步骤:如下

1.获取当前时间，以毫秒为单位。

2.按顺序向5个master节点请求加锁。客户端设置网络连接和响应超时时间，并且超时时间要小于锁的失效时间。（假设锁自动失效时间为10秒，则超时时间一般在5-50毫秒之间,我们就假设超时时间是50ms吧）。如果超时，跳过该master节点，尽快去尝试下一个master节点。

3.客户端使用当前时间减去开始获取锁时间（即步骤1记录的时间），得到获取锁使用的时间。当且仅当超过一半（N/2+1，这里是5/2+1=3个节点）的Redis master节点都获得锁，并且使用的时间小于锁失效时间时，锁才算获取成功。（如上图，10s> 30ms+40ms+50ms+4m0s+50ms）

如果取到了锁，key的真正有效时间就变啦，需要减去获取锁所使用的时间。

如果获取锁失败（没有在至少N/2+1个master实例取到锁，有或者获取锁时间已经超过了有效时间），客户端要在所有的master节点上解锁（即便有些master节点根本就没有加锁成功，也需要解锁，以防止有些漏网之鱼）。

简化下步骤就是：

按顺序向5个master节点请求加锁

根据设置的超时时间来判断，是不是要跳过该master节点。

如果大于等于三个节点加锁成功，并且使用的时间小于锁的有效期，即可认定加锁成功啦。

如果获取锁失败，解锁！

13.Redis的跳跃表
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vruexswh-1666199870770)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165233675.png)]

跳跃表

跳跃表是有序集合zset的底层实现之一
跳跃表支持平均O（logN）,最坏 O（N）复杂度的节点查找，还可以通过顺序性操作批量处理节点。
跳跃表实现由zskiplist和zskiplistNode两个结构组成，其中zskiplist用于保存跳跃表信息（如表头节点、表尾节点、长度），而zskiplistNode则用于表示跳跃表节点。
跳跃表就是在链表的基础上，增加多级索引提升查找效率。

13.MySQL与Redis 如何保证双写一致性
缓存延时双删

删除缓存重试机制

读取biglog异步删除缓存

14.1 延时双删？
什么是延时双删呢？流程图如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gWbDaQeP-1666199870773)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165323371.png)]

延时双删流程

先删除缓存
再更新数据库
休眠一会（比如1秒），再次删除缓存。
这个休眠一会，一般多久呢？都是1秒？

这个休眠时间 = 读业务逻辑数据的耗时 + 几百毫秒。为了确保读请求结束，写请求可以删除读请求可能带来的缓存脏数据。

这种方案还算可以，只有休眠那一会（比如就那1秒），可能有脏数据，一般业务也会接受的。但是如果第二次删除缓存失败呢？缓存和数据库的数据还是可能不一致，对吧？给Key设置一个自然的expire过期时间，让它自动过期怎样？那业务要接受过期时间内，数据的不一致咯？还是有其他更佳方案呢？

14.2 删除缓存重试机制
因为延时双删可能会存在第二步的删除缓存失败，导致的数据不一致问题。可以使用这个方案优化：删除失败就多删除几次呀,保证删除缓存成功就可以了呀~ 所以可以引入删除缓存重试机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VHfmvgNi-1666199870783)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165338183.png)]

删除缓存重试流程

写请求更新数据库
缓存因为某些原因，删除失败
把删除失败的key放到消息队列
消费消息队列的消息，获取要删除的key
重试删除缓存操作
14.3 读取biglog异步删除缓存
重试删除缓存机制还可以吧，就是会造成好多业务代码入侵。其实，还可以这样优化：通过数据库的binlog来异步淘汰key。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VfYhNDfF-1666199870784)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165350979.png)]

以mysql为例吧

可以使用阿里的canal将binlog日志采集发送到MQ队列里面
然后通过ACK机制确认处理这条更新消息，删除缓存，保证数据缓存一致性

15.为什么Redis 6.0 之后改多线程呢？
Redis6.0之前，Redis在处理客户端的请求时，包括读socket、解析、执行、写socket等都由一个顺序串行的主线程处理，这就是所谓的“单线程”。
Redis6.0之前为什么一直不使用多线程？使用Redis时，几乎不存在CPU成为瓶颈的情况， Redis主要受限于内存和网络。例如在一个普通的Linux系统上，Redis通过使用pipelining每秒可以处理100万个请求，所以如果应用程序主要使用O(N)或O(log(N))的命令，它几乎不会占用太多CPU。
redis使用多线程并非是完全摒弃单线程，redis还是使用单线程模型来处理客户端的请求，只是使用多线程来处理数据的读写和协议解析，执行命令还是使用单线程。

这样做的目的是因为redis的性能瓶颈在于网络IO而非CPU，使用多线程能提升IO读写的效率，从而整体提高redis的性能。

16.聊聊Redis 事务机制
Redis通过MULTI、EXEC、WATCH等一组命令集合，来实现事务机制。事务支持一次执行多个命令，一个事务中所有命令都会被序列化。在事务执行过程，会按照顺序串行化执行队列中的命令，其他客户端提交的命令请求不会插入到事务执行命令序列中。

简言之，Redis事务就是顺序性、一次性、排他性的执行一个队列中的一系列命令。

Redis执行事务的流程如下：

开始事务（MULTI）

命令入队

执行事务（EXEC）、撤销事务（DISCARD ）

Redis的Hash 冲突怎么办
Redis 作为一个K-V的内存数据库，它使用用一张全局的哈希来保存所有的键值对。这张哈希表，有多个哈希桶组成，哈希桶中的entry元素保存了key和value指针，其中key指向了实际的键，value指向了实际的值。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KK1qHfxK-1666199870789)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165450384.png)]

哈希表查找速率很快的，有点类似于Java中的HashMap，它让我们在O(1) 的时间复杂度快速找到键值对。首先通过key计算哈希值，找到对应的哈希桶位置，然后定位到entry，在entry找到对应的数据。

什么是哈希冲突？

哈希冲突：通过不同的key，计算出一样的哈希值，导致落在同一个哈希桶中。

Redis为了解决哈希冲突，采用了链式哈希。链式哈希是指同一个哈希桶中，多个元素用一个链表来保存，它们之间依次用指针连接。

有些读者可能还会有疑问：哈希冲突链上的元素只能通过指针逐一查找再操作。当往哈希表插入数据很多，冲突也会越多，冲突链表就会越长，那查询效率就会降低了。

为了保持高效，Redis 会对哈希表做rehash操作，也就是增加哈希桶，减少冲突。为了rehash更高效，Redis还默认使用了两个全局哈希表，一个用于当前使用，称为主哈希表，一个用于扩容，称为备用哈希表。

18.在生成 RDB期间，Redis 可以同时处理写请求么？
可以的，Redis提供两个指令生成RDB，分别是save和bgsave。

如果是save指令，会阻塞，因为是主线程执行的。
如果是bgsave指令，是fork一个子进程来写入RDB文件的，快照持久化完全交给子进程来处理，父进程则可以继续处理客户端的请求。

19.Redis底层，使用的什么协议?
RESP，英文全称是Redis Serialization Protocol,它是专门为redis设计的一套序列化协议. 这个协议其实在redis的1.2版本时就已经出现了,但是到了redis2.0才最终成为redis通讯协议的标准。

RESP主要有实现简单、解析速度快、可读性好等优点。

20.布隆过滤器
应对缓存穿透问题，我们可以使用布隆过滤器。布隆过滤器是什么呢？

布隆过滤器是一种占用空间很小的数据结构，它由一个很长的二进制向量和一组Hash映射函数组成，它用于检索一个元素是否在一个集合中，空间效率和查询时间都比一般的算法要好的多，缺点是有一定的误识别率和删除困难。

布隆过滤器原理是？假设我们有个集合A，A中有n个元素。利用k个哈希散列函数，将A中的每个元素映射到一个长度为a位的数组B中的不同位置上，这些位置上的二进制数均设置为1。如果待检查的元素，经过这k个哈希散列函数的映射后，发现其k个位置上的二进制数全部为1，这个元素很可能属于集合A，反之，一定不属于集合A。

来看个简单例子吧，假设集合A有3个元素，分别为{d1,d2,d3}。有1个哈希函数，为Hash1。现在将A的每个元素映射到长度为16位数组B。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TqzCnhty-1666199870804)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165527441.png)]

我们现在把d1映射过来，假设Hash1（d1）= 2，我们就把数组B中，下标为2的格子改成1，如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Pf1jCAyK-1666199870805)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165534469.png)]

我们现在把d2也映射过来，假设Hash1（d2）= 5，我们把数组B中，下标为5的格子也改成1，如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4XC6f51n-1666199870810)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165541800.png)]

接着我们把d3也映射过来，假设Hash1（d3）也等于 2，它也是把下标为2的格子标1：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nuXZfiVP-1666199870811)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165548537.png)]

因此，我们要确认一个元素dn是否在集合A里，我们只要算出Hash1（dn）得到的索引下标，只要是0，那就表示这个元素不在集合A，如果索引下标是1呢？那该元素可能是A中的某一个元素。因为你看，d1和d3得到的下标值，都可能是1，还可能是其他别的数映射的，布隆过滤器是存在这个缺点的：会存在hash碰撞导致的假阳性，判断存在误差。

如何减少这种误差呢？

搞多几个哈希函数映射，降低哈希碰撞的概率
同时增加B数组的bit长度，可以增大hash函数生成的数据的范围，也可以降低哈希碰撞的概率
我们又增加一个Hash2哈希映射函数，假设Hash2（d1）=6,Hash2（d3）=8,它俩不就不冲突了嘛，如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EU0qtdFW-1666199870815)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019165600361.png)]

即使存在误差，我们可以发现，布隆过滤器并没有存放完整的数据，它只是运用一系列哈希映射函数计算出位置，然后填充二进制向量。如果数量很大的话，布隆过滤器通过极少的错误率，换取了存储空间的极大节省，还是挺划算的。

目前布隆过滤器已经有相应实现的开源类库啦，如Google的Guava类库，Twitter的 Algebird 类库，信手拈来即可，或者基于Redis自带的Bitmaps自行实现设计也是可以的。

2.Elasticsearch

1. 什么是Elasticsearch？
   Elasticsearch 是一个基于 Lucene 的搜索引擎。它提供了具有 HTTP Web 界面和无架构 JSON 文档的分布式，多租户能力的全文搜索引擎。 Elasticsearch 是用 Java 开发的，根据 Apache 许可条款作为开源发布。

2. ES中的倒排索引是什么？
   传统的检索方式是通过文章，逐个遍历找到对应关键词的位置。 倒排索引，是通过分词策略，形成了词和文章的映射关系表，也称倒排表，这种词典 + 映射表即为倒排索引。

其中词典中存储词元，倒排表中存储该词元在哪些文中出现的位置。 有了倒排索引，就能实现 O(1) 时间复杂度的效率检索文章了，极大的提高了检索效率。

加分项： 倒排索引的底层实现是基于：FST（Finite State Transducer）数据结构。

Lucene 从 4+ 版本后开始大量使用的数据结构是 FST。FST 有两个优点： 1）空间占用小。通过对词典中单词前缀和后缀的重复利用，压缩了存储空间； 2）查询速度快。O(len(str)) 的查询时间复杂度。

3. ES是如何实现master选举的？
   前置条件： 1）只有是候选主节点（master：true）的节点才能成为主节点。 2）最小主节点数（min_master_nodes）的目的是防止脑裂。

Elasticsearch 的选主是 ZenDiscovery 模块负责的，主要包含 Ping（节点之间通过这个RPC来发现彼此）和 Unicast（单播模块包含一个主机列表以控制哪些节点需要 ping 通）这两部分； 获取主节点的核心入口为 findMaster，选择主节点成功返回对应 Master，否则返回 null。

选举流程大致描述如下： 第一步：确认候选主节点数达标，elasticsearch.yml 设置的值 discovery.zen.minimum_master_nodes; 第二步：对所有候选主节点根据nodeId字典排序，每次选举每个节点都把自己所知道节点排一次序，然后选出第一个（第0位）节点，暂且认为它是master节点。 第三步：如果对某个节点的投票数达到一定的值（候选主节点数n/2+1）并且该节点自己也选举自己，那这个节点就是master。否则重新选举一直到满足上述条件。

补充：

这里的 id 为 string 类型。

master 节点的职责主要包括集群、节点和索引的管理，不负责文档级别的管理；data 节点可以关闭 http 功能。

4. 如何解决ES集群的脑裂问题
   所谓集群脑裂，是指 Elasticsearch 集群中的节点（比如共 20 个），其中的 10 个选了一个 master，另外 10 个选了另一个 master 的情况。

当集群 master 候选数量不小于 3 个时，可以通过设置最少投票通过数量（discovery.zen.minimum_master_nodes）超过所有候选节点一半以上来解决脑裂问题； 当候选数量为两个时，只能修改为唯一的一个 master 候选，其他作为 data 节点，避免脑裂问题。

5. 详细描述一下ES索引文档的过程？
   这里的索引文档应该理解为文档写入 ES，创建索引的过程。

第一步：客户端向集群某节点写入数据，发送请求。（如果没有指定路由/协调节点，请求的节点扮演协调节点的角色。） 第二步：协调节点接受到请求后，默认使用文档 ID 参与计算（也支持通过 routing），得到该文档属于哪个分片。随后请求会被转到另外的节点。

路由算法：根据文档id或路由计算目标的分片id

shard = hash(document_id) % (num_of_primary_shards)
复制

第三步：当分片所在的节点接收到来自协调节点的请求后，会将请求写入到 Memory Buffer，然后定时（默认是每隔 1 秒）写入到F ilesystem Cache，这个从 Momery Buffer 到 Filesystem Cache 的过程就叫做 refresh； 第四步：当然在某些情况下，存在 Memery Buffer 和 Filesystem Cache 的数据可能会丢失，ES 是通过 translog 的机制来保证数据的可靠性的。其实现机制是接收到请求后，同时也会写入到 translog 中，当 Filesystem cache 中的数据写入到磁盘中时，才会清除掉，这个过程叫做 flush； 第五步：在 flush 过程中，内存中的缓冲将被清除，内容被写入一个新段，段的 fsync 将创建一个新的提交点，并将内容刷新到磁盘，旧的 translog 将被删除并开始一个新的 translog。 第六步：flush 触发的时机是定时触发（默认 30 分钟）或者 translog 变得太大（默认为 512 M）时。

img

补充：关于 Lucene 的 Segement

Lucene 索引是由多个段组成，段本身是一个功能齐全的倒排索引。

段是不可变的，允许 Lucene 将新的文档增量地添加到索引中，而不用从头重建索引。

对于每一个搜索请求而言，索引中的所有段都会被搜索，并且每个段会消耗 CPU 的时钟周、文件句柄和内存。这意味着段的数量越多，搜索性能会越低。

为了解决这个问题，Elasticsearch 会合并小段到一个较大的段，提交新的合并段到磁盘，并删除那些旧的小段。（段合并）

6. 详细描述一下ES更新和删除文档的过程？
   删除和更新也都是写操作，但是 Elasticsearch 中的文档是不可变的，因此不能被删除或者改动以展示其变更。

磁盘上的每个段都有一个相应的 .del 文件。当删除请求发送后，文档并没有真的被删除，而是在 .del 文件中被标记为删除。该文档依然能匹配查询，但是会在结果中被过滤掉。当段合并时，在 .del 文件中被标记为删除的文档将不会被写入新段。

在新的文档被创建时，Elasticsearch 会为该文档指定一个版本号，当执行更新时，旧版本的文档在 .del 文件中被标记为删除，新版本的文档被索引到一个新段。旧版本的文档依然能匹配查询，但是会在结果中被过滤掉。

7. 详细描述一下ES搜索的过程？
   搜索被执行成一个两阶段过程，即 Query Then Fetch； Query阶段： 查询会广播到索引中每一个分片拷贝（主分片或者副本分片）。每个分片在本地执行搜索并构建一个匹配文档的大小为 from + size 的优先队列。PS：在搜索的时候是会查询Filesystem Cache的，但是有部分数据还在Memory Buffer，所以搜索是近实时的。 每个分片返回各自优先队列中 所有文档的 ID 和排序值 给协调节点，它合并这些值到自己的优先队列中来产生一个全局排序后的结果列表。 Fetch阶段： 协调节点辨别出哪些文档需要被取回并向相关的分片提交多个 GET 请求。每个分片加载并 丰富 文档，如果有需要的话，接着返回文档给协调节点。一旦所有的文档都被取回了，协调节点返回结果给客户端。

img

8. 在并发情况下，ES如果保证读写一致？
   可以通过版本号使用乐观并发控制，以确保新版本不会被旧版本覆盖，由应用层来处理具体的冲突； 另外对于写操作，一致性级别支持quorum/one/all，默认为quorum，即只有当大多数分片可用时才允许写操作。但即使大多数可用，也可能存在因为网络等原因导致写入副本失败，这样该副本被认为故障，分片将会在一个不同的节点上重建。 对于读操作，可以设置replication为sync(默认)，这使得操作在主分片和副本分片都完成后才会返回；如果设置replication为async时，也可以通过设置搜索请求参数_preference为primary来查询主分片，确保文档是最新版本。

9. ES对于大数据量（上亿量级）的聚合如何实现？
   Elasticsearch 提供的首个近似聚合是cardinality 度量。它提供一个字段的基数，即该字段的distinct或者unique值的数目。它是基于HLL算法的。HLL 会先对我们的输入作哈希运算，然后根据哈希运算的结果中的 bits 做概率估算从而得到基数。其特点是：可配置的精度，用来控制内存的使用（更精确 ＝ 更多内存）；小的数据集精度是非常高的；我们可以通过配置参数，来设置去重需要的固定内存使用量。无论数千还是数十亿的唯一值，内存使用量只与你配置的精确度相关。

10. 对于GC方面，在使用ES时要注意什么？
    1）倒排词典的索引需要常驻内存，无法GC，需要监控data node上segment memory增长趋势。 2）各类缓存，field cache, filter cache, indexing cache, bulk queue等等，要设置合理的大小，并且要应该根据最坏的情况来看heap是否够用，也就是各类缓存全部占满的时候，还有heap空间可以分配给其他任务吗？避免采用clear cache等“自欺欺人”的方式来释放内存。 3）避免返回大量结果集的搜索与聚合。确实需要大量拉取数据的场景，可以采用scan & scroll api来实现。 4）cluster stats驻留内存并无法水平扩展，超大规模集群可以考虑分拆成多个集群通过tribe node连接。 5）想知道heap够不够，必须结合实际应用场景，并对集群的heap使用情况做持续的监控。

11. 说说你们公司ES的集群架构，索引数据大小，分片有多少，以及一些调优手段？
    根据实际情况回答即可，如果是我的话会这么回答：
    我司有多个ES集群，下面列举其中一个。该集群有20个节点，根据数据类型和日期分库，每个索引根据数据量分片，比如日均1亿+数据的，控制单索引大小在200GB以内。　
    下面重点列举一些调优策略，仅是我做过的，不一定全面，如有其它建议或者补充欢迎留言。
    部署层面：
    1）最好是64GB内存的物理机器，但实际上32GB和16GB机器用的比较多，但绝对不能少于8G，除非数据量特别少，这点需要和客户方面沟通并合理说服对方。
    2）多个内核提供的额外并发远胜过稍微快一点点的时钟频率。
    3）尽量使用SSD，因为查询和索引性能将会得到显著提升。
    4）避免集群跨越大的地理距离，一般一个集群的所有节点位于一个数据中心中。
    5）设置堆内存：节点内存/2，不要超过32GB。一般来说设置export ES_HEAP_SIZE=32g环境变量，比直接写-Xmx32g -Xms32g更好一点。
    6）关闭缓存swap。内存交换到磁盘对服务器性能来说是致命的。如果内存交换到磁盘上，一个100微秒的操作可能变成10毫秒。 再想想那么多10微秒的操作时延累加起来。不难看出swapping对于性能是多么可怕。
    7）增加文件描述符，设置一个很大的值，如65535。Lucene使用了大量的文件，同时，Elasticsearch在节点和HTTP客户端之间进行通信也使用了大量的套接字。所有这一切都需要足够的文件描述符。
    8）不要随意修改垃圾回收器（CMS）和各个线程池的大小。
    9）通过设置gateway.recover_after_nodes、gateway.expected_nodes、gateway.recover_after_time可以在集群重启的时候避免过多的分片交换，这可能会让数据恢复从数个小时缩短为几秒钟。
    索引层面：
    1）使用批量请求并调整其大小：每次批量数据 5–15 MB 大是个不错的起始点。
    2）段合并：Elasticsearch默认值是20MB/s，对机械磁盘应该是个不错的设置。如果你用的是SSD，可以考虑提高到100-200MB/s。如果你在做批量导入，完全不在意搜索，你可以彻底关掉合并限流。另外还可以增加 index.translog.flush_threshold_size 设置，从默认的512MB到更大一些的值，比如1GB，这可以在一次清空触发的时候在事务日志里积累出更大的段。
    3）如果你的搜索结果不需要近实时的准确度，考虑把每个索引的index.refresh_interval 改到30s。
    4）如果你在做大批量导入，考虑通过设置index.number_of_replicas: 0 关闭副本。
    5）需要大量拉取数据的场景，可以采用scan & scroll api来实现，而不是from/size一个大范围。
    存储层面：
    1）基于数据+时间滚动创建索引，每天递增数据。控制单个索引的量，一旦单个索引很大，存储等各种风险也随之而来，所以要提前考虑+及早避免。
    2）冷热数据分离存储，热数据（比如最近3天或者一周的数据），其余为冷数据。对于冷数据不会再写入新数据，可以考虑定期force_merge加shrink压缩操作，节省存储空间和检索效率。

4 微服务
Nacos是什么？
阿里的一个开源产品，是针对微服务架构中的服务发现、配置管理、服务治理的综合型解决方案。

（用来实现配置中心和服务注册中心）

介绍Nacos功能
服务发现和服务健康监测（使服务更容易注册，并通过DNS或HTTP接口发现其他服务，还提供服务的实时健康检查，以防 止向不健康的主机或服务实例发送请求。 ）
支持基于DNS和基于RPC的服务发现。服务提供者使用原生SDK、OpenAPI、或一个独立的Agent TODO注册 Service 后，服务消费者可以使用DNS TODO 或HTTP&API查找和发现服务。
Nacos提供对服务的实时的健康检查，阻止向不健康的主机或服务实例发送请求。Nacos 支持传输层 (PING 或 TCP)和应用层 (如 HTTP、MySQL、用户自定义）的健康检查。 对于复杂的云环境和网络拓扑环境中（如 VPC、边缘网络等）服务的健康检查，Nacos 提供了 agent 上报模式和服务端主动检测2种健康检查模式。Nacos 还提供了统一的健康检查仪表盘，帮助您根据健康状态管理服务的可用性及流量。

动态配置服务
以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。
消除了配置变更时重新部署应用和服务的需要，让配置管理变得更加高效和敏捷。
配置中心化管理让实现无状态服务变得更简单，让服务按需弹性扩展变得更容易。
提供了一个简洁易用的UI (控制台样例 Demo) 帮助管理所有的服务和应用的配置。
Nacos 还提供包括配置版本跟踪、金丝雀发布、一键回滚配置以及客户端配置更新状态跟踪在内的一系列开箱即用的配置管理特性，能更安全地在生产环境中管理配置变更和降低配置变更带来的风险。

动态 DNS 服务
动态 DNS 服务支持权重路由，更容易地实现中间层负载均衡、更灵活的路由策略、流量控制以及数据中心内网的简单DNS解析服务。动态DNS服务还能更容易地实现以 DNS 协议为基础的服务发现，消除耦合到厂商私有服务发现 API 上的风险。
Nacos 提供了一些简单的 DNS APIs TODO ，管理服务的关联域名和可用的 IP:PORT 列表

服务及其元数据管理
从微服务平台建设的视角管理数据中心的所有服务及元数据，包括管理服务的描述、生命周期、服务的静态依赖分析、服务的健康状态、服务的流量管理、路由及安全策略、服务的 SLA 以及最首要的 metrics 统计数据。

服务发现
在微服务架构中一个业务流程需要多个微服务通过网络接口调用完成业务处理，服务消费方从服务注册中心获取服 务提供方的地址，从而进行远程调用，这个过程叫做服务发现。

服务发现流程：
服务实例本身并不记录服务生产方的网络地址，所有服务实例内部都会包含服务发现客户端。

在每个服务启动时会向服务发现中心上报自己的网络位置。在服务发现中心内部会形成一个服务注册表，服务注册表是服务发现的核心部分，是包含所有服务实例的网络地址的数据库。

服务发现客户端会定期从服务发现中心同步服务注册表 ，并缓存在客户端。

当需要对某服务进行请求时，服务实例通过该注册表，定位目标服务网络地址。若目标服务存在多个网络地址，则使用负载均衡算法从多个服务实例中选择出一个，然后发出请求。

总结，在微服务环境中，由于服务运行实例的网络地址是不断动态变化的，服务实例数量的动态变化 ，因此无法使用固定的配置文件来记录服务提供方的网络地址，必须使用动态的服务发现机制用于实现微服务间的相互感知。 各服务实例会上报自己的网络地址，这样服务中心就形成了一个完整的服务注册表，各服务实例会通过服务发现中心来获取访问目标服务的网络地址，从而实现服务发现的机制。

执行流程：

服务提供方将自己注册到服务注册中心
服务消费方从注册中心获取服务地址
进行远程调用

服务发现产品对比
目前市面上用的比较多的服务发现中心有：Nacos、Eureka、Consul和Zookeeper。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VtUuigh9-1666199870817)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019172840566.png)]

nacos和eureka的区别：
Nacos的服务实例分为两种类型：
临时实例：如果实例宕机超过一定时间，会从服务列表剔除，默认的类型。

非临时实例：如果实例宕机，不会从服务列表剔除，也可以叫永久实例。

spring:

cloud:

nacos:

discovery:

ephemeral: false # true:临时实例，false：非临时实例，永久实例

Nacos与eureka的共同点
都支持服务注册和服务拉取

都支持服务提供者心跳方式做健康检测

Nacos与Eureka的区别
Nacos支持服务端主动检测提供者状态：临时实例采用心跳模式，非临时实例采用主动检测模式：【正式工每天都会被检查状态，临时工也就心跳检测自己上报状态】

临时实例心跳不正常会被剔除，非临时实例【正式工】则不会被剔除

Nacos支持服务列表变更的消息推送模式，服务列表更新更及时

Nacos集群默认采用AP(可用)方式，当集群中存在非临时实例时，采用CP(一致性)模式；Eureka采用AP(可用)方式

服务发现数据模型
Namespace 隔离设计
命名空间（Namespace）用于进行租户粒度的隔离，Namespace 的常用场景之一是不同环境的隔离，例如开发测试环境和生产环境的资源（如配置、服务）隔离等。

从一个租户(用户)的角度来看，如果有多套不同的环境，那么这个时候可以根据指定的环境来创建不同的 namespce，以此来实现多环境的隔离。如有开发，测试和生产三个不同的环境，那么使用一套 nacos 集群可以分别建以下三个不同的 namespace。

从多个租户(用户)的角度来看，每个租户(用户)可能会有自己的 namespace，每个租户(用户)的配置数据以及注册的服务数据都会归属到自己的 namespace 下，以此来实现多租户间的数据隔离。

命名空间管理
命名空间(Namespace)是用于隔离多个环境的（如开发、测试、生产），而每个应用在不同环境的同一个配置（如数据库数据源）的值是不一样的。因此，我们应针对企业项目实际研发流程、环境进行规划。 如某软件公司拥有开发、测试、生产三套环境，那么我们应该针对这三个环境分别建立三个namespace。

建立好所有namespace后，在配置管理与服务管理模块下所有页面，都会包含用于切换namespace(环境)的tab按钮；

注意：

namesace 为 public 是 nacos 的一个保留空间，如需要创建自己的 namespace，不要和 public 重名，以一个实际业务场景有具体语义的名字来命名，以免带来字面上不容易区分哪一个 namespace。

在编写程序获取配置集时，指定的namespace参数一定要填写命名空间ID，而不是名称

数据模型
Nacos在经过阿里内部多年生产经验后提炼出的数据模型，是一种 服务-集群-实例 的三层模型，这样基本可以满 足服务在所有场景下的数据存储和管理。

服务：对外提供的软件功能，通过网络访问预定义的接口。

实例：提供一个或多个服务的具有可访问网络地址（IP:Port）的进程，启动一个服务，就产生了一个服务实例。

元信息：Nacos数据（如配置和服务）描述信息，如服务版本、权重、容灾策略、负载均衡策略、鉴权配置、各种自定义标 签 (label)，

从作用范围来分：服务级别的元信息、集群的元信息、实例的元信息。

集群：服务实例的集合，服务实例组成一个默认集群, 集群可以被进一步按需求划分，划分的单位可以是虚拟集群，相同集群下的实例才能相互感知。

应用通过Namespace、Service、Cluster(DEFAULT)的配置，描述了该服务向哪个环境（如开发环境）的哪个集群注册实例。
例子：
指定namespace的id：a1f8e863-3117-48c4-9dd3-e9ddc2af90a8（注意根据自己环境设置namespace的id）
指定集群名称：DEFAULT表示默认集群，可不填写

spring:
application:
name: transaction‐service
cloud:
nacos:
discovery:
server‐addr: 127.0.0.1:8848 # 注册中心地址
namespace: a1f8e863‐3117‐48c4‐9dd3‐e9ddc2af90a8 #开发环境
cluster‐name: DEFAULT #默认集群，可不填

使用Nacos作为配置中心
Nacos除了实现了服务的注册发现之外，还将配置中心功能整合在了一起。通过Nacos的配置管理功能，我们可以将整个架构体系内的所有配置都集中在Nacos中存储。这样做的好处，在以往的教程中介绍Spring Cloud Config时也有提到，主要有以下几点：

分离的多环境配置，可以更灵活的管理权限，安全性更高。
应用程序的打包更为纯粹，以实现一次打包，多处运行的特点。
配置动态刷新（可以在读取配置的类上面添加注解@RefreshScope来实现动态刷新）
配置回滚（可以再历史版本里面查看到配置文件修改的记录，可以选择对应的版本回滚）。
Nacos的配置管理，基础层面都通过DataId和Group来定位配置内容，除此之外还增加了很多其他的管理功能。

分布式配置中心实现原理
本地应用读取云端分布式配置中心文件（第一次读取时建立长连接）
本地应用读取到配置文件后，本地jvm和硬盘都会缓存一份。
本地应用于分布式配置中心服务器端一直保持长连接
当我们的配置文件发生变化（根据版本号|MID判断）。将变化结果通知本地应用及时刷新配置文件。

对于Nacos配置管理，通过Namespace、group、Data ID能够定位到一个配置集。

配置集(Data ID)：
在系统中，一个配置文件通常就是一个配置集，一个配置集可以包含了系统的各种配置信息，如：一个配置集可能包含了数据源、线程池、日志级别等配置项。每个配置集都可以定义一个有意义的名称，就是配置集的ID即Data ID。

配置项：

配置集中包含的一个个配置内容就是配置项。它代表一个具体的可配置的参数与其值域，通常以 key=value 的形式存在。如我们常配置系统的日志输出级别（logLevel=INFO|WARN|ERROR） 就是一个配置项。

配置分组(Group)：
配置分组是对配置集进行分组，通过一个有意义的字符串（如 Buy 或 Trade ）来表示，不同的配置分组下可以有相同的配置集（Data ID）。当在 Nacos 上创建一个配置时，如果未填写配置分组的名称，则配置分组的名称默认采用DEFAULT_GROUP。配置分组的常见场景：可用于区分不同的项目或应用，例如：学生管理系统的配置集可以定义一个group为：STUDENT_GROUP。

命名空间(Namespace)：
命名空间可用于进行不同环境的配置隔离。例如可以隔离开发环境、测试环境和生产环境，因为 它们的配置可能各不相同，或者是隔离不同的用户，不同的开发人员使用同一个nacos管理各自的配置，可通过 namespace隔离。不同的命名空间下，可以存在相同名称的配置分组(Group) 或 配置集。

常见实践用法：

Nacos抽象定义了Namespace、Group、Data ID的概念，具体这几个概念代表什么，取决于我们把它们看成什么，如：

Namespace：代表不同环境，如开发、测试、生产环境；

Group：代表某项目；

DataId：每个项目下往往有若干个工程，每个配置集(DataId)是一个工程的主配置文件

SpringCloud LoadBalancer （下文简称 SCL）
SpringCloud原有的客户端负载均衡方案Ribbon已经被废弃，取而代之的是SpringCloud LoadBalancer。
Spring Cloud 中内部微服务调用默认是 http 请求，主要通过下面三种 API：

RestTemplate：同步 http API

WebClient：异步响应式 http API

三方客户端封装，例如 openfeign
如果项目中加入了 spring-cloud-loadbalancer 的依赖并且配置启用了，那么会自动在相关的 Bean 中加入负载均衡器的特性。

对于 RestTemplate，会自动对所有 @LoadBalanced 注解修饰的 RestTemplate Bean 增加 Interceptor 从而加上了负载均衡器的特性。

对于 WebClient，会自动创建 ReactorLoadBalancerExchangeFilterFunction，我们可以通过加入ReactorLoadBalancerExchangeFilterFunction会加入负载均衡器的特性。

对于三方客户端，一般不需要我们额外配置什么。

（spring cloud 2020） 内置轮询、随机的负载均衡策略，默认轮询策略。

可以通过 LoadBalancerClient 注解，指定服务级别的负载均衡策略

@LoadBalancerClient(value = “demo-provider”, configuration = RandomLoadbalancerConfig.class)
public class RandomLoadbalancerConfig {
@Bean
public ReactorLoadBalancer reactorServiceInstanceLoadBalancer(Environment environment,
LoadBalancerClientFactory loadBalancerClientFactory) {
String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
return new RandomLoadBalancer(
loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class), name);
}
}
自定义负载均衡策略
通过上文可知，目前 SCL 支持的负载均衡策略相较于 Ribbon 还是较少，需要开发者自行实现，好在 SCL 提供了便捷的 API 方便扩展使用。 这里演示自定义一个基于注册中心元数据的灰度负载均衡策略。

定义灰度负载均衡策略

@Slf4j
public class GrayRoundRobinLoadBalancer extends RoundRobinLoadBalancer {
private ObjectProvider serviceInstanceListSupplierProvider;

private String serviceId;

@Override
public Mono<Response> choose(Request request) {
ServiceInstanceListSupplier supplier = serviceInstanceListSupplierProvider
.getIfAvailable(NoopServiceInstanceListSupplier::new);
return supplier.get(request).next().map(serviceInstances -> getInstanceResponse(serviceInstances, request));
}

Response getInstanceResponse(List instances, Request request) {

// 注册中心无可用实例 抛出异常
if (CollUtil.isEmpty(instances)) {
	log.warn("No instance available {}", serviceId);
	return new EmptyResponse();
}

DefaultRequestContext requestContext = (DefaultRequestContext) request.getContext();
RequestData clientRequest = (RequestData) requestContext.getClientRequest();
HttpHeaders headers = clientRequest.getHeaders();

String reqVersion = headers.getFirst(CommonConstants.VERSION);
if (StrUtil.isBlank(reqVersion)) {
	return super.choose(request).block();
}

// 遍历可以实例元数据，若匹配则返回此实例
for (ServiceInstance instance : instances) {
	NacosServiceInstance nacosInstance = (NacosServiceInstance) instance;
	Map<String, String> metadata = nacosInstance.getMetadata();
	String targetVersion = MapUtil.getStr(metadata, CommonConstants.VERSION);
	if (reqVersion.equalsIgnoreCase(targetVersion)) {
		log.debug("gray requst match success :{} {}", reqVersion, nacosInstance);
		return new DefaultResponse(nacosInstance);
	}
}
// 降级策略，使用轮询策略
return super.choose(request).block();

}
针对客户端注入灰度负载均衡策略

@LoadBalancerClient(value = “demo-provider”, configuration = GrayRoundLoadbalancerConfig.class)
优化负载均衡策略注入
如上文所述，所有的个性化负载策略都需要手动通过 LoadBalancerClient 注入非常的不方便。 我们可以参考 LoadBalancerClients 的批量注入逻辑构造自己的 BeanRegistrar

public class GrayLoadBalancerClientConfigurationRegistrar implements ImportBeanDefinitionRegistrar {

@Override
public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry registry) {
Field[] fields = ReflectUtil.getFields(ServiceNameConstants.class);

// 遍历服务名称，注入支持灰度策略的负载均衡器
for (Field field : fields) {
	Object fieldValue = ReflectUtil.getFieldValue(ServiceNameConstants.class, field);
	registerClientConfiguration(registry, fieldValue, GrayLoadBalancerClientConfiguration.class);
}

}
}
什么雪崩问题？如何解决？
微服务间相互调用，因为调用链中的某一个服务器发生故障，引起整个系统都无法提供服务的情况，就是雪崩。

解决方案：

限流：是对服务的保护，避免因瞬间高并发高流量导致服务故障，进而避免雪崩，这是一种预防措施
补救措施：超时处理，熔断处理等等

dubbo的注册中心（nacos）挂掉之后，服务器调用者还能调用服务提供者吗？
可以。因为dubbo拉取从注册中心拉取服务之后，都会在本地缓存服务列表，所以当注册中心挂掉之后，dobbo会直接从本地获取服务地址进行调用。

大型网站架构演变过程
网站架构演变演变过程
传统架构 → 分布式架构 → SOA架构 → 微服务架构

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qocyWp3z-1666199870818)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019173722619.png)]

分布式架构
分布式架构就是将传统结构按照模块进行拆分，不同的人负责不同的模块，不会产生代码冲突问题，方便开发。

SOA架构
SOA架构就是将业务逻辑层提取出来，将相似的业务逻辑形成一个服务，提供外部访问接口，服务之间访问通过RPC调用实现。

微服务架构

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-WTZWE5LK-1666199870819)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019173844580.png)]微服务类似于SOA架构，但是比SOA架构粒度更细，更轻量。
微服务架构与SOA架构区别
SOA基于WebService和ESP实现，底层基于HTTP协议和使用XML方式传输，XML在网络传输过程中会产生大量冗余。微服务由SOA架构演变而来，继承了SOA架构的优点，同时对SOA架构缺点进行改善，数据传输采用JSON格式，相比于XML更轻量和快捷，粒度更细，更加便于敏捷开发。SOA数据库会存在共享，微服务提倡每个服务连接独立的数据库。

Spring Cloud Alibaba 环境搭建
https://www.bilibili.com/video/BV1fe4y1b7ha?p=8&vd_source=3d12dd63f5f7550ce578fe1c95ab79e5

创建父级 Spring Boot 项目

pom.xml

4.0.0

com.pushihao
test
0.0.1-SNAPSHOT
test
test

<java.version>11</java.version>
<spring.boot.version>2.6.7</spring.boot.version>

    <!--spring-boot版本管理器-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>${spring.boot.version}</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>
</dependencies>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

新建两个模块

新建子模块时最好新建 Maven 项目，因为可以设置父项目。如果新建 Spring Boot Initializr 则默认父项目是 spring-boot-starter-parent

这里以订单模块（order）和仓库模块（stock）为例

假设仓库模块为生产者，订单模块为调用者。当调用订单模块时，订单模块调用仓库模块，使库存减一

项目结构如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ltVIEqmv-1666199870820)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019174303483.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KXrFYcXM-1666199870821)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019174314342.png)]

具体文件：

pom.xml （两个模块的 pom.xml 几乎相同）

SpringCloud
com.pushihao
0.0.1-SNAPSHOT

4.0.0

stock

package com.pushihao;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StockApplication {
public static void main(String[] args) {
SpringApplication.run(StockApplication.class, args);
}
}
StockController.java

package com.pushihao.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(“/stock”)
public class StockController {
@GetMapping(“reduct”)
public String reduct() {
System.out.println(“库存减一”);
return “success!”;
}
}
OrderApplication.java

package com.pushihao;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;

@SpringBootApplication
public class OrderApplication {
public static void main(String[] args) {
SpringApplication.run(OrderApplication.class, args);
}
@Bean
public RestTemplate restTemplate(RestTemplateBuilder builder) {
return builder.build();
}
}
OrderController.java

package com.pushihao.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;

@RestController
@RequestMapping(“/order”)
public class OrderController {
@Autowired
private RestTemplate restTemplate;

@GetMapping(“add”)
public String add() {
System.out.println(“订单加一”);
//result为返回结果
String result = restTemplate.getForObject(“http://localhost:9001/stock/reduct”, String.class);
return “success!”;
}
}
至此，一个简单的分布式环境就搭建好了，使用浏览器调用 http://localhost:9002/order/add 就可以看到结果

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ai7ALRLz-1666199870824)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019174603281.png)]

二、Spring Cloud Alibaba 环境搭建
可以直接在原有的分布式环境上直接引用 Spring Cloud Alibaba 即可

项目结构如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ECN2vtq9-1666199870826)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019174618486.png)]

导入 Spring Cloud Alibaba 和 Spring Cloud 的坐标

注意：版本号一定要选对（按照要求）参考 版本说明：https://github.com/alibaba/spring-cloud-alibaba/wiki/%E7%89%88%E6%9C%AC%E8%AF%B4%E6%98%8E

稳定版本依赖关系

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KPNxAsgY-1666199870827)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019174646802.png)]

组件版本关系（一般由Spring Cloud Alibaba 版本管理器直接控制，我们不用关心）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hcQE7gTn-1666199870831)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019174821364.png)]

这里使用最新稳定版即可

4.0.0

com.pushihao
test
pom
0.0.1-SNAPSHOT

order
stock
order-nacos
stock-nacos

test
test

<java.version>11</java.version>
<spring.boot.version>2.3.12.RELEASE</spring.boot.version>
<spring.cloud.version>Hoxton.SR12</spring.cloud.version>
<spring.cloud.alibaba.version>2.2.7.RELEASE</spring.cloud.alibaba.version>

    <!--spring-boot版本管理器-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>${spring.boot.version}</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>

    <!--spring-cloud版本管理器-->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-dependencies</artifactId>
        <version>${spring.cloud.version}</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>

    <!--spring-cloud-alibaba版本管理器-->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-alibaba-dependencies</artifactId>
        <version>${spring.cloud.alibaba.version}</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>
</dependencies>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

pom.xml

SpringCloud
com.pushihao
0.0.1-SNAPSHOT

4.0.0
stock-nacos

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

	<!--注册与发现-->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
            <configuration>
                <source>11</source>
                <target>11</target>
            </configuration>
        </plugin>
    </plugins>
</build>

server:
port: 9001
spring:
application:
name: stock-service
cloud:
nacos:
server-addr: 127.0.0.1:8848
discovery:
username: nacos
password: nacos
cluster-name: public
StockNacosApplication.java

package com.pushihao;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class StockNacosApplication {
public static void main(String[] args) {
SpringApplication.run(StockNacosApplication.class, args);
}
}
StockController.java

package com.pushihao.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(“/stock”)
public class StockController {

@GetMapping("reduct")
public String reduct() {
    System.out.println("库存减一");
    return "success!";
}

}
2.2 新建 order-nacos 模块

pom.xml

test
com.pushihao
0.0.1-SNAPSHOT

4.0.0

<artifactId>order-nacos</artifactId>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
            <configuration>
                <source>11</source>
                <target>11</target>
            </configuration>
        </plugin>
    </plugins>
</build>

server:
port: 9001
spring:
application:
name: order-service
cloud:
nacos:
server-addr: 127.0.0.1:8848
discovery:
username: nacos
password: nacos
namespace: public
# ephemeral: false #是否是临时实例 默认是true（临时实例） 永久实例：哪怕宕机了也不会删除实例
OrderNacosApplication.java

package com.pushihao;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;

@SpringBootApplication
public class OrderNacosApplication {
public static void main(String[] args) {
SpringApplication.run(OrderNacosApplication.class, args);
}

//加上@LoadBalanced就配上了默认的负载均衡器Ribbon
@Bean
@LoadBalanced
public RestTemplate restTemplate(RestTemplateBuilder restTemplateBuilder) {
    return restTemplateBuilder.build();
}

}
OrderController.java

package com.pushihao.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;

@RestController
@RequestMapping(“/order”)
public class OrderController {

@Autowired
RestTemplate restTemplate;

@GetMapping("add")
public String add() {
    System.out.println("下单成功");

    //这里就可以把IP地址替换成对应的服务名，调用时就会启用默认的负载均衡机制
    String msg = restTemplate.getForObject("http://stock-service/stock/reduct", String.class);

    return "success!";
}

}
至此 Spring Cloud Alibaba 环境就搭建完毕了

依次启动 nacos 服务器、stock-nacos、order-nacos

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zSUtsQXd-1666199870833)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019175338661.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vmjsSJ5j-1666199870834)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019175346701.png)]

成功！

三、使用 Aliyun Java Initializr 快速构建
不过更多情况下，都是使用 Idea 工具进行快速构建，如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MI3gLgUh-1666199870837)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019175353920.png)]

以上

8 容器
Docker

1. 什么是 Docker 容器？
   Docker 容器在应用程序层创建抽象并将应用程序及其所有依赖项打包在一起。这使我们能够快速可靠地部署应用程序。容器不需要我们安装不同的操作系统。相反，它们使用底层系统的 CPU 和内存来执行任务。这意味着任何容器化应用程序都可以在任何平台上运行，而不管底层操作系统如何。我们也可以将容器视为 Docker 镜像的运行时实例。

2. 什么是 DockerFile？
   Dockerfile 是一个文本文件，其中包含我们需要运行以构建 Docker 映像的所有命令。Docker 使用 Dockerfile 中的指令自动构建镜像。我们可以docker build用来创建按顺序执行多个命令行指令的自动构建。

3. 如何从 Docker 镜像创建 Docker 容器？
   为了从镜像创建容器，我们从 Docker 存储库中提取我们想要的镜像并创建一个容器。我们可以使用以下命令：

1

$ docker run -it -d <image_name>

4. Docker Compose 可以使用 JSON 代替 YAML 吗？
   是的，我们可以对Docker Compose文件使用 JSON 文件而不是YAML

$ docker-compose -f docker-compose.json up

5. 什么是Docker Swarm？
   Docker Swarm 是一个容器编排工具，它允许我们跨不同主机管理多个容器。使用 Swarm，我们可以将多个 Docker 主机变成单个主机，以便于监控和管理。

6. 如果你想使用一个基础镜像并对其进行修改，你怎么做？
   我们可以使用以下 Docker 命令将图像从 Docker Hub 拉到我们的本地系统上：

$ docker pull <image_name>

7. 如何启动、停止和终止容器？
   要启动 Docker 容器，请使用以下命令：

$ docker start <container_id>

要停止 Docker 容器，请使用以下命令：

$ docker stop <container_id>

要终止 Docker 容器，请使用以下命令：

$ docker kill <container_id>

8. Docker 运行在哪些平台上？
   Docker 在以下 Linux 发行版上运行：

CentOS 6+

Gentoo

ArchLinux

CRUX 3.0+

openSUSE 12.3+

RHEL 6.5+

Fedora 19/20+

Ubuntu 12.04、13.04

Docker 还可以通过以下云服务在生产中使用：

微软Azure

谷歌计算引擎

亚马逊 AWS EC2

亚马逊 AWS ECS

机架空间

提示：我们始终建议您在面试之前进行一些公司研究。要为这个特定问题做准备，请了解公司如何使用 Docker 并在您的答案中包含他们使用的平台。

9. 解释 Docker 组件。
   三个架构组件包括 Docker 客户端、主机和注册表。

Docker 客户端：该组件执行构建和运行操作以与 Docker 主机通信。

Docker 主机：该组件包含 Docker 守护程序、Docker 镜像和 Docker 容器。守护进程建立到 Docker Registry 的连接。

Docker Registry：该组件存储 Docker 镜像。它可以是公共注册表，例如 Docker Hub 或 Docker Cloud，也可以是私有注册表。

10. 虚拟化和容器化有什么区别？
    虚拟化

虚拟化帮助我们在单个物理服务器上运行和托管多个操作系统。在虚拟化中，管理程序为客户操作系统提供了一个虚拟机。VM 形成了硬件层的抽象，因此主机上的每个 VM 都可以充当物理机。

容器化

容器化为我们提供了一个独立的环境来运行我们的应用程序。我们可以在单个服务器或 VM 上使用相同的操作系统部署多个应用程序。容器构成了应用层的抽象，所以每个容器代表一个不同的应用。

11. 管理程序的功能是什么？
    管理程序或虚拟机监视器是帮助我们创建和运行虚拟机的软件。它使我们能够使用单个主机来支持多个来宾虚拟机。它通过划分主机的系统资源并将它们分配给已安装的来宾环境来实现这一点。可以在单个主机操作系统上安装多个操作系统。有两种类型的管理程序：

Native：本机管理程序或裸机管理程序，直接在底层主机系统上运行。它使我们可以直接访问主机系统的硬件，并且不需要基本服务器操作系统。

托管：托管管理程序使用底层主机操作系统。

12.如何构建Dockerfile？
为了使用我们概述的规范创建映像，我们需要构建一个 Dockerfile。要构建 Dockerfile，我们可以使用以下docker build命令：

$ docker build

13. 使用什么命令将新镜像推送到 Docker Registry？
    要将新镜像推送到 Docker Registry，我们可以使用以下docker push命令：

$ docker push myorg/img

14.什么是Docker引擎？
Docker Engine 是一种开源容器化技术，我们可以使用它来构建和容器化我们的应用程序。Docker Engine 由以下组件支持：

Docker 引擎 REST API

Docker 命令行界面 (CLI)

Docker 守护进程

15. 如何访问正在运行的容器？
    要访问正在运行的容器，我们可以使用以下命令：

$ docker exec -it <container_id> bash

16.如何列出所有正在运行的容器？
要列出所有正在运行的容器，我们可以使用以下命令：

$ docker ps

17. 描述 Docker 容器的生命周期。
    Docker 容器经历以下阶段：

创建容器

运行容器

暂停容器（可选）

取消暂停容器（可选）

启动容器

停止容器

重启容器

杀死容器

销毁容器

18. 什么是Docker对象标签？
    Docker 对象标签是存储为字符串的键值对。它们使我们能够将元数据添加到 Docker 对象，例如容器、网络、本地守护进程、图像、Swarm 节点和服务。

19. 使用Docker Compose时如何保证容器1先于容器2运行？
    Docker Compose 在继续下一个容器之前不会等待容器准备就绪。为了控制我们的执行顺序，我们可以使用“取决于”条件，depends_on。这是在 docker-compose.yml 文件中使用的示例：

version: “2.4”

services:

backend:

build: .

depends_on:

 - db

db:

image: postgres
该docker-compose up命令将按照我们指定的依赖顺序启动和运行服务。

20.docker create命令有什么作用？
该docker create命令在指定映像上创建可写容器层，并准备该映像以运行指定命令。

Docker部署安装
简介
什么是Docker?

Docker是开发人员和系统管理员使用容器开发、部署和运行应用程序的平台。使用Linux容器来部署应用程序称为集装箱化。使用docker轻松部署应用程序

Docker将应用程序与该程序的依赖，打包在一个文件里面。运行这个文件，就会生成一个虚拟容器。程序在这个虚拟容器里运行，就好像在真实的物理机上运行一样。有了Docker，就不用担心环境问题。

总体来说，Docker的接口相当简单，用户可以方便地创建和使用容器，把自己的应用放入容器。容器还可以进行版本管理、复制、分享、修改，就像管理普通的代码一样

Centos7上安装docker
依赖安装：

yum install -y yum-utils device-mapper-persistent-data lvm2
-阿里源安装：

yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
docker安装：

yum install docker-ce # 安装指定版本，例如yum install -y docker-ce-18.09
启动并加入开机自启

systemctl start docker
systemctl enable docker
修改docker数据存放目录位置（默认目录是/var/lib/docker，这里防止/目录满我修改到数据盘内）和镜像加速

vim /etc/docker/daemon.json

{
“graph”: “/data/docker”,
“registry-mirrors”: [“https://01xxgaft.mirror.aliyuncs.com”]
}
重启docker载入新配置：

systemctl restart docker
基本使用
这里可以docker --help查看所有用法，下面是比较常用的

镜像获取查看
镜像搜索：

docker search 想搜索的镜像包名 （例如：docker search nginx）

镜像下载：docker pull 想下载的镜像包名 （例如：docker pull nginx）

查看本地所有的镜像：docker image ls

容器启动和管理

容器启动：docker run -d 镜像名 （直接后台启动容器 例如：docker run -d nginx）

容器交互模式启动：docker run -d -ti 镜像名 /bin/bash （使用交互模式启动容器 例如：docker run -d -ti centos /bin/bash）

容器启动打上标记名：docker run -d --name=标记名 镜像名 （给容器打上标记名 例如：docker run -d --name=test_nginx nginx）

容器启动映射端口：docker run -d -p 需要映射的宿主机端口:容器端口 镜像名 （将容器端口映射到宿主机 例如：docker run -d -p 8080:80 nginx 这里就将容器的80端口映射到了宿主机的8080端口）

进入容器内部：docker exec -ti 镜像名 /bin/bash （进入容器内部 例如：docker exec -ti centos /bin/bash 推出时按住ctrl不放，另一个只手按p，然后再按q，这样就是安全退出，容器也不会死掉，如果是用交互模式运行的，直接exit退出就行）

镜像和容器的查看停止与删除
查看正在运行的容器，这里会显示各种信息：docker ps

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QhJKxuqi-1666199870838)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019175951164.png)]

显示所有的容器，包括已经停止运行的：docker ps -a

停止容器：docker stop 容器id或者标记的name （这里id可以docker ps 看到，标记的name是启动时你设置的，例如停止上图的nginx：docker stop 7baea3ea0701 或 docker stop nginx_test1）

删除容器：docker rm 容器id或者标记的name （主要用于删除已经停止运行的容器，但是容器必须是停止状态才能删除，例如：docker rm 7baea3ea0701 或 docker rm nginx_test1）

删除镜像：docker rmi 镜像名 （主要用于删除镜像 例如：docker rmi nginx）

镜像的各种打包
将当前运行的容器状态打包成镜像：docker commit -m ‘提交内容’ -a ‘提交人’ 已启动并修改过的容器名 新的镜像名以及版本号 已启动并修改过的容器名 新的镜像名以及版本号 （主要用于将运行中的容器进入内部修改后将其打包成新的镜像）
这里我拿上图的nginx来打包 例如：docker commit -m ‘test’ -a ‘test’ nginx_test1 nginx:testv1.0.0
使用dockerfile打包：docker build -t 新的镜像名以及版本号 目录位置 （这里是使用dockerfile进行打包，需要新建一个目录并在里面创建一个Dockerfile文件）
例如，我新建一个nginxfile目录 mkdir nginxfile &&; cd nginxfile && touch Dockerfile

编辑Dockerfile内容，格式为[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zR9RGnVc-1666199870839)(C:UsersshaoAppDataRoamingTyporatypora-user-imagesimage-20221019180028352.png)]

FROM 镜像名 RUN 打包过程中执行的命令，可以有多个，当有多个时用一条RUN，然后用 && 和 来连接，否则或创建多层镜像 CMD 运行容器时执行的命令，只能有一个生效

注意：CMD内的内容必须是前台运行的程序，否则后台后容器会挂掉，因为容器启动时会记住CMD后面这个命令执行的pid，并将其标识为1，如果该进程没有存活，容器会认为任务已经运行结束，会自动挂掉，上面我将nginx前台启动

镜像的导入导出

镜像文件导出：docker save -o 保存的文件位置 镜像名 （例如：docker save -o /tmp/nginx.tar.gz nginx）

镜像文件导入：docker load < 镜像包位置 （例如 docker load < /tmp/nginx.tar.gz nginx）

容器和宿主机的数据交互

拷贝文件：docker cp 文件位置 容器id或标记的name:容器内位置 （将宿主机内文件拷贝到容器内，也可以反着来将容器内文件拷贝出来，例如：docker cp /tmp/1.txt nginx_name:/tmp/ 反向拷贝：docker cp nginx_name:/tmp/ /tmp/1.txt ）

挂载本地目录：docker run d -v 宿主机目录:容器内目录 镜像名（将本地目录映射挂载到容器内目录，例如：docker run d -v /tmp/logs:/data/logs nginx）

四，大数据
1 hadoop