谷粒商城（三）

Elasticsearch - 全文检索

简介

全文搜索属于最常见的需求，开源的 Elasticsearch 是目前全文搜索引擎的首选，它可以快速地储存、搜索和分析海量数据。维基百科、Stack Overflow、Github 都采用它。
Elastic 的底层是开源库 Lucene。但是，你没法直接用 Lucene，必须自己写代码去调用它的接口。Elastic 是 Lucene 的封装，提供了 REST API 的操作接口，开箱即用。

REST API：天然的跨平台。
官方文档（推荐）
官方中文
社区中文1
社区中文2

1、基本概念

1、Index（索引）
动词，相当于 MySQL 中的 insert；
名词，相当于 MySQL 中的 Database

2、Type（类型）
在 Index（索引）中，可以定义一个或多个类型。
类似于 MySQL 中的 Table；每一种类型的数据放在一起；
es6 之后要求一个索引只能包含一个类型

3、Document（文档）
保存在某个索引（Index）下，某种类型（Type）的一个数据（Document），文档是 JSON 格式的，Document 就像是 MySQL 中的某个 Table 里面的内容；

4、搜索原理：倒排索引机制

2、Docker 安装 Es

1、下载镜像文件

docker pull elasticsearch:7.4.2 存储和检索数据
docker pull kibana:7.4.2 可视化检索数据

2、创建实例

ElasticSearch
配置

mkdir -p /mydata/elasticsearch/config # 用来存放配置文件
mkdir -p /mydata/elasticsearch/data  # 数据
echo "http.host: 0.0.0.0" >/mydata/elasticsearch/config/elasticsearch.yml # 允许任何机器访问
chmod -R 777 /mydata/elasticsearch/ ## 设置elasticsearch文件可读写权限

启动

docker run --name elasticsearch -p 9200:9200 -p 9300:9300 \
-e  "discovery.type=single-node" \      单例启动
-e ES_JAVA_OPTS="-Xms64m -Xmx512m" \	分配内存
-v /mydata/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml \
-v /mydata/elasticsearch/data:/usr/share/elasticsearch/data \
-v  /mydata/elasticsearch/plugins:/usr/share/elasticsearch/plugins \
-d elasticsearch:7.4.2 

开机启动 elasticsearch

docker update elasticsearch --restart=always
以后再外面装好插件重启就可

特别注意：
-e ES_JAVA_OPTS=“-Xms64m -Xmx128m” \ 测试环境下，设置 ES 的初始内存和最大内存，否则导致过大启动不了ES

Kibana

docker run --name kibana -e ELASTICSEARCH_HOSTS=http://192.168.56.10:9200 -p 5601:5601 -d kibana:7.4.2

安装完 es 、Kibana后
访问 http://192.168.56.10:9200 检测是否成功运行 es
访问 http://192.168.56.10:5601 可视化检索工具 Kibana

安装nginx
如果 docker 没有安装过，也可以直接启动一个 nginx 实例，启动时会自动帮你下载

docker run -p80:80 --name nginx -d nginx:1.10   

将容器内的配置文件拷贝到当前目录 （注意后面有个小点）

以下都是在mydata目录下执行

docker container cp nginx:/etc/nginx .   #将容器内的配置文件拷贝到当前目录：注意容器名要一致 nginx: 这里是容器名
ls   ##会发现mydata目录下出现 nginx 目录
mv nginx conf    #修改文件名称nginx为conf
mkdir nginx   #再重新创建文件夹
mv conf nginx/      #将conf移入nginx

执行以下命令运行

 docker run -p 80:80 --name nginx \
 -v /mydata/nginx/html:/usr/share/nginx/html \
 -v /mydata/nginx/logs:/var/log/nginx \
 -v /mydata/nginx/conf/:/etc/nginx \
 -d nginx:1.10

给 nginx 的 html 文件夹下面放的所有资源可以直接访问，一般 index.html 放在此文件夹下，则访问 IP:80 的nginx地址时就会访问此html文件，要访问 html 目录下的文件也只需要修改路径直接访问即可

例如后面自定义分词器，创建 html/es/fenci.txt 文件，文件内放入自定义分词 乔碧萝 等，访问 IP:80/es/fenci.txt 可以在浏览器看到

3、初步检索

3.1 _cat

GET /_cat/nodes：查看所有节点
GET /_cat/health：查看 es 健康状况
GET /_cat/master：查看主节点
GET /_cat/indices：查看所有索引 show databases;

3.2 索引一个文档（保存）

保存一个数据，保存在哪个索引的哪个类型下，指定用哪个唯一标识

#在 customer 索引下的 external 类型下保存 1 号数据
PUT customer/external/1
{ "name": "John Doe"}

PUT 和 POST 都可以：

• POST 新增。如果不指定 id，会自动生成 id。指定 id 就会修改这个数据，并新增版本号
• PUT 可以新增可以修改。PUT 必须指定 id；由于 PUT 需要指定 id，我们一般都用来做修改操作，不指定 id 会报错。

3.3 查询文档

GET customer/external/1

结果：

{
	"_index": "customer", //在哪个索引
	"_type": "external", //在哪个类型
	"_id": "1", //记录 id
	"_version": 2, //版本号
	"_seq_no": 1, //并发控制字段，每次更新就会+1，用来做乐观锁
	"_primary_term": 1, //同上，主分片重新分配，如重启，就会变化
	"found": true, "_source": { //真正的内容
		"name": "John Doe"
	}
}

因为有两个并发控制字段，所以更新的时候可以在语句后携带 ?if_seq_no=0&if_primary_term=1

3.4 更新文档

//POST带_update
POST customer/external/1/_update
{
    "doc":{ "name": "John Doew"}
}

//或者 POST不带_update
POST customer/external/1
{ "name": "John Doe2"}

//或者 PUT
PUT customer/external/1
{ "name": "John Doe"}

不同：

• POST带_update，对比原来数据，与原来一样就什么都不做，version/seq_no 都不变
• POST不带_update、PUT ，如果数据不变，还是会更新版本

看场景；

• 对于大并发更新，不带 update；
• 对于大并发查询偶尔更新，带 update；对比更新，重新计算分配规则。

更新同时增加属性

POST customer/external/1/_update
{ 
  "doc": { 
    "name": "Jane Doe", 
    "age": 20 
  }
}
PUT 和 POST 不带_update 也可以

3.5 删除文档&索引

DELETE customer/external/1
DELETE customer

3.6 bulk 批量 API

POST customer/external/_bulk
{"index":{"_id":"1"}}
{"name": "John Doe" }
{"index":{"_id":"2"}}
{"name": "Jane Doe" }

语法格式：

{ action: { metadata }}\n
{ request body }\n
{ action: { metadata }}\n
{ request body }\n

复杂实例：

POST /_bulk
{ "delete": { "_index": "website", "_type": "blog", "_id": "123" }}
{ "create": { "_index": "website", "_type": "blog", "_id": "123" }}
{ "title": "My first blog post" }
{ "index": { "_index": "website", "_type": "blog" }}
{ "title": "My second blog post" }
{ "update": { "_index": "website", "_type": "blog", "_id": "123", "_retry_on_conflict" : 3} }
{ "doc" : {"title" : "My updated blog post"} }

bulk API 以此按顺序执行所有的 action（动作）。如果一个单个的动作因任何原因而失败，它将继续处理它后面剩余的动作。当 bulk API 返回时，它将提供每个动作的状态（与发送的顺序相同），所以您可以检查是否一个指定的动作是不是失败了。

3.7 样本测试数据

准备了一份顾客银行账户信息的虚构的 JSON 文档样本。每个文档都有下列的 schema（模式）:

{ 
  "account_number": 0, 
  "balance": 16623, 
  "firstname": "Bradshaw", 
  "lastname": "Mckenzie", 
  "age": 29, 
  "gender": "F", 
  "address": "244 Columbus Place", 
  "employer": "Euron", 
  "email": "bradshawmckenzie@euron.com", 
  "city": "Hobucken", 
  "state": "CO"
}

ES测试数据 Gitee
ES测试数据语雀

将测试数据 JSON 拷贝进 ES 开发窗口，使用以下语句：

POST bank/account/_bulk
测试数据

4、进阶检索

4.1 SearchAPI

ES 支持两种基本方式检索 :

一个是通过使用 REST request URI 发送搜索参数（uri+检索参数）
另一个是通过使用 REST request body 来发送它们（uri+请求体）

1 检索信息

一切检索从_search 开始

GET bank/_search  //检索 bank 下所有信息，包括 type 和 docs
GET bank/_search?q=*&sort=account_number:asc   //请求参数方式检索

响应结果解释：
took - Elasticsearch 执行搜索的时间（毫秒）
time_out - 告诉我们搜索是否超时
_shards - 告诉我们多少个分片被搜索了，以及统计了成功/失败的搜索分片
hits - 搜索结果
hits.total - 搜索结果
hits.hits - 实际的搜索结果数组（默认为前 10 的文档）
sort - 结果的排序 key（键）（没有则按 score 排序）
score 和 max_score –相关性得分和最高得分（全文检索用）

uri+请求体进行检索

GET bank/_search
{ 
  "query": 
  { 
    "match_all": {}
  },
  "sort": [
    { 
      "account_number": { 
        "order": "desc"
      }
    }
  ]
}

我们 POST 一个 JSON 风格的查询请求体到 _search API。
需要了解，一旦搜索的结果被返回，Elasticsearch 就完成了这次请求，并且不会维护任何服务端的资源或者结果的 cursor（游标）

4.2 Query DSL

1 基本语法格式

Elasticsearch 提供了一个可以执行查询的 Json 风格的 DSL（domain-specific language 领域特定语言）。这个被称为 Query DSL。该查询语言非常全面，并且刚开始的时候感觉有点复杂，真正学好它的方法是从一些基础的示例开始的。

一个查询语句 的典型结构
{
QUERY_NAME: {
ARGUMENT: VALUE, ARGUMENT: VALUE,... }
}

如果是针对某个字段，那么它的结构如下：
{
  QUERY_NAME: {
    FIELD_NAME: {
      ARGUMENT: VALUE, 
      ARGUMENT: VALUE,... 
    }
  }
}

示例：

GET bank/_search
{ 
  "query": {   //query 定义如何查询
    "match_all": {}   //match_all 查询类型【代表查询所有的所有】
  },
  "from": 0, 
  "size": 5,   // from+size 完成分页功能
  "sort": [   // sort 排序，多字段排序
    { 
      "account_number": { 
        "order": "desc"
      }
    }
  ]
}

match_all 查询类型【代表查询所有的所有】，es 中可以在 query 中组合非常多的查询类型完成复杂查询
除了 query 参数之外，我们也可以传递其它的参数以改变查询结果。如 sort，size
from+size 限定，完成分页功能
sort 排序，多字段排序，会在前序字段相等时后续字段内部排序，否则以前序为准

2 返回部分字段

GET bank/_search
{ 
  "query": {
    "match_all": {}
  },
  "from": 0, 
  "size": 5, 
  "_source": ["age","balance"]    //返回部分字段
}

3 match【匹配查询】

基本类型（非字符串），精确匹配

GET bank/_search
{ 
  "query": { 
    "match": {       //match 返回 account_number=20 的
      "account_number": "20"
    }
  }
}

字符串，全文检索

//查询出address中包含 mill 单词的所有记录
GET bank/_search
{ 
  "query": { 
    "match": {
      "address": "mill"
    }
  }
}

match 当搜索字符串类型的时候，会进行全文检索，并且每条记录有相关性得分。

字符串，多个单词（分词+全文检索）

//查询出address中包含 mill 或者 road 或者 mill road 的所有记录
GET bank/_search
{ 
  "query": { 
    "match": { 
      "address": "mill road"
    }
  }
}

最终查询出 address 中包含 mill 或者 road 或者 mill road 的所有记录，并给出相关性得分

4 match_phrase【短语匹配】

将需要匹配的值当成一个整体单词（不分词）进行检索

//查出 address 中包含 mill road 的所有记录
GET bank/_search
{ 
  "query": { 
    "match_phrase": { 
      "address": "mill road"
    }
  }
}

查出 address 中包含 mill road 的所有记录，并给出相关性得分

5 multi_match【多字段匹配】

//查询 state 或者 address 包含 mill
GET bank/_search
{ 
  "query": { 
    "multi_match": { 
      "query": "mill", 
      "fields": ["state","address"]
    }
  }
}

只要指定字段里面包含要查询的值就返回
并且会分词 多个值只要包含其中一个值就会返回

//"990 Mill Road"作为一个整体查询，是短语匹配，只要包含以下内容就可以查询
GET bank/_search
{
  "query": {
    "match_phrase": { 
      "address": "990 Mill Road"
    }
  }
}

//"990 Mill Road"作为一个整体查询，是绝对匹配，必须一模一样
//每一个字符串的字段都可以使用 .keyword
GET bank/_search
{
  "query": {
    "match": { 
      "address.keyword": "990 Mill Road"
    }
  }
}

6 bool【复合查询】

bool 用来做复合查询：
复合语句可以合并 任何 其它查询语句，包括复合语句，了解这一点是很重要的。这就意味着，复合语句之间可以互相嵌套，可以表达非常复杂的逻辑。

must： 必须达到 must 列举的所有条件

GET bank/_search
{ 
  "query": { 
    "bool": { 
      "must": [
        { 
          "match": { "address": "mill" } 
        },
        { 
          "match": { "gender": "M" } 
        }
      ]
    }
  }
}

should： 应该达到 should 列举的条件，如果达到会增加相关文档的评分，并不会改变查询的结果。
如果 query 中只有 should 且只有一种匹配规则，那么 should 的条件就会被作为默认匹配条件而去改变查询结果

GET bank/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "match": { "address": "mill" }
        }
      ],
      "should": [
        {
          "match": {"address": "lane"}
        }
      ]
    }
  }
}

must_not： 必须不是指定的情况

GET bank/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "match": {"address": "mill"}
        },
        {
          "match": {"gender": "M"}
        }
      ],
      "must_not": [
        {
          "match": {"email": "baluba.com"}
        }
      ]
    }
  }
}

address 包含 mill，并且 gender 是 M，，但是 email 必须不包含 baluba.com

7 filter【结果过滤】

并不是所有的查询都需要产生分数，特别是那些仅用于 “filtering”（过滤）的文档。为了不计算分数 Elasticsearch 会自动检查场景并且优化查询的执行。

GET bank/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "match": { "address": "mill"}
        }
      ],
      "filter": {
        "range": {
          "balance": {
            "gte": 10000,
            "lte": 20000
          }
        }
      }
    }
  }
}

8 term

和 match 一样。匹配某个属性的值。全文检索字段用 match，其他非 text 字段匹配用 term。

GET bank/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "term": {
            "age": { "value": "28" }
          }
        },
        {
          "match": { "address": "990 Mill Road"}
        }
      ]
    }
  }
}

9 aggregations（执行聚合）

聚合提供了从数据中分组和提取数据的能力。最简单的聚合方法大致等于 SQL的 GROUP BY 和 SQL 聚合函数

在 Elasticsearch 中，您有执行搜索返回 hits（命中结果），并且同时返回聚合结果，把一个响应中的所有 hits（命中结果）分隔开的能力

这是非常强大且有效的，您可以执行查询和多个聚合，并且在一次使用中得到各自的（任何一个的）返回结果，使用一次简洁和简化的 API 来避免网络往返。

//搜索address中包含 mill 的所有人的年龄分布以及平均年龄，但不显示这些人的详情
GET bank/_search
{
  "query": {
    "match": {"address": "mill"}
  },
  "aggs": {     //执行聚合
    "group_by_state": {     //这次聚合的名字，方便展示在结果集中
      "terms": {      //聚合的类型
        "field": "age"
      }
    },
    "avg_age": {
      "avg": {		 //聚合的类型
        "field": "age"
      }
    }
  },
  "size": 0   //不显示搜索数据
}

//按照年龄聚合，并且请求这些年龄段的这些人的平均薪资
GET bank/account/_search
{
  "query": { "match_all": {} },
  "aggs": {
    "age_avg": {
        "terms": {
          "field": "age",
          "size": 1000     //给出1000种不同结果
        },
        "aggs": {
          "banlances_avg": {
            "avg": {"field": "balance"}
          }
        }
    }
  },
  "size": 1000
}

//查出所有年龄分布
//并且这些年龄段中 M 的平均薪资和 F 的平均薪资
//以及这个年龄段的总体平均薪资
GET bank/account/_search
{
  "query": {"match_all": {} },
  "aggs": {
    "age_agg": {
	      "terms": { 
	        "field": "age",		//按年龄分组
	        "size": 100     //给出100种不同结果
	      },
	      "aggs": {
	        "gender_agg": {
		          "terms": {
		            "field": "gender.keyword",   //在按年龄分组情况下再按性别分组
		            "size": 100
		          },
		          "aggs": {
		            "balance_avg": {
		              "avg": {
		                "field": "balance"		//求平均值
		              }
		            }
		          }
	        },
	        "balance_avg": {
	              "avg": {"field": "balance"} 	//在按年龄分组情况下求平均值
	        }
	      }
    }
  },
  "size": 1000
}

4.3 Mapping

1 字段类型

2 Mapping（映射）

Mapping 是用来定义一个文档（document），以及它所包含的属性（field）是如何存储和索引的。比如，使用 mapping 来定义：

• 哪些字符串属性应该被看做全文本属性（full text fields）。
• 哪些属性包含数字，日期或者地理位置。
• 文档中的所有属性是否都能被索引（_all 配置）。
• 日期的格式。
• 自定义映射规则来执行动态添加属性。

//查看 mapping 信息：
GET bank/_mapping

修改 mapping 信息

自动猜测的映射类型

3 新版本改变

Es7 及以上移除了 type 的概念。

关系型数据库中两个数据表示是独立的，即使他们里面有相同名称的列也不影响使用，但 ES 中不是这样的。elasticsearch 是基于 Lucene 开发的搜索引擎，而 ES 中不同 type 下名称相同的 filed 最终在 Lucene 中的处理方式是一样的。

两个不同 type 下的两个 user_name，在 ES 同一个索引下其实被认为是同一个 filed，你必须在两个不同的 type 中定义相同的 filed 映射。否则，不同 type 中的相同字段名称就会在处理中出现冲突的情况，导致 Lucene 处理效率下降。

去掉 type 就是为了提高 ES 处理数据的效率。

Elasticsearch 7.x
URL 中的 type 参数为可选。比如，索引一个文档不再要求提供文档类型。

Elasticsearch 8.x
不再支持 URL 中的 type 参数。

解决：
1）、将索引从多类型迁移到单类型，每种类型文档一个独立索引
2）、将已存在的索引下的类型数据，全部迁移到指定位置即可。详见数据迁移

创建映射

//创建索引并指定映射
PUT /my-index
{
  "mappings": {
    "properties": {
      "age": {
        "type": "integer"
      },
      "email": {
        "type": "keyword"
      },
      "name": {
        "type": "text"
      }
    }
  }
}

//添加新的字段映射
PUT /my-index/_mapping
{
  "properties": {
    "employee-id": {
      "type": "keyword",
      "index": false
    }
  }
}

更新映射：
对于已经存在的映射字段，我们不能更新。更新必须创建新的索引进行数据迁移

数据迁移

先创建出 new_twitter 的正确映射。然后使用如下方式进行数据迁移

POST _reindex     //固定写法
{
  "source": {
    "index": "twitter"   //旧映射
  },
  "dest": {
    "index": "new_twitter"   //新的映射
  }
}

示例：

//将旧索引的 type 下的数据进行迁移
POST _reindex
{
  "source": {
    "index": "twitter",
    "type": "tweet"
  },
  "dest": {
    "index": "tweets"
  }
}

4.4 分词

一个 tokenizer（分词器）接收一个字符流，将之分割为独立的 tokens（词元，通常是独立的单词），然后输出 tokens 流

例如，whitespace tokenizer 遇到空白字符时分割文本。它会将文本"Quick brown fox!" 分割为 [Quick, brown, fox!]

该 tokenizer（分词器）还负责记录各个 term（词条）的顺序或 position 位置（用于 phrase 短语和 word proximity 词近邻查询），以及 term（词条）所代表的原始 word（单词）的 start（起始）和 end（结束）的 character offsets（字符偏移量）（用于高亮显示搜索的内容）

Elasticsearch 提供了很多内置的分词器，可以用来构建 custom analyzers（自定义分词器）

1 安装 ik 分词器

安装 ik 分词器：注意版本要与 es 版本一致

注意：不能用默认 elasticsearch-plugin install xxx.zip 进行自动安装
对应 es 版本安装地址

//进入 es 容器内部 plugins 目录
docker exec -it 容器id /bin/bash

cd plugins/
wget https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.4.2/elasticsearch-analysis-ik-7.4.2.zip

//unzip 下载的文件
rm –rf *.zip
mv elasticsearch/ ik

//可以确认是否安装好了分词器
cd ../bin
elasticsearch plugin list  //即可列出系统的分词器

或者可以使用 secureFXP 将下载的 zip 文件（直接放解压后的也可以）放到本地的es的plugins 文件夹内

如果不知道运行的容器本地文件的位置，可执行以下命令

docker inspect <容器id>

2 测试分词器

使用默认

POST _analyze
{
  "text": "我是中国人"
}

使用分词器

POST _analyze
{
  "analyzer": "ik_smart",
  "text": "我是中国人"
}

另外一个分词器 ik_max_word

POST _analyze
{
  "analyzer": "ik_max_word",
  "text": "我是中国人"
}

能够看出不同的分词器，分词有明显的区别，所以以后定义一个索引不能再使用默认的 mapping 了，要手工建立 mapping, 因为要选择分词器。

3 自定义词库

修改/usr/share/elasticsearch/plugins/ik/config/中的 IKAnalyzer.cfg.xml

使用 http://192.168.128.130/fenci/myword.txt 的前提是配置了 nginx ，然后将文件 myword.txt 放在对应位置

修改后的 xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
	<comment>IK Analyzer 扩展配置</comment>
	<!--配置自己的扩展字典 -->
	<entry key="ext_dict"></entry>
	<!--配置自己的扩展停止词字典-->
	<entry key="ext_stopwords"></entry>
	<!--指定自定义词库 -->
	<entry key="remote_ext_dict">http://192.168.128.130/fenci/myword.txt</entry>
	<!--配置远程扩展停止词字典-->
	<!-- <entry key="remote_ext_stopwords">words_location</entry> -->
</properties>

原来的 xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
	<comment>IK Analyzer 扩展配置</comment>
	<!--用户可以在这里配置自己的扩展字典 -->
	<entry key="ext_dict"></entry>
	<!--用户可以在这里配置自己的扩展停止词字典-->
	<entry key="ext_stopwords"></entry>
	<!--用户可以在这里配置远程扩展字典 -->
	<!-- <entry key="remote_ext_dict">words_location</entry> -->
	<!--用户可以在这里配置远程扩展停止词字典-->
	<!-- <entry key="remote_ext_stopwords">words_location</entry> -->
</properties>

利用 nginx 发布静态资源,按照请求路径，创建对应的文件夹以及文件，放在 nginx 的 html 下，然后重启 es 服务器，重启 nginx。

在 kibana 中测试分词效果，更新完成后，es 只会对新增的数据用新词分词。历史数据是不会重新分词的。如果想要历史数据重新分词

需要执行：

POST my_index/_update_by_query?conflicts=proceed

5、Elasticsearch-Rest-Clien

1. 9300：TCP
   spring-data-elasticsearch:transport-api.jar；
   springboot 版本不同， transport-api.jar 不同，不能适配 es 版本
   7.x 已经不建议使用，8 以后就要废弃
2. 9200：HTTP
   JestClient： 非官方，更新慢
   RestTemplate： 模拟发 HTTP 请求，ES 很多操作需要自己封装，麻烦
   HttpClient： 同上
   Elasticsearch-Rest-Client： 官方 RestClient，封装了 ES 操作，API 层次分明，上手简单

最终选择 Elasticsearch-Rest-Client（elasticsearch-rest-high-level-client）

5.1 SpringBoot 整合

依赖

新建检索模块，只添加web依赖，因为此处的 ES 依赖最高只支持6.3版本

然后添加es整合依赖，注意版本需要和服务器上es的版本一致

<!-- 导入es的 rest-high-level-client -->
<dependency>
	<groupId>org.elasticsearch.client</groupId>
	<artifactId>elasticsearch-rest-high-level-client</artifactId>
	<version>7.4.2</version>
</dependency>

参考官网

添加后发现注入的 es 服务版本不对

是因为 springboot 里面指定了 es 版本

只需要自己指定一下版本即可

<properties>
    <java.version>1.8</java.version>
    <elasticsearch.version>7.12.1</elasticsearch.version>
    <spring-cloud.version>Greenwich.SR3</spring-cloud.version>
</properties>

需要注入nacos：common模块注入，配置文件，注册发现注解

<dependency>
    <groupId>afei.gulimall</groupId>
    <artifactId>gulimall-common</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>

配置类

参考官网

/**
 * 1、导入配置
 * 2、编写配置，给容器注入一个RestHighLevelClient
 * 3、参照API 官网进行开发
 */
@Configuration
public class GulimallElasticsearchConfig {
    public static final RequestOptions COMMON_OPTIONS;
    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
//        builder.addHeader("Authorization", "Bearer " + TOKEN);
//        builder.setHttpAsyncResponseConsumerFactory(
//                new HttpAsyncResponseConsumerFactory
//                        .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    @Bean
    public RestHighLevelClient esRestClient() {
        RestClientBuilder builder = null;
        builder = RestClient.builder(new HttpHost("192.168.56.10",
											 	   9200, "http"));
        RestHighLevelClient client = new RestHighLevelClient(builder);
        
//        RestHighLevelClient client = new RestHighLevelClient(
//                RestClient.builder(
//                        new HttpHost("localhost", 9200, "http"),
//                        new HttpHost("localhost", 9201, "http")));
        return client;
    }

}

使用

参照官方文档

简单 添加 / 更新 数据

@Autowired
private RestHighLevelClient client;

//测试注入是否成功
@Test
public void contextLoads() {
    System.out.println(client);
}

/**
 * 添加或者更新
 */
@Test
public void indexData() throws IOException {
    IndexRequest indexRequest = new IndexRequest("users");
    User user = new User();
    user.setAge(19);
    user.setGender("男");
    user.setUserName("张三");
    String jsonString = JSON.toJSONString(user);
    
    indexRequest.source(jsonString,XContentType.JSON);
    // 执行操作
    IndexResponse index = client.index(indexRequest, GulimallElasticsearchConfig.COMMON_OPTIONS);
    // 提取有用的响应数据
    System.out.println(index);
}

复杂检索

参照官方文档

@Test
public void searchTest() throws IOException {
    // 1、创建检索请求
    SearchRequest searchRequest = new SearchRequest();
    // 指定索引
    searchRequest.indices("bank");
    // 指定 DSL，检索条件
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

    sourceBuilder.query(QueryBuilders.matchQuery("address", "mill"));

    //1、2 按照年龄值分布进行聚合
    TermsAggregationBuilder aggAvg = AggregationBuilders.terms("ageAgg").field("age").size(10);
    sourceBuilder.aggregation(aggAvg);

    //1、3 计算平均薪资
    AvgAggregationBuilder balanceAvg = AggregationBuilders.avg("balanceAvg").field("balance");
    sourceBuilder.aggregation(balanceAvg);

    System.out.println("检索条件" + sourceBuilder.toString());

    searchRequest.source(sourceBuilder);

    // 2、执行检索
    SearchResponse searchResponse = client.search(searchRequest, GulimallElasticsearchConfig.COMMON_OPTIONS);

    // 3、分析结果
    System.out.println(searchResponse.toString());

    // 4、拿到命中的结果
    SearchHits hits = searchResponse.getHits();
    //搜索请求的匹配
    SearchHit[] searchHits = hits.getHits();
    for (SearchHit hit : searchHits) {
    	/**
    	 *   "_index" : "bank",
    	 *   "_type" : "account",
    	 *   "_id" : "970",
    	 *   "_score" : 6.5032897,
    	 *   "_source" : {}
    	 */
    	//hit.getIndex();hit.getType();hit.getId(); 等等
        // 拿到 _source 里面完整结果字符串
        String sourceAsString = hit.getSourceAsString();
        // 转换成实体类
        Accout accout = JSON.parseObject(sourceAsString, Accout.class);
        System.out.println("account:" + accout );
    }

    // 5、拿到聚合
    Aggregations aggregations = searchResponse.getAggregations();
//  for (Aggregation aggregation : aggregations.asList()) {
//  }
    // 6、通过先前名字拿到对应聚合
    Terms ageAgg1 = aggregations.get("ageAgg");
    for (Terms.Bucket bucket : ageAgg1.getBuckets()) {
        // 7、拿到结果
        String keyAsString = bucket.getKeyAsString();
        System.out.println("年龄:" + keyAsString);
        long docCount = bucket.getDocCount();
        System.out.println("个数：" + docCount);
    }
    Avg balanceAvg1 = aggregations.get("balanceAvg");
    System.out.println("平均薪资：" + balanceAvg1.getValue());
    System.out.println(searchResponse.toString());
}

执行结果：

accout:GulimallSearchApplicationTests.Accout(account_number=970, balance=19648, firstname=Forbes, lastname=Wallace, age=28, gender=M, address=990 Mill Road, employer=Pheast, email=forbeswallace@pheast.com, city=Lopezo, state=AK)accout:GulimallSearchApplicationTests.Accout(account_number=136, balance=45801, firstname=Winnie, lastname=Holland, age=38, gender=M, address=198 Mill Lane, employer=Neteria, email=winnieholland@neteria.com, city=Urie, state=IL)accout:GulimallSearchApplicationTests.Accout(account_number=345, balance=9812, firstname=Parker, lastname=Hines, age=38, gender=M, address=715 Mill Avenue, employer=Baluba, email=parkerhines@baluba.com, city=Blackgum, state=KY)accout:GulimallSearchApplicationTests.Accout(account_number=472, balance=25571, firstname=Lee, lastname=Long, age=32, gender=F, address=288 Mill Street, employer=Comverges, email=leelong@comverges.com, city=Movico, state=MT)年龄:38
个数:2
年龄:28
个数:1
年龄:32
个数:1
平均薪水:25208.0

ELK
Elasticsearch ：用于检索数据

Logstach：存储数据

Kibana：视图化查看数据

性能压测 - 压力测试

1、压力测试

使用压力测试，我们有希望找到很多种用其他测试方法更难发现的错误。有两种错误类型是：内存泄漏，并发与同步
有效的压力测试系统将应用以下这些关键条件：重复，并发，量级，随机变化

1.1 性能指标

• 响应时间（Response Time: RT）
  响应时间指用户从客户端发起一个请求开始，到客户端接收到从服务器端返回的响应结束，整个过程所耗费的时间
• HPS（Hits Per Second） ：每秒点击次数，单位是次/秒
• TPS（Transaction per Second）：系统每秒处理交易数，单位是笔/秒
• QPS（Query per Second）：系统每秒处理查询次数，单位是次/秒
  对于互联网业务中，如果某些业务有且仅有一个请求连接，那么TPS=QPS=HPS，一般情况下用 TPS 来衡量整个业务流程，用 QPS 来衡量接口查询次数，用 HPS 来表示对服务器单击请求。
• 无论 TPS、QPS、HPS,此指标是衡量系统处理能力非常重要的指标，越大越好，根据经验，一般情况下：
  金融行业：1000TPS~50000TPS，不包括互联网化的活动
  保险行业：100TPS~100000TPS，不包括互联网化的活动
  制造行业：10TPS~5000TPS
  互联网电子商务：10000TPS~1000000TPS
  互联网中型网站：1000TPS~50000TPS
  互联网小型网站：500TPS~10000TPS
• 最大响应时间（Max Response Time） 指用户发出请求或者指令到系统做出反应（响应）的最大时间
• 最少响应时间（Mininum ResponseTime） 指用户发出请求或者指令到系统做出反应（响应）的最少时间
• 90%响应时间（90% Response Time） 是指所有用户的响应时间进行排序，第 90%的响应时间
• 从外部看，性能测试主要关注如下三个指标
  • 吞吐量：每秒钟系统能够处理的请求数、任务数。
  • 响应时间：服务处理一个请求或一个任务的耗时。
  • 错误率：一批请求中结果出错的请求所占比例。

1.2 JMeter

JMeter 安装

下载地址
下载对应的压缩包，解压运行 jmeter.bat 即可

JMeter 压测示例

1、添加线程组

线程组参数详解：

• 线程数：虚拟用户数。一个虚拟用户占用一个进程或线程。设置多少虚拟用户数在这里也就是设置多少个线程数
• Ramp-Up Period(in seconds)准备时长：设置的虚拟用户数需要多长时间全部启动。如果线程数为 10，准备时长为 2，那么需要 2 秒钟启动 10 个线程，也就是每秒钟启动 5 个线程。
• 循环次数：每个线程发送请求的次数。如果线程数为 10，循环次数为 100，那么每个线程发送 100 次请求。总请求数为 10*100=1000 。如果勾选了“永远”，那么所有线程会一直发送请求，一到选择停止运行脚本。
• Delay Thread creation until needed：直到需要时延迟线程的创建。
• 调度器：设置线程组启动的开始时间和结束时间(配置调度器时，需要勾选循环次数为永远)
• 持续时间（秒）：测试持续时间，会覆盖结束时间
• 启动延迟（秒）：测试延迟启动时间，会覆盖启动时间
• 启动时间：测试启动时间，启动延迟会覆盖它。当启动时间已过，手动只需测试时当前时间也会覆盖它。
• 结束时间：测试结束时间，持续时间会覆盖它

2、添加 HTTP 请求

3、添加监听器

4、启动压测&查看分析结果

结果分析

• 有错误率同开发确认，确定是否允许错误的发生或者错误率允许在多大的范围内；
• Throughput 吞吐量每秒请求的数大于并发数，则可以慢慢的往上面增加；若在压测的机器性能很好的情况下，出现吞吐量小于并发数，说明并发数不能再增加了，可以慢慢的往下减，找到最佳的并发数；
• 压测结束，登陆相应的 web 服务器查看 CPU 等性能指标，进行数据的分析;
• 最大的 tps，不断的增加并发数，加到 tps 达到一定值开始出现下降，那么那个值就是最大的 tps。
• 最大的并发数：最大的并发数和最大的 tps 是不同的概率，一般不断增加并发数，达到一个值后，服务器出现请求超时，则可认为该值为最大的并发数。
• 压测过程出现性能瓶颈，若压力机任务管理器查看到的 cpu、网络和 cpu 都正常，未达到 90%以上，则可以说明服务器有问题，压力机没有问题。
• 影响性能考虑点包括：
  数据库、应用程序、中间件（tomact、Nginx）、网络和操作系统等方面
• 首先考虑自己的应用属于 CPU 密集型 还是 IO 密集型

1.3 JMeter Address Already in use 错误解决

windows 本身提供的端口访问机制的问题。
Windows 提供给 TCP/IP 链接的端口为 1024-5000，并且要四分钟来循环回收他们。就导致我们在短时间内跑大量的请求时将端口占满了。

解决：

1. cmd 中，用 regedit 命令打开注册表
2. 在HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\Tcpip\Parameters 目录下，
   1. 右击 parameters，添加一个新的 DWORD，名字为 MaxUserPort
   2. 然后双击 MaxUserPort，输入数值数据为 65534，基数选择十进制（如果是分布式运行的话，控制机器和负载机器都需要这样操作哦）
   3. 同理增加 TCPTimedWaitDelay，值为十进制 30
3. 修改配置完毕之后记得重启机器才会生效
   说明文档

1.4 性能监控-jvisualvm使用

Window+R ，输入 jvisualvm ，便可打开

idea 启动服务，在 jvisualvm 里面便可监控性能情况

安装插件方便查看 gc

工具->插件

如果 503 错误解决：
打开网址 https://visualvm.github.io/pluginscenters.html

cmd 查看自己的 jdk 版本，找到对应的

复制上面查询出来的链接。并重新设置上即可

2、性能优化

缓存 - 分布式锁

1、缓存使用-本地缓存

对于复杂的业务，已经不能够通过代码层面的优化和数据库层面的优化，来达到增加吞吐量的目的。这就想要使用到缓存。

哪些数据适合放入缓存？

• 即时性、数据一致性要求不高的
• 访问量大且更新频率不高的数据（读多，写少）

	data = cache.load(id);  //从缓存加载数据
	if(data == null){
		data = db.load(id);  //从数据库加载数据
		cache.put(id,data);  //保存到 cache 中
	}
	return data;

注意： 在开发中，凡是放入缓存中的数据我们都应该指定过期时间，使其可以在系统即使没有主动更新数据也能自动触发数据加载进缓存的流程。避免业务崩溃导致的数据永久不一致问题

2、整合 redis

pom.xml

注意此处引入的是 jedis 的客户端依赖

<!--引入redis-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <!--不加载自身的 lettuce-->
    <exclusions>
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<!--jedis-->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

默认使用的是 lettuce 客户端

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

配置文件配置 redis

spring:
  redis:
    host: 192.168.56.10
    port: 6379

测试：

使用 RedisTemplate 操作 redis

	@Autowired
	StringRedisTemplate stringRedisTemplate;
	
	@Test
	public void testStringRedisTemplate() {
	    stringRedisTemplate.opsForValue().set("hello","world_" + UUID.randomUUID().toString());
	    String hello = stringRedisTemplate.opsForValue().get("hello");
	    System.out.println("之前保存的数据是：" + hello);
	}

查看 springboot 自动配置类

3、优化三级分类获取

先从缓存中获取分类三级分类数据，如果没有再从数据库中查询，并且将查询结果以JSON字符串的形式存放到Reids中的

给缓存中放 json 字符串、拿出的是 json 字符串，还要逆转为能用的对象类型【序列化和反序列化】

修改原方法，改方法名为 getCatelogJsonFromDb

public Map<String, List<Catalog2Vo>> getCatelogJsonFromDb() {
    //查询所有的一级分类
    List<CategoryEntity> l1List = getLevel1Categorys();
    Map<String, List<Catalog2Vo>> map = l1List.stream().collect(Collectors.toMap(k-> k.getCatId().toString(), l1 -> {

        //根据一级分类查询二级分类信息
        List<CategoryEntity> l2entities = baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", l1.getCatId()));
        List<Catalog2Vo> l2List = null;
        if (l2entities != null) {
            l2List = l2entities.stream().map(l2 -> {
                Catalog2Vo catalog2Vo = new Catalog2Vo();
                catalog2Vo.setCatalog1Id(l1.getCatId().toString());
                catalog2Vo.setId(l2.getCatId().toString());
                catalog2Vo.setName(l2.getName());

                //根据二级分类查询三级分类信息
                List<CategoryEntity> l3entities = baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", l2.getCatId()));
                if (l3entities != null) {
                    List<Catalog2Vo.Catalog3Vo> l3List = l3entities.stream().map(l3 -> {
                        Catalog2Vo.Catalog3Vo catalog3Vo = new Catalog2Vo.Catalog3Vo(l3.getCatId().toString(), l2.getCatId().toString(), l3.getName());
                        return catalog3Vo;
                    }).collect(Collectors.toList());
                    catalog2Vo.setCatalog3List(l3List);
                }
                return catalog2Vo;
            }).collect(Collectors.toList());
        }
        return l2List;
    }));
    return map;
}

另编写方法 getCatalogJson()使用缓存

// 给缓存中放 json 字符串、拿出的是 json 字符串，还要逆转为能用的对象类型【序列化和反序列化】
@Override
public Map<String, List<Catalog2Vo>> getCatalogJson() {
    // 1、加入缓存逻辑，缓存中放的数据是 json 字符串：JSON 跨语言、跨平台兼容
    String catalogJson = stringRedisTemplate.opsForValue().get("CatalogJson");
    if (StringUtils.isEmpty(catalogJson)) {
        // 2、缓存没有，从数据库中查询
        Map<String, List<Catalog2Vo>> catelogJsonFromDb = getCatelogJsonFromDb();
        // 3、查询到数据，将数据转成 JSON 后放入缓存中
        String s = JSON.toJSONString(catelogJsonFromDb);
        stringRedisTemplate.opsForValue().set("CatalogJson",s);
        return catelogJsonFromDb;
    }
    // 4、从缓存中获取的json数据转换为我们指定的对象
    Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {});
    return result;
}

1）、压力测试出的内存泄露及解决

如果使用 lettuce 客户端，jmeter 测试高并发环境下会产生堆外内存溢出 OutOfDirectMemoryError

1、SpringBoot2.0以后默认使用 Lettuce作为操作redis的客户端，它使用netty进行网络通信
2、lettuce的bug导致netty堆外内存溢出，可以通过-Dio.netty.maxDirectMemory 进行设置 netty 堆内存
当指定了服务使用的VM options最大内存-Xmx300m，但是没有指定 netty 堆内存，netty 就默认使用 -Xmx300m

如果只是调大内存，只会延缓异常的出现
3、解决方案：不能使用 -Dio.netty.maxDirectMemory调大内存
1)、升级 lettuce客户端；
2)、切换使用jedis，不使用 lettuce
4、注意：
lettuce、jedis 操作redis的底层客户端，Spring再次封装。
springboot底层封装的时候将两个都封装为 redisTemplate，所以代码不需要修改

2）、加本地锁解决缓存击穿问题

修改代码：将 查库操作 锁起来

@Override
public Map<String, List<Catalog2Vo>> getCatalogJson() {
    // 1、加入缓存逻辑，缓存中放的数据是 json 字符串：JSON 跨语言、跨平台兼容
    String catalogJson = stringRedisTemplate.opsForValue().get("CatalogJson");
    if (StringUtils.isEmpty(catalogJson)) {
        // 2、缓存没有，从数据库中查询
        System.out.println("缓存不命中.....将要查询数据库...");
        Map<String, List<Catalog2Vo>> catelogJsonFromDb = getCatelogJsonFromDbLock();
        // 3、查询到数据，将数据转成 JSON 后放入缓存中
        String s = JSON.toJSONString(catelogJsonFromDb);
        stringRedisTemplate.opsForValue().set("CatalogJson",s);
        return catelogJsonFromDb;
    }
    System.out.println("缓存命中...直接返回...");
    // 4、从缓存中获取的json数据转换为我们指定的对象
    Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {});
    return result;
}

public Map<String, List<Catalog2Vo>> getCatelogJsonFromDbLock() {
    /**
     *  为了解决缓存击穿问题，给业务逻辑加锁：
     *  只要是同一把锁，就能锁住需要这个锁的所有线程
     *     1、synchronized(this)：SpringBoot所有组件在容器中都是单例的，所有可行。
     *      只不过分布式情况下，每个商品服务都有自己的一把锁
     *  TODO：本地锁：synchronized、JUC(lock),在分布式情况下想要锁住所有，必须使用分布式锁
     */
    synchronized (this){
        //得到锁以后，我们应该再去缓存中确定一次，如果没有才需要继续查询
        String catalogJSON = stringRedisTemplate.opsForValue().get("CatalogJson");
        if (!StringUtils.isEmpty(catalogJSON)){
            //如果缓存不为null直接缓存
            Map<String,List<Catalog2Vo>> result = JSON.parseObject(catalogJSON,new TypeReference<Map<String,List<Catalog2Vo>>>(){});
            return result;
        }
        System.out.println("查询了数据库。。。。。");
        //查询三级分类数据
        此处是业务逻辑代码++++++++++++++++++++++++
    }
}

使用 JMeter 进行高并发访问测试，发现并不是线程安全的

因为当 查库后，释放锁，在将结果放入缓存的这段时间里，有其他线程确认缓存没有，又再次查询了数据库，因此我们要将结果放入缓存也进行加锁

3）、锁时序问题

修改代码：将 查库操作、结果放入缓存的操作 锁起来

@Override
public Map<String, List<Catalog2Vo>> getCatalogJson() {
    // 1、加入缓存逻辑，缓存中放的数据是 json 字符串：JSON 跨语言、跨平台兼容
    String catalogJson = stringRedisTemplate.opsForValue().get("CatalogJson");
    if (StringUtils.isEmpty(catalogJson)) {
        // 2、缓存没有，从数据库中查询
        System.out.println("缓存不命中.....将要查询数据库...");
        Map<String, List<Catalog2Vo>> catelogJsonFromDb = getCatelogJsonFromDbLock();
        // 3、查询到数据，将数据转成 JSON 后放入缓存中
        // String s = JSON.toJSONString(catelogJsonFromDb);
        // stringRedisTemplate.opsForValue().set("CatalogJson",s);
        return catelogJsonFromDb;
    }
    System.out.println("缓存命中...直接返回...");
    // 4、从缓存中获取的json数据转换为我们指定的对象
    Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {});
    return result;
}

public Map<String, List<Catalog2Vo>> getCatelogJsonFromDbLock() {
    /**
     *  为了解决缓存击穿问题，给业务逻辑加锁：
     *  只要是同一把锁，就能锁住需要这个锁的所有线程
     *     1、synchronized(this)：SpringBoot所有组件在容器中都是单例的，所有可行。
     *      只不过分布式情况下，每个商品服务都有自己的一把锁
     *  TODO：本地锁：synchronized、JUC(lock),在分布式情况下想要锁住所有，必须使用分布式锁
     */
    synchronized (this){
        //得到锁以后，我们应该再去缓存中确定一次，如果没有才需要继续查询
        String catalogJSON = stringRedisTemplate.opsForValue().get("CatalogJson");
        if (!StringUtils.isEmpty(catalogJSON)){
            //如果缓存不为null直接缓存
            Map<String,List<Catalog2Vo>> result = JSON.parseObject(catalogJSON,new TypeReference<Map<String,List<Catalog2Vo>>>(){});
            return result;
        }
        System.out.println("查询了数据库。。。。。");
        //查询三级分类数据
        此处是业务逻辑代码++++++++++++++++++++++++

        //将结果放入缓存的操作也锁起来
        String s = JSON.toJSONString(map);
        stringRedisTemplate.opsForValue().set("CatalogJson",s);
        return map;
    }
}

使用 JMeter 进行高并发访问测试，解决问题

这里我们使用了双端检锁机制来控制线程的并发访问数据库。一个线程进入到临界区之前，判断缓存中是否有数据，进入到临界区后，再次判断缓存中是否有数据，这样做的目的是避免阻塞在临界区的多个线程，在其他线程释放锁后，重复进行数据库的查询和放缓存操作。

注：关于双端检锁机制的简单了解，可以参照：https://blog.csdn.net/maritimesun/article/details/7831065

4）、本地锁在分布式下的问题

通过观察日志，能够发现只有一个线程查询了数据库，其他线程都是直接从缓存中获取到数据的。所以在单体应用上实现了多线程的并发访问。

由于这里我们的“gulimall-product”就部署了一台，所以看上去一切祥和，但是在如果部署了多台，问题就出现了，主要问题就集中在我们所使用的锁上。我们锁使用的是“synchronized ”，这是一种本地锁，它只是在一台设备上有效，无法实现分布式情况下，锁住其他设备的相同操作。

我们现在的操作模型，表现为如下的形式：

将商品服务复制几个

启动四个服务，JMeter 测试访问配置的 nginx

我没有配置域名和nginx，所以将请求转给网关，通过网关负载均衡到 “gulimall” 服务的多个实例上

执行结果发现每次服务都会查询一次数据库，虽然都查一次，但是每个服务都查一次就有问题

4、分布式锁的使用——使用 redis

下面使用redis来实现分布式锁，使用的是SET key value [EX seconds] [PX milliseconds] [NX|XX]

参考文档

1）、阶段一

使用 redis 获取锁，若获取到锁则执行业务，若没有则等待重试

public Map<String, List<Catalog2Vo>> getCatelogJsonWithRedisLock1() {
    //1、占分布式锁。去redis占坑
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111");
    if (lock) {
        //2、加锁成功，执行业务
        Map<String, List<Catalog2Vo>> dataFromDB = getCatelogJsonFromDbWithCache();
        //3、删除锁
        stringRedisTemplate.delete("lock");
        return dataFromDB;
    }else {
        //4、没获取到锁，等待100ms重试
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return getCatelogJsonWithRedisLock1();  //自旋的方式回调
    }
}

service业务方法

// 给缓存中放 json 字符串、拿出的是 json 字符串，还要逆转为能用的对象类型【序列化和反序列化】
@Override
public Map<String, List<Catalog2Vo>> getCatalogJson() {
    // 1、加入缓存逻辑，缓存中放的数据是 json 字符串：JSON 跨语言、跨平台兼容
    String catalogJson = stringRedisTemplate.opsForValue().get("CatalogJson");
    if (StringUtils.isEmpty(catalogJson)) {
        // 2、缓存没有，从数据库中查询
        // 3、查询到数据，将数据转成 JSON 后放入缓存中（查询和放入缓存应该是原子操作，都被锁住）
        System.out.println("缓存不命中.....将要查询数据库...");
        Map<String, List<Catalog2Vo>> catelogJsonFromDb = getCatelogJsonWithLocalLock1();
        return catelogJsonFromDb;
    }
    System.out.println("缓存命中...直接返回...");
    // 4、从缓存中获取的json数据转换为我们指定的对象
    Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {
    });
    return result;
}

查询数据库（使用redis做缓存）

public Map<String, List<Catalog2Vo>> getCatelogJsonFromDbWithCache() {
    //得到锁以后，我们应该再去缓存中确定一次，如果没有才需要继续查询
    String catalogJSON = stringRedisTemplate.opsForValue().get("CatalogJson");
    if (!StringUtils.isEmpty(catalogJSON)) {
        //如果缓存不为null直接缓存
        Map<String, List<Catalog2Vo>> result = JSON.parseObject(catalogJSON, new TypeReference<Map<String, List<Catalog2Vo>>>() {});
        return result;
    }
    
    //业务逻辑：查询三级分类
    System.out.println("查询了数据库。。。。。");
    List<CategoryEntity> l1List = getLevel1Categorys();
    Map<String, List<Catalog2Vo>> dataFromDB = l1List.stream().collect(Collectors.toMap(k -> k.getCatId().toString(), l1 -> {
        //根据一级分类查询二级分类信息
        List<CategoryEntity> l2entities = baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", l1.getCatId()));
        List<Catalog2Vo> l2List = null;
        if (l2entities != null) {
            l2List = l2entities.stream().map(l2 -> {
                Catalog2Vo catalog2Vo = new Catalog2Vo();
                catalog2Vo.setCatalog1Id(l1.getCatId().toString());
                catalog2Vo.setId(l2.getCatId().toString());
                catalog2Vo.setName(l2.getName());
                //根据二级分类查询三级分类信息
                List<CategoryEntity> l3entities = baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", l2.getCatId()));
                if (l3entities != null) {
                    List<Catalog2Vo.Catalog3Vo> l3List = l3entities.stream().map(l3 -> {
                        Catalog2Vo.Catalog3Vo catalog3Vo = new Catalog2Vo.Catalog3Vo(l3.getCatId().toString(), l2.getCatId().toString(), l3.getName());
                        return catalog3Vo;
                    }).collect(Collectors.toList());
                    catalog2Vo.setCatalog3List(l3List);
                }
                return catalog2Vo;
            }).collect(Collectors.toList());
        }
        return l2List;
    }));
    //将结果放入缓存的操作也锁起来
    String s = JSON.toJSONString(dataFromDB);
    stringRedisTemplate.opsForValue().set("CatalogJson", s);
    return dataFromDB;
}

• 问题：
  第一步加锁成功 setnx 占好了位之后
  业务代码异常或者程序在页面过程中宕机，没有执行删除锁逻辑，这就造成了死锁
• 解决：
  设置锁的自动过期，即使没有删除，会自动删除

2）、阶段二

设置锁的自动过期，即使没有删除，会自动删除

public Map<String, List<Catalog2Vo>> getCatelogJsonWithRedisLock2() {
    //1、占分布式锁。去redis占坑
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111");
    if (lock) {
        //2、设置过期时间
        stringRedisTemplate.expire("lock",30, TimeUnit.SECONDS);
        //3、加锁成功，执行业务
        Map<String, List<Catalog2Vo>> dataFromDB = getCatelogJsonFromDbWithCache();
        //4、删除锁
        stringRedisTemplate.delete("lock");
        return dataFromDB;
    }else {
        //5、没获取到锁，等待100ms重试（已省略）
        return getCatelogJsonWithRedisLock2();  //自旋的方式回调
    }
}

• 问题：
  第二步正要设置过期时间之前
  设置过期时间之前宕机，又死锁了
• 解决：
  设置锁的自动过期和占位（加锁）必须是原子的，redis 支持 setnx ex 命令

3）、阶段三

设置锁的自动过期和占位（加锁）必须是原子的，redis 支持 setnx ex 命令

public Map<String, List<Catalog2Vo>> getCatelogJsonWithRedisLock3() {
    //1、占分布式锁。去redis占坑
    //设置过期时间和加锁是一个原子操作
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "1111",30, TimeUnit.SECONDS);
    if (lock) {
        //设置过期时间 stringRedisTemplate.expire("lock",30, TimeUnit.SECONDS);
        //2、加锁成功，执行业务
        Map<String, List<Catalog2Vo>> dataFromDB = getCatelogJsonFromDbWithCache();
        //3、删除锁
        stringRedisTemplate.delete("lock");
        return dataFromDB;
    }else {
        //4、没获取到锁，等待100ms重试（已省略）
        return getCatelogJsonWithRedisLock3();  //自旋的方式回调
    }
}

• 问题：
  第三步删除锁直接删除？？？
  如果由于业务时间很长，锁自己过期了，我们直接删除，有可能把别人正在持有的锁删除了
• 解决：
  占锁的时候，值指定为 uuid，每个人匹配确定是自己的锁才删除

4）、阶段四

占锁的时候，值指定为 uuid，每个人匹配确定是自己的锁才删除

public Map<String, List<Catalog2Vo>> getCatelogJsonWithRedisLock4() {
    //1、占分布式锁。去redis占坑
    //设置过期时间和加锁是一个原子操作
    String uuid = UUID.randomUUID().toString();
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,30, TimeUnit.SECONDS);
    if (lock) {
        //2、加锁成功，执行业务
        Map<String, List<Catalog2Vo>> dataFromDB = getCatelogJsonFromDbWithCache();
        //3、删除锁
        //要确定删除的是我们自己加的锁
        String lockValue = stringRedisTemplate.opsForValue().get("lock");
        if (uuid.equals(lockValue)) {
            stringRedisTemplate.delete("lock");
        }
        return dataFromDB;
    }else {
        //4、没获取到锁，等待100ms重试（已省略）
        return getCatelogJsonWithRedisLock3();  //自旋的方式回调
    }
}

• 问题：
  第三步当判断确定是自己的锁之后，删除之前
  如果正好判断是当前锁，正要删除锁的时候，锁已经过期，别人已经设置到了新的值。那么我们删除的是别人的锁
• 解决：
  删除锁必须保证原子性。使用 redis + lua 脚本完成

5）、阶段五

删除锁必须保证原子性。使用 redis + lua 脚本完成

public Map<String, List<Catalog2Vo>> getCatelogJsonWithRedisLock5() {
    //1、占分布式锁。去redis占坑
    //设置过期时间和加锁是一个原子操作
    String uuid = UUID.randomUUID().toString();
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,30, TimeUnit.SECONDS);
    if (lock) {
        Map<String, List<Catalog2Vo>> dataFromDB;
        try {
            //2、加锁成功，执行业务
            dataFromDB = getCatelogJsonFromDbWithCache();
        } finally {
            //3、删除锁
            //获取锁值对比 + 对比成功删除锁 =是一个原子操作，使用Lua脚本解锁
            String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then\n" +
                    "    return redis.call(\"del\",KEYS[1])\n" +
                    "else\n" +
                    "    return 0\n" +
                    "end";
            //删除锁
            Long lock1 = stringRedisTemplate.execute(
                    new DefaultRedisScript<>(script, Long.class),
                    Arrays.asList("lock"),
                    uuid);
        }
        return dataFromDB;
    }else {
        //4、没获取到锁，等待100ms重试（已省略）
        return getCatelogJsonWithRedisLock5();  //自旋的方式回调
    }
}

实现，加锁【占位+过期时间】和删除锁【判断+删除】的原子性。
未实现，锁的自动续期

官网文档上详细说明了 不推荐使用 setnx 来实现分布式锁，在 Java 语言环境下推荐使用 Redisson

5、Redisson

1）、配置

中文文档

商品服务导入依赖

<!--以后使用 redisson 作为分布锁，分布式对象等功能-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.12.0</version>
</dependency>

配置类

@Configuration
public class MyRedissonConfig {
    /**
     * 所有对Redisson的使用都是通过 RedissonClient 对象
     */
    @Bean(destroyMethod = "shutdown")
    public RedissonClient redisson(){
        // 1 创建配置
        Config config = new Config();
        //注意：用"rediss://"或者"redis://"
        config.useSingleServer().setAddress("redis://150.158.171.144:6379");
        // 2 根据Config创建RedissonClient实例
        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }
}

测试是否注入了 RedissonClient

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class GulimallProductApplicationTests {
	
	@Autowired
    RedissonClient redissonClient;

    @Test
    public void redissonClient(){
        System.out.println(redissonClient);
    }
}

2）、Redisson - Lock 锁测试 & Redisson - Lock 看门狗原理 - Redisson 如何解决死锁

Redison分布式锁

@ResponseBody
@GetMapping("hello")
public String hello(){
    // 1 获取一把锁：只要锁的名字一样，就是同一把锁
    RLock mylock = redisson.getLock("mylock");
    // 2 加锁
    mylock.lock(); //是阻塞式等待，类似于JUC中synchronized阻塞等待，不需要手动回调
    // 3 执行业务逻辑
    try {
        System.out.println("加锁成功，执行业务..." + Thread.currentThread().getId());
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        // 4 解锁
        //假如解锁代码没有运行，reidsson会不会出现死锁
        System.out.println("释放锁...." + Thread.currentThread().getId());
        mylock.unlock();
    }
    return "";
}

提问：假设当一个服务执行加锁之后宕机，未执行解锁，是否会死锁？
—— 在加锁后，即便我们没有释放锁，也会自动的释放锁，这是因为在 Redisson 中会为每个锁加上 “leaseTime”，默认是30秒

提问：假设业务执行所需要的时间超过 30 秒，锁会不会被自动删掉？
—— 锁的自动续期，如果业务超长，运行期间自动给锁续上新的30s，不用担心业务时间长，锁自动过期后被删掉
—— 加锁的业务只要运行完成，就不会给当前锁续期，即使不手动解锁，锁默认会在30s以后自动删除

提问：锁的过期时间只能是 30s 吗？我偏要 40s
—— 加锁时可以指定过期时间

mylock.lock(40, TimeUnit.SECONDS); //40s 后自动删除

提问：在锁到期后，是否会自动续期？
—— 如果传递了锁的超时时间，就发送给 redis 执行脚本，进行占锁，默认超时就是我们指定的时间（在指定了超时时间后，不会进行自动续期，即便业务仍然在执行，时间到了后锁会自动删除）
—— 如果没有指定锁的超时时间，就是用 LockWatchchdogTimeout 看门狗的默认时间 30s
并且只要占锁成功，就会启动一个定时任务，【重新给锁设置过期时间，新的过期时间就是看门狗的默认时间】,每隔10s（看门狗时间 / 3）就自动续期

源码：看门狗原理

3）、Redisson - 读写锁

读写锁保证一定能读取到最新数据，修改期间，写锁是一个排他锁（互斥锁，独享锁）读锁是一个共享锁

写锁没释放读锁就必须等待，只要有写的存在，都必须等待

• 读 + 读 相当于无锁，并发读，只会在 reids中记录好，所有当前的读锁，他们都会同时加锁成功
• 写 + 读 等待写锁释放
• 写 + 写 阻塞方式
• 读 + 写 有读锁，写也需要等待

代码如下：

@GetMapping("write")
@ResponseBody
public String write(){
    //改数据加写锁，读数据加读锁
    RReadWriteLock lock = redisson.getReadWriteLock("rw-lock");
    RLock writeLock = lock.writeLock();
    String s = "";
    try {
        writeLock.lock();
        Thread.sleep(1000);
        s = UUID.randomUUID().toString();
        stringRedisTemplate.opsForValue().set("writeValue",s);
        System.out.println("写锁加锁成功" + Thread.currentThread().getId());
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        writeLock.unlock();
        System.out.println("写锁释放成功" + Thread.currentThread().getId());
    }
    return s;
}

@GetMapping("read")
@ResponseBody
public String read(){
    //获取一把锁：只要锁的名字一样，就是同一把锁
    RReadWriteLock lock = redisson.getReadWriteLock("rw-lock");
    RLock readLock = lock.readLock();
    String s = "";
    try {
        readLock.lock();
        Thread.sleep(1000);
        s = stringRedisTemplate.opsForValue().get("writeValue");
        System.out.println("读锁成功" + Thread.currentThread().getId());
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        readLock.unlock();
        System.out.println("读锁释放" + Thread.currentThread().getId());
    }
    return s;
}

4）、Redisson - 信号量

先在 redis 里面新建 键 count 值为 3

@GetMapping("park")
@ResponseBody
public String park(){
    RSemaphore count = redisson.getSemaphore("count");
    // count.acquire();  //获取一个信号，获取一个值，占用一个车位
    boolean b = count.tryAcquire();
    if (b){
        //执行业务
    }else {
        return "error";
    }
    return b+"";
}

@GetMapping("go")
@ResponseBody
public String go(){
    RSemaphore count = redisson.getSemaphore("count");
    count.release(1);  //释放一个车位
    return "OK";
}

访问一次 http://localhost:10000/go 键值就会 +1

访问一次 http://localhost:10000/park 键值就会 -1

5）、Redisson - 闭锁

@GetMapping("lockDoor")
@ResponseBody
public String lockDoor() throws InterruptedException {
    RCountDownLatch lockDoor = redisson.getCountDownLatch("lockDoor");
    lockDoor.trySetCount(5);    //总数为5
    lockDoor.await();
    return "OK";
}

@GetMapping("leave")
@ResponseBody
public String leave(){
    RCountDownLatch lockDoor = redisson.getCountDownLatch("lockDoor");
    lockDoor.countDown();   //计数器减一
    return "OK";
}

当首次访问 http://localhost:10000/lockDoor 时，redis 里面会自己新建键 lockDoor 值为 5

访问一次 http://localhost:10000/leave ，值就会 -1

当值为 0 时就会自动删除此键 lockDoor

6）、Redission - 缓存一致性解决

优化三级分类获取，使用 redisson 实现分布式锁

//使用redisson
public Map<String, List<Catalog2Vo>> getCatelogJsonWithRedisson(){
    // 注意锁的名字。锁的粒度越细越快
    RLock lock = redissonClient.getLock("catelogJson-lock");
    lock.lock();
    Map<String, List<Catalog2Vo>> dataFromDB;
    try {
    	//在查询数据库的基础上，使用了redis做缓存，将查询的结果放入缓存
    	//（如果缓存有数据将不会查询数据库，会导致当数据库修改而缓存未改时，数据不一致问题）
        dataFromDB = getCatelogJsonFromDbWithCache();
    } finally {
        lock.unlock();
    }
    return dataFromDB;
}

缓存一致性是为了解决数据库和缓存的数据不同步问题的，有两种解决方法

方法一：双写模式（在写数据库方法中同时写缓存）

两个线程写 最终只有一个线程写成功，后写成功的会把之前写的数据给覆盖，这就会造成脏数据

解决方法：可以给写数据库和写缓存操作加锁，使其成为一个原子性操作。
但是即便如此，数据库和缓存之间总会有不一致问题。
可以给缓存加上过期时间，即缓存过期后又能有最新数据。
双写模式：读到最新数据有延迟，是暂时的脏数据——最终一致性

方法二：失效模式（在写数据库方法中同时删除缓存）

三个连接：
一号连接 写数据库 然后删缓存
二号连接 写数据库时网络连接慢，还没有写入成功
三号连接 发现无缓存便直接读取数据库，读到的是一号连接写入的数据，此时 二号连接写入数据成功并删除了缓存，三号开始读取结束更新缓存发现更新的是二号的缓存，但是更新的是一号数据

缓存数据一致性解决方案

无论是双写模式还是失效模式，都会到这缓存不一致的问题，即多个实力同时更新会出事，怎么办？

1、如果是用户纯度数据（订单数据、用户数据），这并发几率很小，几乎不用考虑这个问题，缓存数据加上过期时间，每隔一段时间触发读的主动更新即可
2、如果是菜单，商品介绍等基础数据，也可以去使用 canal 订阅，binlog 的方式
3、缓存数据 + 过期时间也足够解决大部分业务对缓存的要求
4、通过加锁保证并发读写，写写的时候按照顺序排好队，读读无所谓，所以适合读写锁，（业务不关心脏数据，允许临时脏数据可忽略）

总结：

• 我们能放入缓存的数据本来就不应该是实时性、一致性要求超高的。所以缓存数据的时候加上过期时间，保证每天拿到当前的最新值即可
• 我们不应该过度设计，增加系统的复杂性
• 遇到实时性、一致性要求高的数据，就应该查数据库，即使慢点

6）、Redission - Redlock

高级的分布式锁算法 Redlock，来处理集群部署分布式锁问题

如果线程一在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时，master节点发生故障，一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦，但线程一也已经拿到锁了，锁的安全性就没了。

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）
4. 如果取到了锁，key的真正有效时间就变啦，需要减去获取锁所使用的时间。
5. 如果获取锁失败（没有在至少N/2+1个master实例取到锁，有或者获取锁时间已经超过了有效时间），客户端要在所有的master节点上解锁（即便有些master节点根本就没有加锁成功，也需要解锁，以防止有些漏网之鱼）。

简化下步骤就是：

1. 按顺序向5个master节点请求加锁
2. 根据设置的超时时间来判断，是不是要跳过该master节点。
3. 如果大于等于3个节点加锁成功，并且使用的时间小于锁的有效期，即可认定加锁成功啦。
4. 如果获取锁失败，解锁！

6、Spring Cache

1）、简介

Spring 从 3.1 开始定义了 org.springframework.cache.Cache 和 org.sprngframework.cache.CacheManager 接口来统一不同的缓存技术，并支持使用 JCache（JSR-107）注解简化我们的开发。

Cache 接口为缓存的组件规范定义，包含缓存的各种操作集合；Cache 接口下 Spring 提供了各种 XXXCache 的实现：如 RedisCache、EhCache、ConcrrentMapCache 等等

每次调用需要缓存功能的方法时，Spring 会检查指定参数的指定目标方法是否已经被调用过。如果有就直接从缓存中获取方法调用后的结果，如果没有就调用方法并缓存结果后返回给用户，下次调用直接从缓存中获取

使用 Spring 缓存抽象时我们需要关注以下两点：
1、确定方法需要被缓存以及他们的的缓存策略
2、从缓存中读取之前缓存存储的数据

官网地址

2）、整合 SpringCache

引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-cache</artifactId>
</dependency>

（因为使用redis缓存，还需引入redis依赖）

<!--引入redis-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <!--不加载自身的 lettuce-->
    <exclusions>
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<!--jedis-->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

编写配置

/** 
 * 8、整合SpringCache简化缓存开发
 *      1）、依赖
 *         spring-boot-starter-cache、spring-boot-starter-data-redis
 *      2）、写配置
 *          1 自动配置了哪些
 *            CacheAutoConfiguration里的CacheConfigurationImportSelector()方法会根据类型导入 RedisCacheConfiguration
 *            RedisCacheConfiguration里已经配置 RedisCacheManager
 *            并在 determineConfiguration()方法里指明若有 RedisCacheConfiguration 则使用里面的配置，若无则使用默认配置
 *          2 配置文件指定使用redis作为缓存
 *            Spring.cache.type=redis
 *      3）、测试使用缓存
 *          1 开启缓存功能  @EnableCaching
 *          2 只需使用注解即可完成缓存操作
 *              @Cacheable：触发将数据保存到缓存的操作
 *              @CacheEvict：触发将数据从缓存删除的操作，缓存一致性解决中的失效模式
 *              @CachePut：不影响方法执行更新缓存，缓存一致性解决中的双写模式
 *              @Caching：组合以上多个操作
 *              @CacheConfig：在类级别共享缓存的相同配置
 *      4)、原理：
 *          CacheAutoConfiguration 导入 RedisCacheConfiguration ——> 其中自动配置了缓存管理器 RedisCacheManager
 *          RedisCacheManager 里调用 determineConfiguration() 方法初始化所有缓存，每个缓存决定使用什么配置 ——>
 *          如果 RedisCacheConfiguration 有就用已有的，若无就用默认配置 defaultCacheConfig() ——>
 *          想改缓存的配置，只需要给容器中放一个 RedisCacheConfiguration，就会应用到当前RedisCacheManager管理的所有缓存分区中
 */

启动类上使用注解 @EnableCaching

配置文件配置 redis 地址，并且指定 redis 作为缓存

spring:
  # redis
  redis:
    host: 111.111.111.111
    port: 6379
  # 配置使用redis作为缓存
  cache:
    type: redis

3）、@Cacheable - 存入缓存

/**
 * 1、每个需要缓存的数据都要指定缓存分区名（按照业务类型分）
 * 2、 @Cacheable("categorys")
 *      表示当前方法结果需要缓存，若缓存中有则直接取缓存，不执行方法
 *      若缓存中无，会调用方法，最后将结果存入缓存
 * 3、默认行为
 *      1）如果缓存命中，方法不会执行
 *      2）redis中 key 默认自动生成：缓存名::SimpleKey[]（自动生成的key值）此处即为 【categorys::SimpleKey []】
 *      3）redis中 value 默认使用jdk序列化机制，将序列化后的数据存到redis
 *      4）默认ttl时间 -1 ，永不过期
 * 4、自定义行为
 *      1）指定生成的缓存的 key
 *          注解属性key指定，传入SpEL表达式，参考 https://docs.spring.io/spring-framework/docs/5.2.10.RELEASE/spring-framework-reference/integration.html#cache-spel-context
 *          key = "'level1Categorys'"   即为 【categorys::level1Categorys】
 *          key = "#root.method.name"   即为 【categorys::getLevel1Categorys】
 *      2）指定缓存数据的存活时间，毫秒为单位
 *          配置文件中指定 Spring.cache.redis.time-to-live=3600000
 *      3）指定缓存的数据以 json 格式存储
 *          自定义配置 RedisCacheConfiguration 即可
 *      4）配置文件中的一些自定义配置
 *          Spring.cache.redis.cache-null-values=true  是否缓存空值，可防止缓存穿透
 *          Spring.cache.redis.key-prefix=CACHE_  缓存key值前缀，若使用指定前缀即为 【CACHE_getLevel1Categorys】，若不指定，默认缓存分区名为前缀
 *          Spring.cache.redis.use-key-prefix=true   缓存key值是否使用前缀，若为false，即为 【getLevel1Categorys】
 */

每个需要缓存的数据都要指定缓存分区名（按照业务类型分）

使用注解 @Cacheable("categorys")
表示当前方法的结果需要缓存，若缓存中有则直接取缓存，不执行方法
若缓存中无，则会调用方法，最后将结果存入缓存

@Cacheable("categorys")
@Override
public List<CategoryEntity> getLevel1Categorys() {
    return baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", 0));
}

启动服务，查看 redis

发现当不进行任何配置时，会有一些 默认行为 ：

1. 如果缓存命中，方法不会执行
2. redis中 key 默认自动生成： 缓存名::SimpleKey[] （自动生成的key值）
   此处即为 categorys::SimpleKey []
3. redis中 value 默认使用 jdk序列化机制 ，将序列化后的数据存到redis
4. 默认 ttl时间 -1 ，永不过期

以下是 自定义行为：

1、指定生成的缓存的 key

使用注解属性 key 指定，传入SpEL表达式，参考官方文档

key = “‘level1Categorys’” 即为 categorys::level1Categorys

@Cacheable(value = {"categorys"},key = "'level1Categorys'")
@Override
public List<CategoryEntity> getLevel1Categorys() {
    return baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", 0));
}

key = “#root.method.name” 即为 categorys::getLevel1Categorys

@Cacheable(value = {"categorys"}, key = "#root.method.name")
@Override
public List<CategoryEntity> getLevel1Categorys() {
    return baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", 0));
}

2、指定缓存数据的存活时间，毫秒为单位

配置文件中指定 Spring.cache.redis.time-to-live=3600000

spring:
  # redis
  redis:
    host: 111.111.111.111
    port: 6379
  # 配置使用redis作为缓存
  cache:
    type: redis
    # 指定缓存数据的存活时间，毫秒为单位
    redis:
      time-to-live: 3600000

3、指定缓存的数据以 json 格式存储

自定义配置 RedisCacheConfiguration 即可

@Configuration
public class MyCacheConfig {

    @Bean
    RedisCacheConfiguration redisCacheConfiguration(CacheProperties cacheProperties){
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
        
        //自定义配置 key-value 的序列化机制
        config = config.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()));
        config = config.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
        
        //注意：配置类会覆盖配置文件的配置，若想配置文件生效，需要以下代码
        CacheProperties.Redis redisProperties = cacheProperties.getRedis();
        if (redisProperties.getTimeToLive() != null) {
            config = config.entryTtl(redisProperties.getTimeToLive());
        }
        if (redisProperties.getKeyPrefix() != null) {
            config = config.prefixKeysWith(redisProperties.getKeyPrefix());
        }
        if (!redisProperties.isCacheNullValues()) {
            config = config.disableCachingNullValues();
        }
        if (!redisProperties.isUseKeyPrefix()) {
            config = config.disableKeyPrefix();
        }
        return config;
    }
}

4、配置文件中的一些自定义配置

Spring.cache.redis.cache-null-values=true
是否缓存空值，可防止缓存穿透

Spring.cache.redis.key-prefix=CACHE_
缓存key值前缀，若使用指定前缀即为 CACHE_getLevel1Categorys，若不指定，默认缓存分区名为前缀 【建议使用默认前缀】

如果指定了前缀，则在 redis 可视化界面里面就不会有分区显示，如果使用默认分区名为前缀，就会有分区显示，如下：

Spring.cache.redis.use-key-prefix=true
缓存key值是否使用前缀，若为false，即为 getLevel1Categorys

spring:
  # redis
  redis:
    host: 111.111.111.111
    port: 6379
  # 配置使用redis作为缓存
  cache:
    type: redis
    redis:
      time-to-live: 3600000
      # 是否缓存空值，可防止缓存穿透
      cache-null-values: true
      key-prefix: CACHE_
      use-key-prefix: true

4）、@CacheEvict - 失效模式

@Cacheable 注解解决了将数据库数据存入缓存，并且是从缓存中直接获取数据

当数据库更新时需要同步更新缓存，有以下两种方式：
@CacheEvict - 失效模式
@CachePut - 双写模式

数据库修改之后会返回最新的数据结果，可以使用双写模式，将返回的结果放入缓存

此处我们的修改方法没有返回数据，所以演示 @CacheEvict 的使用

1、 仅仅更新一个缓存

@CacheEvict(value = {"categorys"}, key = "'getLevel1Categorys'")
@Transactional
@Override
public void updateDetail(CategoryEntity category){}

注意这里 key 也是使用表达式，所以双引号里面还要有单引号

当更新数据后，发现 redis 里面缓存被删除

2、 更新多个缓存

可以使用注解 @Caching ：表示组合多个操作

public @interface Caching {
   Cacheable[] cacheable() default {};
   CachePut[] put() default {};
   CacheEvict[] evict() default {};
}

@Caching(evict = {
     @CacheEvict(value = {"categorys"}, key = "'getLevel1Categorys'"),
     @CacheEvict(value = {"categorys"}, key = "'getCatalogJson'")
})
@Transactional
@Override
public void updateDetail(CategoryEntity category)

3、更新整个分区的缓存

除了使用组合多个操作的方式，还可以直接 删除整个分区的缓存，如下

@CacheEvict(value = {"categorys"},allEntries = true)
@Transactional
@Override
public void updateDetail(CategoryEntity category)

5）、原理与不足

• 缓存穿透：查询一个null数据
  解决：可以通过配置文件缓存空数据spring.cache.redis.cache-null-values=true
• 缓存击穿：大量并发进来同时查询一个正好过期的数据
  解决：加锁。 默认是无加锁
• 缓存雪崩：大量的key同时过期
  解决：加上随机时间，spring.cache.redis.time-to-live

只有注解 @Cacheable 可以加锁，通过属性 sync = true 实现，默认 false

@Cacheable(value = {"categorys"},key = "#root.methodName",sync = true)

总结：

常规数据（读多写少，即时性，一致性要求不高的数据）完全可以使用SpringCache 写模式（ 只要缓存数据有过期时间就足够了）

异步&线程池

1、线程回顾

1）、初始化线程的四种方式

1. 继承 Thread

   ThreadThr threadThr = new ThreadThr();
   threadThr.start();
   
   public static class ThreadThr extends Thread {
       @Override
       public void run() {
           System.out.println("当前线程" + Thread.currentThread().getId());
           int i = 10 / 3;
           System.out.println("运行结果" + i);
       }
   }
   

2. 实现 Runnable

   Thread thread = new Thread(new ThreadRun());
   thread.start();
   
   public static class ThreadRun implements Runnable {
       @Override
       public void run() {
           System.out.println("当前线程" + Thread.currentThread().getId());
           int i = 10 / 3;
           System.out.println("运行结果" + i);
       }
   }
   

3. 实现 Callable 接口 + FutureTask（可以拿到返回结果，可以处理异常）

   FutureTask<Integer> futureTask = new FutureTask<>(new ThreadCall());
   new Thread(futureTask).start();
   //阻塞等待整个线程执行完成，获取返回结果
   Integer integer = futureTask.get();
   System.out.println(integer);
   
   public static class ThreadCall implements Callable<Integer> {
       @Override
       public Integer call() throws Exception {
           System.out.println("当前线程" + Thread.currentThread().getId());
           int i = 10 / 3;
           System.out.println("运行结果" + i);
           return i;
       }
   }
   

4. 线程池
   以上三种启动线程方式都不用，将所有的多线程异步任务都交给线程池执行

   ExecutorService service = Executors.newFixedThreadPool(10);
   service.execute(new ThreadRun());
   //或者
   new ThreadPollExecutor(corePoolSize,maximumPoolSize,keepAliveTime,TimeUnit,unit,workQueue,threadFactory,handler);
   

2）、线程池详解

线程池【ExecutorService】

创建

通过 Executors，或者 new ThreadPoolExecutor()。后者需要传入七大参数

七大参数：

1. corePoolSize —— 核心线程数；线程池创建好后就准备就绪的线程数量，等待接收异步任务去执行

2. maximumPoolSize —— 最大线程数量；控制资源

3. keepAliveTime —— 存活时间；当线程空闲一定的时间，就释放这些线程【注意是核心线程外的线程】

4. unit —— 时间单位

5. workQueue —— 阻塞队列；如果任务很多，将多的任务放在队列里，只要有线程空闲，就会去队列里取新的任务执行

6. threadFactory —— 线程的创建工厂

   Executors.defaultThreadFactory()  //若不满足使用还可手动重写
   

7. handler —— 拒绝策略 RejectedExecutionHandler；如果队列满了，无法执行更多的任务，按照指定的拒绝策略来拒绝接收任务

   new ThreadPoolExecutor.AbortPolicy()  //直接丢弃任务，会抛异常
   

示例：

ThreadPoolExecutor executor = new ThreadPoolExecutor(5,
       200,
       10,
       TimeUnit.SECONDS,
       new LinkedBlockingDeque<>(100000),
       Executors.defaultThreadFactory(),  //若不满足使用还可手动重写
       new ThreadPoolExecutor.AbortPolicy());  //直接丢弃任务，会抛异常

运行流程

1. 线程池创建，准备好 core 数量的核心线程，准备接受任务
2. core 满了，就将再进来的任务放入阻塞队列中。空闲的 core 就会自己去阻塞队列获取任务执行
3. 阻塞队列满了，就直接开新线程执行，最大只能开到 max 指定的数量
4. max 满了，就用RejectedExecutionHandler拒绝任务
5. max 都执行完成。max-core 数量空闲的线程会在keepAliveTime指定的时间后自动销毁。最终保持到core大小

细节

阻塞队列 new LinkedBlockingDeque<>() 默认是Integer的最大值，可能会导致内存不够，所以一般需要指定值

提问：一个线程池 core 7； max 20 ，queue：50，100 并发进来怎么分配的？
—— 7个立即执行，50 个进入队列排队，再多开 13 个继续执，剩下 30 个默认拒绝策略。

常见的 4 种线程池

1. Executors.newCachedThreadPool(); core为0，所有都可回收
2. Executors.newFixedThreadPool(); 固定大小，core为max，所有都不可回收
3. Executors.newScheduledThreadPool(); 定时任务线程池
4. Executors.newSingleThreadExecutor(); 单线程的线程池，core和max都为1，后台从队列中获取任务，挨个执行

2、CompletableFuture 异步编排

通过线程池性能稳定，也可以获取执行结果，并捕获异常，但是，在业务复杂情况下，一个异步调用可能 会依赖另一个异步调用的执行结果

Future 是 Java 5 添加的类，用来描述一个异步计算的结果

在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture，提供了非常强大的
Future 的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类，都可以获取线程的执行结果

1）、创建异步对象

1. runXxxx 都是没有返回结果的，supplyXxx 都是可以获取返回结果的
2. 可以传入自定义的线程池，否则就用默认的线程池；

示例：

无返回值，传入 void run()

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    //无返回值  void run();
    CompletableFuture.runAsync(()->{
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 3;
        System.out.println("运行结果" + i);
    },service);
    System.out.println("main......end......");
}

有返回值，传入 T get()

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    //有返回值  T get();
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 2;
        return i;
    }, service);
    Integer integer = future.get();  //阻塞式等待
    System.out.println("结果是："+integer);
    System.out.println("main......end......");
}

2）、完成回调和异常感知

1. whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况。
2. 方法不以 Async 结尾，意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）
3. whenComplete 和 whenCompleteAsync 的区别：
   1. whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
   2. whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。
4. 注意：二者都只能是由 有返回结果的 CompletableFuture 调用

示例：

whenComplete 传入 void accept(T t, U u)

接收结果和异常，当有异常时结果为 null

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 0;
        return i;
        //void accept(T t, U u)
    }, service).whenCompleteAsync((result,exception)->{
        // 虽然能得到异常信息，但是没法修改返回的数据
        System.out.println("异步任务成功完成了...结果是：" +result + "异常是：" + exception);
        //R apply(T t);
    });
    System.out.println("main......end......");
}

exceptionally 传入 R apply(T t)

可以感知到异常，同时返回默认值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 0;
        return i;
    }, service).whenCompleteAsync((result,exception)->{    //void accept(T t, U u)
        // 虽然能得到异常信息，但是没法修改返回的数据
        System.out.println("异步任务成功完成了...结果是：" +result + "异常是：" + exception);
    }).exceptionally((excep)->{    //R apply(T t)
        // 可以感知到异常，同时返回默认值
        return 999;
    });
    Integer integer = future.get();  //阻塞式等待
    System.out.println("结果是："+integer);
    System.out.println("main......end......");
}

3）、handle 最终处理

和 complete 一样，可对结果做最后的处理（可处理异常），但同时还可改变返回默认值

示例：

传入 R apply(T t, U u)

接收结果和异常，当有异常时可指定返回默认值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 0;
        return i;
        
    }, service).handle((result, exception) -> {     //R apply(T t, U u)
        if (result != null) {
            return result * 2;
        }
        if (exception != null) {
            return 999;
        }
        return 0;
    });
    Integer integer = future.get();  //阻塞式等待
    System.out.println("结果是："+integer);
    System.out.println("main......end......");
}

4）、线程串行化

1. thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回
2. thenAccept 方法：消费处理结果。接收任务的处理结果，并消费处理，返回结果
3. thenRun 方法：只要上面的任务执行完成，就开始执行 thenRun，只是处理完任务后，执行thenRun 的后续操作
4. 带有 Async 默认是异步执行的。同之前
5. 以上都要前置任务成功完成

示例：

thenRunAsync 传入 void run()

不能获取上一步的执行结果，同时也无返回值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture.supplyAsync(()->{
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 4;
        System.out.println("运行结果："+i);
        return i;
    },service).thenRunAsync(()->{    //void run()
        System.out.println("任务2启动了。。。。");
    },service);
    System.out.println("main......end......");
}

thenAcceptAsync 传入 void accept(T t)

能获取上一步的执行结果，无返回值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture.supplyAsync(()->{
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 4;
        System.out.println("运行结果："+i);
        return i;
    },service).thenAcceptAsync((result)->{   //void accept(T t)
        System.out.println("任务2启动了。。"+result);
    },service);
    System.out.println("main......end......");
}

thenApplyAsync 传入 R apply(T t)

能获取上一步的执行结果，有自己的返回值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("当前线程" + Thread.currentThread().getId());
        int i = 10 / 4;
        System.out.println("运行结果：" + i);
        return i;
    }, service).thenApplyAsync((result) -> {      //R apply(T t)
        System.out.println("任务2启动了。。" + result);
        return result * 10;
    }, service);
    Integer integer = future.get();
    System.out.println("任务2结果是："+ integer );
    System.out.println("main......end......");
}

5）、两任务组合 - 都要完成

1. 两个任务必须都完成，触发该任务
2. thenCombine：组合两个 future，获取两个 future 的返回结果，并返回当前任务的返回值
3. thenAcceptBoth：组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值
4. runAfterBoth：组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后，处理该任务

示例：AB都完成后才会执行C

runAfterBothAsync 传入 void run()

都完成了计算才会执行，不获取结果，没有返回值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务 01 线程" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("任务 01 结果：" + i);
        return i;
    }, service);
    CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务 02 线程：" + Thread.currentThread().getId());
        System.out.println("任务 02 结束：");
        return "Hello";
    }, service);
    
    future01.runAfterBothAsync(future02,()->{     //void run()
        System.out.println("任务 03 线程：" + Thread.currentThread().getId());
    },service);
    System.out.println("main......end......");
}

thenAcceptBothAsync 传入 void accept(T t, U u)

都完成了计算才会执行，获取结果，没有返回值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务 01 线程" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("任务 01 结果：" + i);
        return i;
    }, service);
    CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务 02 线程：" + Thread.currentThread().getId());
        System.out.println("任务 02 结束：");
        return "Hello";
    }, service);
    
    future01.thenAcceptBothAsync(future02,(f1,f2)->{    //void accept(T t, U u)
        System.out.println("任务 03 线程：" + Thread.currentThread().getId());
        System.out.println("任务 03 结束："+f1+"----->"+f2);
    },service);
    System.out.println("main......end......");
}

thenCombineAsync 传入 R apply(T t, U u)

都完成了计算才会执行，获取结果，返回当前任务的返回值

public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("main......start....");
    CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务 01 线程" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("任务 01 结果：" + i);
        return i;
    }, service);
    CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任务 02 线程：" + Thread.currentThread().getId());
        System.out.println("任务 02 结束：");
        return "Hello";
    }, service);
    
    CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> {    //R apply(T t, U u)
        System.out.println("任务 03 线程：" + Thread.currentThread().getId());
        System.out.println("任务 03 结束：" + f1 + "----->" + f2);
        return f1 + f2;
    }, service);
    String s = future.get();
    System.out.println("任务 03结果是："+ s );
    System.out.println("main......end......");
}