知识库是企业经营过程中的面向客户和内部员工的知识沉淀文档库，里面包含各类教程、问答、案例等，知识库的检索匹配是自然语言处理(NLP)中一个重要的基础问题，本质是进行文本语义的相似度计算，也就是语义匹配，我们很多领域的任务都可以抽象为文本匹配检索任务，例如检索引擎、智能客服、知识检索、信息推荐等领域。
　　知识库检索匹配可以概述为：给定一个query和大量候选知识库的文档，从这些文档中找出与用户输入query最匹配的TopK个文档。

二、架构流程

2.1、整体架构

2.2、请求链路

三、算法模型

3.1、DSL改写

　　检索优化第一步：DSL改写，接手前业务方自己已经对检索结果做过优化，调整不同字段的匹配权重，这一方法的已经难以继续优化。从知识运营的角度出发，在用户检索时，将运营认为重要的文档推到前面，由于文档之间互相有链接引用，可以使用PageRank算法给每个文档计算重要分(PR值)。
　　PageRank的核心思想是，被引用次数越多的文档越重要。算法原理如下，假设只有四个网页ABCD，以AB间的箭头为例，代表可以从B网页跳转到A网页，对B即一次引用（链出），对A则一次被引用（链入）。L(B) 表示B网页的链出数量，PR(B)表示B网页的PageRank分数。 　　假设所有文档的初始PR值是0.25，这里L(B)=2，L(C)=1，L(D)=3，计算出PR(A)=0.458，接下来计算所有其他被引用（有链入）的文档PR值，PageRank是个迭代算法，反复计算以后所有的PR值会收敛，那就是最终每个文档的PR值，也是用来改写DSL的关键信息：

new score = old score * log(1+2*PageRank)  

　　old score指原来不同字段加不同权重由ES算出来的BM25分数，PageRank缺失值使用1代替。

3.２、文本召回

　　文本召回是召回中最常用的一种策略，最常见的方式是通过对Query直接进行分词，然后将分词后的关键词到ES构建倒排索引，进行tf-idf等相似计算匹配索引召回，这种召回方式的优点是实现简单，不需要训练模型、低资源需求、检索速度快，然而它的缺点也很明显，文本是具有语义的、是有语法结构的，文本召回忽略了语句的语法结构，同时也无法解决一词多义和同义词的问题，对 query 进行语义层面相似的召回效果就比较一般，解决这个问题就要用到向量召回。

3.３、向量召回

　　向量召回的思想就是计算搜索词的向量和文档标题/相似问的向量的余弦相似度，返回相似度分数最高的TopK个文档，计算向量相似度的步骤放在Milvus进行，Milvus作为向量检索库，对计算过程有优化，性能更好。
　　向量计算模型结构： 　　文本转向量的算法模型由embedding、两层transformer和MLP组成，模型最后会对编码向量做L2归一化，采用典型的双塔模式，可以将左塔的搜索词和右塔的文档标题形成独立的子网络，左右塔的结构分离但编码器参数共享，双塔结构天然的可以用于召回，将这个模型部署到小盒子就可以在线计算搜索词的向量，将海量的知识库文档作为右塔离线训练成文本向量后刷入向量检索工具Milvus。
　　双塔召回模型的核心思想是将query/item嵌入到共享低维空间，然后通过向量距离来度量相关性。
　　向量召回流程：

　　工程实现部分的DP离线任务中实现了计算文档标题和相似问向量并将其导入Milvus的功能。由于Milvus对string类型属性信息存储检索不够友好，所以在DB阶段会请求mysql库表对召回结果进行扩展，匹配补全相关信息。

3.4、精排序

　　经过召回和粗排后，可以理解为将重要相关的文档排在了前面，但是距离用户真正的检索意图还有差距，可以使用用户的检索记录对结果再进行排序。
　　基于所有场景的用户检索点击数据，有点击行为就认为检索词和文档标题匹配（正样本），其他就认为没有那么匹配（负样本）
　　训练数据样例：
　　采用 in_batch 负采样就不需要提前构造负样本，模型的设计如下： 　　检索词与正负样本的相似度会进入InfoNCE（info Noise Contrastive Estimation，噪声对比估计）的函数计算损失，使用这个损失来更新模型参数。这个由很多负样本组成的双塔结构也称为对比学习，核心思想是去拉近相似的样本，推开不相似的样本，目标是要从样本中学习到一个好的语义表示空间。
　　精排模型与文本转向量的算法原理相同。
　　InfoNCE计算公式：（可以理解为带温度超参的CrossEntropy）

L_i=-log(\frac{exp(sim(z_i, z_i^+)/\tau)}{\sum_j{exp(sim(z_i, z_j)/\tau)}})  

　　分子是正例对的相似度，分母是正例对+所有负例对的相似度，最小化infoNCE loss，就是最大化正例对的相似度，最小化负例对的相似度。
　　在计算损失时，label可以在batch内生成，检索词和文档的编码向量经过矩阵乘法可以得到一个相似度方阵，对角位置就是互相匹配的检索词和文档的分数，如果batchsize=4，那每行对应的label就是 [0,1,2,3]。inbatch负采样损失计算示意图：

　　模型训练好以后，就得到文本的编码器，输入两个文本，就可以得到一个匹配的分数，将这个模型部署到小盒子，在需要排序时，输入候选的文档标题和检索词，按计算出来的分数从高到低排序，就完成了一次对检索结果的排序。

3.5、排序优化

　　上述向量召回介绍的在模型服务中计算两个文本相似度的方法，在只需要对20个文档（一页）排序时是没有问题的，但是每个文档还会有若干个相似问，只使用20个商品标题没法很好的代表整个文档，如果能使用每个文档的标题和全部相似问，那效果会更稳定一些，但这样的话，每次进入小盒子排序的文档数就不固定了，少则20，多则几百，上百的batch_size走小盒子很容易超时，就算切成小的batch，也会有一个失败全部失败的情况，而且总体上rt也没下降多少。
　　与向量召回时类似，模型也改为输入只有一个文本，输出这个文本的向量。每个文档的标题和全部相似问向量都与Query向量算相似度后计算均值，等价于先计算文档的标题和全部相似问的向量均值，再与Query向量计算相似度。基于此，排序任务也可以转换为向量召回任务。
　　在进入Milvus之前，会按照向量召回和ES召回的文档ID作为过滤条件，一次计算的RT就得以保证，可以支持的QPS也可以比基于小盒子的版本高：
　　此方法可以解决RT和QPS问题，但是也有局限，只能优化“每个文档的标题和全部相似问向量都与Query向量算相似度后计算均值”这个均值计算逻辑，其他的比如“取最大的相似度”就不能这么做了，而且Query与文档的交互太少，只在最后算相似度，可能不如多次交互的模型的效果好。

四、工程实现

　　当线上接受一条检索请求文本后，先调用在线推理-小盒子计算Query向量，然后去Milvus向量库中和知识库向量进行相似度计算，并返回距离最近的Top N个Item作为向量召回的结果。

4.1、离线训练（DP平台）

　　海量的知识语料库向量化计算在自研DP平台离线运行，使得全库文本匹配速度较快：
　　1）语料库预处理：包括语料库的文本清洗、文本筛选等预处理逻辑
　　2）语料库向量化：利用上述的向量计算模型进行向量化
　　3）导入Milvus库：将集合部署在Milvus集群，依次批量导入更新机器的集合保证线上可用

4.2、在线推理（Sunfish平台）

　　自研算法平台（Sunfish）对模型训练提供一站式闭环服务，支持分布式训练、GPU/CPU切换、模型版本管理、一键式运行和部署等功能，其中：
　　1）算法工程模块：一键运行训练、任务作业管理、模型输出
　　2）模型管理模块：实现训练任务、地址导入、本地上传等多途径模型来源选择
　　3）模型部署模块：简单配置、模型格式选择、线上资源配置等便捷署方式

{
    "inputs":[
        {
            "name":"INPUT",
            "shape":[
                1,
                1
            ],
            "datatype":"BYTES",
            "data":[
                "满足条件满减送没有赠品"
            ]
        }
    ],
    "outputs":[
        {
            "name":"OUTPUT"
        }
    ]
}

4.3、Milvus向量检索

　　Milvus 是一款开源的、针对海量特征向量的向量相似性检索(ANNS，Approximately nearest neighbor search)引擎，集成了 Faiss、Annoy 等广泛应用的向量索引，成本更低、性能更好、高度灵活、稳定可靠以及高速查询等特点，十亿向量检索仅毫秒响应。
　　1、Milvus向量索引列表如下：
　　简言之，每种索引都有自己的适用场景，如何选择合适的索引可以简单遵循如下原则：
　　1）当查询数据规模小，且需要 100％查询召回率时，用 FLAT；
　　2）当需要高性能查询，且要求召回率尽可能高时，用 IVF_FLAT；
　　3）当需要高性能查询，且磁盘、内存、显存资源有限时，用 IVFSQ8H；
　　4）当需要高性能查询，且磁盘、内存资源有限，且只有 CPU 资源时，用 IVFSQ8。

　　2、Milvus 目前支持的距离计算方式与数据格式、索引类型之间的兼容关系： 　　选择合适的距离计算方式比较向量间的距离，能很大程度地提高数据分类和聚类性能，主要采用内积 (IP)的计算方式，内积更适合计算向量的方向。
　　内积计算两条向量之间的夹角余弦，并返回相应的点积。内积距离的计算公式为：
　　假设有 A 和 B 两条向量，则 ||A|| 与 ||B|| 分别代表 A 和 B 归一化后的值。cosθ 代表 A 与 B 之间的余弦夹角。
　　在向量归一化之后，内积与余弦相似度等价。因此 Milvus 并没有单独提供余弦相似度作为向量距离计算方式。

4.4、AI模型接口服务

　　算法模型接口服务由ai-service和ai-app两个服务组成，ai-service负责调用算法模型在线推理、Milvus实时向量召回等接入库，ai-app负责业务逻辑的开发。
　　1、ai-service配置示例：

{
    "model_name": "similarity_jira",
    "model_source_type": "YZ_MODEL",
    "model_version": 1,
    "model_invoke_timeout": 3000,
    "protocol": "kfserving",
    "infer_type": "triton",
    "feature_maps": [
        {
            "model_feature_key": "INPUT",
            "data_type": "string",
            "shape": "(-1,1)",
            "default_value": "",
            "feature_source": "PARAMS",
            "source_key": "jira_text",
            "is_required": 1
        }
    ],
    "param_mapping": {
        "jira_text": "<objectList.jira_text>"
    }
}

　　2、ai-app接口设计
　　实现业务逻辑开发测试后，发布上线即可提供前后端调用。
　　a、Maven示例:

<dependency>  
  <groupId>com.youzan</groupId>
  <artifactId>ai-app-api</artifactId>
  <version>1.0.13-RELEASE</version>
</dependency>  

　　b、请求示例：

invoke com.youzan.ai.app.api.service.jira.Service.retrieve({"fromApp":"test","scene":"similarity_predict","Title":"满足条件没有赠品","Key":"XXX"})  

　　c、返回示例：

{
    "code":200,
    "data":{
        "Similaritys":[
            {
                "createdAt":1648137600000,
                "score":0.9390,
                "key":"XXX0123442334",
                "title":"满足条件没有赠品"
            },
            {
                "createdAt":1636214400000,
                "score":0.9010,
                "key":"XXX0123365819",
                "title":"满足条件没有送赠品"
            },
            {
                "createdAt":1653408000000,
                "score":0.8735,
                "key":"XXX0123482446",
                "title":"订单满足条件没有送赠品"
            },
            {
                "createdAt":1655308800000,
                "score":0.8312,
                "key":"XXX0123496337",
                "title":"订单满足条件但是没有送赠品"
            },
            {
                "createdAt":1659628800000,
                "score":0.8028,
                "key":"XXX0123527965",
                "title":"订单满条件但是赠品没有送"
            }
        ]
    },
    "success":true,
    "message":"successful"
}

五、服务场景

5.1、官网帮助中心

5.2、相似商品推荐

六、总结

　　本文大致介绍了知识库检索匹配的算法和工程服务化实践过程，服务应用的业务场景比较广泛，并对类似场景的接入做了低代码封装和策略配置，便于快速轻量级的服务化落地。在知识库检索匹配服务化实践过程中，后续值得关注以下几点：
　　1）对于知识库中低频或缺失的新文档或新商品的Embedding学习还不够充分，可考虑利用图结算法结构，把更多query和其他属性的语义信息聚合，进而提升query和知识库文档的语义表征能力；
　　2）Embedding 的实现算法比较多，可考虑结合业务需要，将词嵌入模型模块内置化；
　　3）服务的性能和稳定性，确保服务在高准确率和高QPS下的响应性能依然是重中之重。

七、招聘号外

　　有赞数据中台团队，目前有数据平台研发以及数据开发岗位虚位以待，有需要的同学请关注官网。

一般做算法任务时，都需要搜集大量标注的数据，假如我们要预测一个商品的产品词（中心词），下面是一个商品标题：

三亚亚龙湾玫瑰谷JESS玫瑰臻白颜透润花瓣 免洗面膜收缩毛孔

这个商品的产品词就是“面膜”，任务就是要把面膜识别出来，看起来是个标准的NER任务，我们也确实使用了CRF和指针网络之类的方法，对于上面这种标题效果还不错，但是由于SaaS商家的经营习惯不同于平台，很少依赖平台搜索流量，所以很多标题很简短甚至不会包含产品词，比如：

迪奥丝绒系列760 专属女团色 蓝调正红 可盐可甜
澳优能立多3段

对于这种问题，NER相关的算法就无解了，模型无法在商品标题中找到合适的产品词，当然也可以认为是标题中没有产品词。但是这种模型很多时候会从标题里预测出来奇奇怪怪的结果，导致产品词太发散，业务方很难基于产品词制定规则。

去年我们接触到了对比学习，被OpenAI的CLIP: Connecting Text and Images惊艳到了，效果之好，方法之简单，令人兴奋。而且微软相似的模型Turing Bletchley: A Universal Image Language Representation model by Microsoft - Microsoft Research甚至还展现了OCR的能力，再加上微博做的文本对匹配的效果，为我们打开了新的思路。如果利用对比学习学到商品标题和产品词的表示，那只需要清理一批产品词计算向量存到向量计算引擎，需要预测的商品标题计算完表示以后做一次向量召回就会得到语义相关的产品词，就算商品标题里不含产品词也可以找出一个合理的产品词，而且产品词词库可控，将会是一个理想的解决方案。

2. 对比学习的原理

对比学习的思想很简单，就是学习对象的表示（向量），相关对象的表示要接近，不相关对象的表示要远离。对比学习也算是自监督学习，跟其他方法的区别可以看下图，对比学习也是有label的，但是跟监督学习不同的是，这个label不是最终任务的label。

谷歌的FaceNet以及微软的DSSM等模型，也是这种思想。我们以人脸识别为例，如果使用监督学习的方式，任务就是基于人脸图片预测人的唯一ID。那就需要收集每个人很多张人脸照片（不同角度不同年龄不同光照等），这几乎无法实现。所以可行的方法就是让模型学会表示人脸，同一个人不同照片得到的表示接近，不同的人的照片表示不接近。FaceNet正是基于这种思想，使用Triplet Loss，拉近Anchor(用户A的照片1)与Positive(用户A的照片2)的距离，推远Anchor(用户A的照片1)与Negative(用户B的照片1)的距离，从而学会生成有区分度的人脸表示。这里每个样本中的负样本数量为1，即：(Q, P, N)，这里用Q表示Anchor，P表示正样本，N表示负样本。DSSM中负样本的数量为4，即：(Q, P, N_1,N_2,N_3,N_4)。

从SimCLR到CLIP，再到最近大火的DALL·E 2，对比学习的潜力被大量开发，文本与文本的对比，图像与图像的对比，文本与图像的对比，效果都非常好，而且最重要的是，工业界落地容易，只要提前计算好向量表示就可以实时推理。

2.1 损失函数

我们来看看对比学习是怎么训练的，负样本的数量对模型效果的影响还是很大的，SimCLR文章给出了batch_size（可以理解为负样本数量）与模型效果的关系：

上面我们提过FaceNet一个负样本，DSSM4个负样本，而CLIP的负样本数量达到了32767， 这么多的负样本就是用来计算损失的，损失函数为infoNCE，形式如下：

sim(x, y)代表两个表示的相似度，可以是余弦相似度（一般模型最后一层会是L2归一化，所以余弦相似度 sim(x, y) 的结果在-1到1之间）。τ为温度超参，如果τ为1，sim(x, y)看做logits，那就是个标准的交叉熵。一般交叉熵我们用于分类任务，这么说的话对比学习也算一种分类任务。

以下图为例，这组样本可以表示为：(Q, P, N_1, N_2)，即有两个负样本，模型会计算出sim(Q, P)，sim(Q, N_1)， sim(Q, N_2)，为了最大化sim(Q, P)这一“类别”的概率，我们可以认为该样本的one-hot label是[1, 0, 0]。如果我们按照(Q, P, N_1, N_2)的方式准备所有训练数据，那每个样本的label就都是[1, 0, 0]，而且损失函数是交叉熵，形式上已经是分类任务了，不同的是，这个“类别”没有实际意义。从另外一个角度想，分类任务可以看到所有的负样本（所有的类别），而对比学习只能看到有限的负样本，所以负样本越多学习难度会越大，训练出来的模型更稳健。

2.2 Batch内负采样

如果像上图一样，每条训练数据中都提前准备好若干个负样本（可以全局负采样），假设两个负样本，即（Q, P, N_1, N_2）,当按batch训练时，需要分别计算batch_Q, batch_P, batch_N1, batch_N2的向量表示，然后再计算对比损失，计算量非常大，而且负样本数量越多训练越慢。所以一种经典的方法就是batch内负采样，训练数据只需要是（Q, P），实际计算时，一个batch内的其他Q对应的P就可以作为当前Q的N，前向计算只需要计算batch_Q和batch_P，只需要提高batch_size就可以增加负样本数量。

用CLIP的图来解释batch内负采样的计算过程，如下图所示，batch文本经过Text Encoder得到batch_size * dim的矩阵，记做ET，也就是图里T_1, T_2, ...,T_N的向量表示。batch图像经过Image Encoder得到batch_size * dim的矩阵，记做EI，进行矩阵乘法 ET * transpose(EI)后就得到batch_size * batch_size的相似度方阵，对角位置就是相关的文本和图像的相似度，即 sim(Q, P)，其他位置就是不相关的文本和图像的相似度 sim(Q, N)，所以这个batch的稀疏label就是（0，1，2，...，batch_size-1）。有了logits和label就可以使用infoNCE计算损失函数。

3. 模型设计

在我们决定尝试用对比学习（语义匹配）的方式做商品的产品词预测时，就想到了用搜索点击数据训练模型。搜索日志数据量非常大且容易获取，搜索点击是一种弱监督的数据，包含用户对query和商品相关性的认可，虽然包含很多噪声，但是数据量大的情况下模型有能力学会忽略这些噪音，反而有可能增加模型的稳健性。所以我们的训练数据（Q，P）就是（query，goods_info），goods_info可以只是简单的使用商品标题，也可以加入其他信息如店铺名，商品类型，商品描述等等。

模型设计为经典的双塔结构，由于商品搜索点击的query和goods_info分布上很接近，所以双塔的结构和参数完全共享，也就是一个塔用两次。

当goods_info 包含商品标题外的其他信息时，使用segment_id区分。这里我们输入的序列长度是100，embedding维度为512，6层Transformer encoder layer，总参数量17M，只使用6层transformer是基于线上服务性能考虑，独占单卡，100并发，可以达到60ms的RT和1500的QPS，且现有两层transformer的模型表现也不错，6层目前来说是比较经济的选择。

4. 训练技巧

4.1 温度超参

infoNCE相对于crossEntropy多出来的温度超参τ，按照Understanding the Behaviour of Contrastive Loss中的解释，具有控制表示分布的能力，小的温度超参学习出来的分布更加均匀，大的温度超参学习出来的分布类间更加远离，如下图：

CLIP中将温度超参设置为可学习的参数，我们也曾尝试过，但是模型学习到最后，为了loss继续降低，会“偷懒”将温度参数设置的很小，之前训练得到过0.0037这么小的温度参数，当模型有分辨能力后，过小的温度超参会阻碍模型表示能力的进一步提高，如下图所示，所以最终训练时，我们还是使用了常数0.1作为温度超参的值。

4.2 分布式训练

由于对比学习对负样本的依赖，理论上负样本越多模型的表示能力越稳健，使用batch内负采样的技术需要增加batch_size以增加负样本数量，而batch_size的大小受限于显卡的显存。那既然一张卡的显存有限，使用多个机器的多张卡一起训练呢？不仅可以增加batch_size，还能提高训练速度，于是我们设计了多机多卡的训练方案，如下图所示：

这里以三个节点为例，每个节点为一台单卡的机器，当模型前向计算出goods_embedding和query_embedding时，对于单机就可以直接计算相似度矩阵了，但是为了获得更大的相似度矩阵，我们使用all_gather操作，将其他节点的goods_embedding和query_embedding收集过来，拼成更大的embedding矩阵再计算相似度，原来单机的相似度矩阵维度为（batch_size， batch_size），这里分布式的相似度矩阵会增加为（batch_size * num_node，batch_size * num_node），如果单机最大batch_size是500，三台分布式训练batch_size（all_gather后）就可以达到1500。只要增加节点就可以增加负样本数量，分布式训练看起来挺适合对比学习。

实际训练时我们使用5台机器做分布式训练，将训练数据切分为5份，每台机器读取一份，训练框架使用Pytorch Lightning。

4.3 一些节省显存的技巧

分布式训练靠多台机器增加batch_size，单节点的batch_size可以靠优化模型训练时的显存占用提高。根据ZeRO里的说明，模型在GPU上训练时，显存中会存在模型的参数、各层的激活值、优化器的状态、梯度等，要减少显存占用就要想办法优化这些数据的显存开销。

4.3.1 ZeRO

ZeRO就是基于分布式提出的节省单节点显存的方案，一般分布式每个节点都存储着完整的模型数据（模型参数、优化器的状态、模型的梯度、前向计算的激活值等），如果将优化器的状态参数拆分，保存在每个节点上，每个节点只维护部分参数，那单个节点的显存占用会显著下降，同样的，梯度，参数都分布式保存，显存占用会极大的得到优化。

我们使用Pytorch Lightning中的ddp_sharded，即将优化器的状态参数和模型的梯度分布式保存在每个节点上，详细说明参见官方文档。

4.3.2 Activation Checkpointing

Activation Checkpointing也叫Gradient Checkpointing或者重计算技术。模型训练前向计算时，每层的计算结果（激活值）都会保存，方便反向传播时更新梯度，当batch_size增加时，其占用的显存将非常大，而Activation Checkpointing的思想就是部分层的计算结果不保存，当反向传播到这层时，重新前向计算这一层的激活值，再进行梯度更新，会增加一些训练时间，但是可以节省很多显存，如图，反向传播要等重计算完成：

根据网上的测试，使用Activation Checkpointing可以减少约60%的显存占用，同时增加约25%的训练时间，这对显存紧张的我们来说无疑是很划得来的。而且很重要的一点是，对模型训练的精度没有影响：

陈天奇大佬2016年提出了这个方法，那个时候BERT还没出呢，详细见大佬论文Training Deep Nets with Sublinear Memory Cost 。

我们的模型开发由Tensorflow迁移到Pytorch，也有Pytorch官方自带这个功能torch.utils.checkpoint.checkpoint的原因，不过需要尽量避免重算dropout层。具体实现时，我们继承TransformerEncoderLayer类，将MultiheadAttention层改为允许重计算的层。

4.3.3 Mixed Precision

混合精度训练（Mixed Precision Training）是指模型训练时，模型权重和激活值以FP16存在，在有Tensor Core的GPU上可以加速，FP16也可以节省显存。虽然在前向和反向传播时使用FP16，但在更新权重的时候还会使用FP32。

实际测试中，大约可以节省30%的显存，节约50%的训练时间。

4.3.4 训练配置

使用上面提到的这些优化显存占用的方法，单卡显存占用可以节约55%，16G显存的GPU可以放下700的batch_size，5个节点，最终计算对比损失时的batch_size可以达到3500。（但是要注意batch_size也不是越大越好，当你的训练数据多样性不足而batch_size较大时，会引入较多False Negative，即其他Q的P与当前Q也是语义上相近的，但是计算loss时却是当作负样本处理的，有点类似于做分类时有类别重复）

我们总共收集184M的搜索点击加购日志，5台GPU机器分布式训练，计算loss的相似度矩阵大小是3500*3500，温度超参为0.1。

5. 效果

我们提前整理了一个产品词词库，用来做向量匹配，来看一下效果：

只要商品标题里出现了产品词，预测起来是比较容易的，标题中出现多个产品词时，模型也有能力找出最合适的那个（不过也有badcase，比如“足浴盐”），当标题中没有产品词时，也可以通过向量匹配的方式找一个合理的产品词，这是之前的NER方案完全不能处理的。而且可以通过卡阈值的方式找出词库中语义相近的产品词。

用来匹配文案效果也不错：

而且测试还发现，属性信息也被编码进了表示中，可以区分品牌、颜色、男女、季节等。

再来一个更有挑战的任务：匹配类目。由于类目文本的形式是：一级类目名>二级类目名>三级类目名>... 这种形式与搜索词和商品标题偏差较大，正好可以测试一下模型的泛化能力：

可以看出就算类目文本和训练数据分布不一致，模型也有能力做出预测，叶子类目预测的效果也超出预期，这是Zero-shot的效果，非常令人兴奋。想想之前如果要做文本分类，我们需要标注大量数据，加上类目体系庞大，要取得好的效果需要花费大量资源，有了对比学习，就不用从0分学起了，上来就是60分，只需要少量的标注数据，就可以达到令人满意的效果。人工智能终于可以不用那么人工了，大数据的优势也体现出来了。

商品文本编码可以使用商品搜索日志训练，类似的，图像的编码器可以通过商品标题与商品图片的相关性学习出来。至此，我们就拥有了商品文本和图像各自的编码器，可以做文本和文本的匹配，文本和图像的匹配，图像和图像的匹配。

图文匹配的CLIP模型已经开源，可以在这里找到https://huggingface.co/youzanai/clip-product-title-chinese

6. 应用

基于对比学习预训练加微调和向量召回的方案，目前已经在有赞的商品产品词预测、商品类目预测、相似商品推荐、搜索召回、搜索排序、智能文案、商品风控等场景上线使用，稳健性都要好于之前纯有监督的方案。

可以说，有了对比学习这个强力工具，以前做不了的都可以做了，思路都打开了。对比学习，真的大有可为。

作为后端开发，redis是工作中最绕不开的中间件之一，在工作中通常有以下几个常用用途

1.缓存，可以抗十万级别的qps
2.计数器，如点赞数，pv等
3.分布式锁
4.限流

另外丰富的redis数据类型支持了一些扩展功能，如排行榜，消息队列，布隆过滤器，位图等等。而redis的底层实现是十分简单的，核心源码也仅有几万行。本文就带大家来领略，小小的redis是如何实现这些复杂功能的~

*注:本文介绍的源码为redis 5.0.14版本 *

二、字符串

C语言存储字符串的问题

1.二进制安全

C语言中表示字符串结尾的符号是'\0',如果字符串本身就具有'\0'字符，就会被截断，即非二进制安全。

2.计算字符串的长度性能低

C语言中有一个计算字符串长度的函数strlen，但这个函数与Java的不一样，需要遍历整个字符串来计算长度，时间复杂度是O（n），如果需要在循环中计算，性能将十分低下

3.字符串拼接性能低

因为C语言字符串不记录长度，对于一个长度n的字符串来说，底层是n+1的字符数组

char a[n+1]  

如果需要增长字符串，则需要对底层的字符数组进行重分配的操作

接下来由数据结构入手，看看redis是如何解决这几个问题的

2.1 数据结构

struct sds{  
    int len; //buf中已占字符数
    int free; //buf中空闲字符数
    char buf[];
}

除了保存字符串的指针buf，还需要记录使用空间和空闲的空间。redis老版本也是这样设计的，这样的设计解决了开头的三个问题:

1.计算字符串长度的时候，时间复杂度是O(1)

2.使用len变量得出字符串的长度，而不是’\0‘，保证了二进制安全。

3.对于字符串的拼接操作，进行预分配空间，减少内存重分配的次数

小字符串空间浪费的问题

在64位系统中，字符串头部的len和free各占四个字节，对于大字符串而言，这个数字还好，但是如果是小字符串呢，比如buf本身只有一个字节，而头部就占了八个字节，肯定不合适。

redis新版本就给了一种方案，根据buf字符串的长度不同，使用不同的结构体存储，同时新增一个单字节变量flags，保存不同的类型。

但是对于那种只有一个字节长的字符串，如何优化呢？ 对于那种小字符串，redis中使用一个字节的标志位flags表示 低三位存储类型（type），高五位存储长度（len)，而高五位 2^5-1=31 可以存储最多31个字节的字符串。

而大于31个字节的其他几种类型字符串，一个字节存不下，就使用两个变量保存已使用空间和总长度（保留flags字段标识类型，新增len字段标记长度）。

sdshdr8,sdshdr16,sdshdr32,sdshdr64 结构都是一样的，区别在于存储的变量大小。

struct __attribute__ ((__packed__)) sdshdr5 {  
    unsigned char flags; /* 3 lsb of type, and 5 msb of string length */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr8 {  
    uint8_t len; //已使用
    uint8_t alloc; // 总长度
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr16 {  
    uint16_t len; /* used */
    uint16_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr32 {  
    uint32_t len; /* used */
    uint32_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr64 {  
    uint64_t len; /* used */
    uint64_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};

2.2 基本操作

只介绍扩容操作，其它操作都比较简单，可自行阅读 扩容源码如下

sds sdsMakeRoomFor(sds s, size_t addlen) {  
    struct sdshdr *sh, *newsh; //定义两个 sdshdr 结构体指针
    size_t free = sdsavail(s); // 获取 s 目前空闲空间长度
    size_t len, newlen; // 前者存储扩展前 sds 字符串长度，后者存储扩展后 sds 字符串长度

    if (free >= addlen) return s; // 如果空余空间足够，直接返回
    len = sdslen(s); // 获取 s 目前已占用空间的长度
    sh = (void*) (s-(sizeof(struct sdshdr))); //结构体指针赋值
    newlen = (len+addlen); // 字符串数组 s 最少需要的长度
    // 根据新长度，为 s 分配新空间所需的大小
    if (newlen < SDS_MAX_PREALLOC) // 如果新长度小于 SDS_MAX_PREALLOC（默认1M），那么为它分配两倍于所需长度的空间
        newlen *= 2; 
    else
        newlen += SDS_MAX_PREALLOC; // 否则，分配长度等于目前长度加上 SDS_MAX_PREALLOC（默认1M）
    newsh = zrealloc(sh, sizeof(struct sdshdr)+newlen+1);
    if (newsh == NULL) return NULL;

    newsh->free = newlen - len;
    return newsh->buf;
}

三、跳跃表

跳跃表类似一个多层的链表，首先从最高层开始查找，如果下一个节点的值大于要查找的值或者下一个节点为null,则往下一层查找。通过空间换时间的策略，将时间复杂度控制在O(logn)。

3.1 一个例子

例如查找51这个数

首先从第一层开始查找，找到第二个节点，发现后面为null。

从第二层查找 查找到第四个节点，发现后面的节点为61，大于当前的数。

从第三层查找 查找到第六个节点 结束 一共查找四次，比遍历一次少了两次。数据量大的情况下，这个性能会提升的很明显。

3.2 跳跃表结构

首先看一下zskiplistNode的数据结构，zskiplistNode表示跳跃表中的一个节点。

typedef struct zskiplistNode {  
    sds ele;// 数据
    double score; //权重比
    struct zskiplistNode *backward; //后退指针，指向当前节点底层 前一个节点
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 指向当前层的前一个节点
        unsigned long span; //forward 指向前一个节点的与当前节点的间距
    } level[];
} zskiplistNode;

zskiplist 表示跳跃表

typedef struct zskiplist {  
    struct zskiplistNode *header, *tail; //分别指向头结点和尾结点
    unsigned long length; //跳跃表总长度
    int level; //跳跃表总高度
} zskiplist;

其中，头节点是跳跃表的一个特殊节点，它的level数组元素个数为64。头节点在有序集合中不存储任何member和score值，ele值为NULL, score值为0；也不计入跳跃表的总长度。头节点在初始化时，64个元素的forward都指向NULL, span值都为0。

3.3基本操作

3.3.1 创建跳跃表

zskiplist *zslCreate(void) {  
    int j;
    zskiplist *zsl;

    zsl = zmalloc(sizeof(*zsl));
    zsl->level = 1;
    zsl->length = 0;
    // 头结点
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

简单来说就是创建了头结点，创建了64个level数组。

3.3.2 随机层高

创建和插入节点的之前，当前节点需要在哪几层出现，是通过计算当前节点的level值， 而level值是redis通过伪随机得出的，层数越高，节点出现的概率越小。

int zslRandomLevel(void) {  
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

层高的数学期望

高度为1的概率 (1-p)

高度为2的概率 p(1-p)

高度为3的概率 (p2)*(1-p)

….

高度为n的概率(p(n-1))*(1-p)

期望层高 E=1*(1-p)+2p(1-p)+3p2(1-p)... =1/(1-p)

当 p=0.25（redis默认值） 时，跳跃表节点的期望层高为 1/(1-0.25)≈1.33。 即多浪费了30%的空间，redis的跳表使用了较低的空间成本，实现了时间复杂度的大减少

3.3.3 插入节点

插入节点总的来说一共四步

1.查找插入位置
2.调整高度
3.插入节点
4.调整backward

源码如下：

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {  
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));
    // 查找节点
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                    sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    /* we assume the element is not already inside, since we allow duplicated
     * scores, reinserting the same element should never happen since the
     * caller of zslInsert() should test in the hash table if the element is
     * already inside or not. */
    //调整高度
    level = zslRandomLevel();
    if (level > zsl->level) {
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        zsl->level = level;
    }
    x = zslCreateNode(level,score,ele);
    //插入节点
    for (i = 0; i < level; i++) {
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;

        /* update span covered by update[i] as x is inserted here */
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    /* increment span for untouched levels */
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }

    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    zsl->length++;
    return x;
}

3.3.4.删除节点

1.查找节点 (同插入节点)
2.删除节点
3.修改高度

void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {  
    int i;
    for (i = 0; i < zsl->level; i++) {
        if (update[i]->level[i].forward == x) { // update[i].level[i] 的 forward 节点是 x 的情况，需要更新 span 和 forward
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {// update[i].level[i] 的 forward 节点不是 x 的情况，只需要更新 span
            update[i]->level[i].span -= 1;
        }
    }
    if (x->level[0].forward) { // 如果 x 不是尾节点，更新 backward 节点
        x->level[0].forward->backward = x->backward;
    } else { // 否则 更新尾节点
        zsl->tail = x->backward;
    }
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--; //更新跳跃表 level
    zsl->length--; // 更新跳跃表长度
}

3.4.5 跳跃表的应用

zset集合插入第一个元素时，会判断下面两种条件：

zset-max-ziplist-entries 的值是否等于 0；
zset-max-ziplist-value 小于要插入元素的字符串长度。

满足任一条件 Redis 就会采用跳跃表作为底层实现，否则采用压缩列表作为底层实现方式。

3.4.6 题外话

Q:为什么redis使用跳跃表而不是红黑树呢

引用一下原作者的话

There are a few reasons: They are not very memory intensive. It's up to you basically. Changing parameters about the probability of a node to have a given number of levels will make then less memory intensive than btrees.
A sorted set is often target of many ZRANGE or ZREVRANGE operations, that is, traversing the skip list as a linked list. With this operation the cache locality of skip lists is at least as good as with other kind of balanced trees.
They are simpler to implement, debug, and so forth. For instance thanks to the skip list simplicity I received a patch (already in Redis master) with augmented skip lists implementing ZRANK in O(log(N)). It required little changes to the code.About the Append Only durability & speed, I don't think it is a good idea to optimize Redis at cost of more code and more complexity for a use case that IMHO should be rare for the Redis target (fsync() at every command). Almost no one is using this feature even with ACID SQL databases, as the performance hint is big anyway.About threads: our experience shows that Redis is mostly I/O bound. I'm using threads to serve things from Virtual Memory. The long term solution to exploit all the cores, assuming your link is so fast that you can saturate a single core, is running multiple instances of Redis (no locks, almost fully scalable linearly with number of cores), and using the "Redis Cluster" solution that I plan to develop in the future.

简单翻译一下

1.这并不会浪费太多的空间，并且树的高度可以动态调整的。

2.ZRANGE 和 ZREVRANGE命令，跳表性能比红黑树好

3.红黑树比较复杂...作者懒得实现

四、整数集合

整数集合（intset）是一个有序的、存储整型数据的结构。 conding决定了的element的长度，对应关系如下

4.1 基本数据结构

typedef struct intset {  
    //编码
    uint32_t encoding;
    //元素个数
    uint32_t length;
    // 柔性数组，根据encoding 决定几个字节表示一个数组
    int8_t contents[];
} intset;

4.2 基本操作

4.2.1 查询元素

uint8_t intsetFind(intset *is, int64_t value) {  
　　uint8_t valenc = _intsetValueEncoding(value); //判断编码方式
　　//编码方式如果大于当前intset的编码方式，直接返回0。否则调用intsetSearch函数进行查找
　　return valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,NULL);


static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {  
int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;  
int64_t cur = -1;  
/*如果intset中没有元素，直接返回0 */
if (intrev32ifbe(is->length) == 0) {  
　　if (pos) *pos = 0;
　　return 0;
} else {
/* 如果元素大于最大值或者小于最小值，直接返回0 */
if (value > _intsetGet(is,max)) {  
　　if (pos) *pos = intrev32ifbe(is->length);
　　　　return 0;
} else if (value < _intsetGet(is,0)) {
　　if (pos) *pos = 0;
　　return 0;
    }
}

while(max >= min) { //二分查找该元素  
　　mid = ((unsigned int)min + (unsigned int)max) >> 1;
　　cur = _intsetGet(is,mid);
if (value > cur) {  
　　min = mid+1;
} else if (value < cur) {
　　max = mid-1;
} else {
　　break;
    }
}

if (value == cur) { //查找到返回1，未查找到返回0  
　　if (pos) *pos = mid;
　　return 1;
} else {
　　if (pos) *pos = min;
　　return 0;
    }
  }
}

4.2.2 插入元素

插入元素比较简单，不再赘述源代码，感兴趣的同学可以查看方法

intset *intsetAdd(intset *is, int64_t value, uint8_t *success);  

4.2.3 删除元素

删除元素比较简单，不再赘述源代码，感兴趣的同学可以查看方法

intset *intsetRemove（intset *is, int64_t value, int*success）  

4.3 应用场景

当Redis集合类型的元素都是整数并且都处在64位有符号整数范围之内时，使用该结构体存储。

在两种情况下，底层编码会发生转换。

1.一种情况为当元素个数超过一定数量之后（默认值为512），即使元素类型仍然是整型，也会将编码转换为hashtable。
2.往集合中添加了非整型变量

五、字典

字典底层类似Java的HashMap，但是扩容的方式有一定的区别。

5.1 基本数据结构

哈希表

typedef struct dictht {  
    // 二维数组
    dictEntry **table;
    // table总大小
    unsigned long size;
    // 掩码=size-1
    unsigned long sizemask;
    // 已经保存的键值对
    unsigned long used;
} dictht;

二维数组中的键值对

typedef struct dictEntry {  
    //键
    void *key;
    //值
    union {
        void *val; //值
        uint64_t u64;
        int64_t s64; //过期时间
        double d;
    } v;
    // hash冲突的next指针
    struct dictEntry *next;
} dictEntry;

字典，使用Hash表包了一层

typedef struct dict {  
    //操作类型
    dictType *type;
    // 依赖的数据
    void *privdata;
    // Hash表
    dictht ht[2];
    // -1代表没有进行rehash值，否则代表hash操作进行到了哪个索引
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    // 当前运行的迭代器数
    unsigned long iterators; /* number of iterators currently running */
} dict;

5.2基本操作

5.2.1 添加元素

dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing) /* 入参字典、键、Hash 表节点地址 */  
{
    long index;
    dictEntry *entry;
    dictht *ht;

    if (dictIsRehashing(d)) _dictRehashStep(d); /* 该字典是否在进行 rehash 操作，是则执行一次 rehash */

    if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1) /* 查找键，找到则直接返回 -1，并把老节点存入 existing 字段，否则把新节点的索引值返回。如果遇到 Hash 表容量不足，则进行扩容 */
        return NULL;

    ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0]; /* 是否在进行 rehash 操作中，是则插入至散列表 ht[1] 中，否则插入散列表 ht[0] */
    entry = zmalloc(sizeof(*entry)); /* 申请新节点内存 */
    entry->next = ht->table[index]; /* 将该节点的 next 指针指向 ht->table[index] 指针指向的位置 */
    ht->table[index] = entry; /* 将 ht->table[index] 指针指向该节点 */
    ht->used++;

    dictSetKey(d, entry, key); /* 给新节点存入键信息 */
    return entry;
}

其中查找元素的代码

dictHashKey(d,key), existing)// 根据字典的hash函数得到key的hash值

idx = hash & d->ht[table].sizemask; //利用key的hash值与掩码进行与操作(因为与操作的速度比取余快，也就是为什么要存一个掩码)  

5.2.2 扩容

int dictExpand(dict *d, unsigned long size)  
{
    if (dictIsRehashing(d) || d->ht[0].used > size) /* 如果此时正在扩容，或者是扩容大小小于 ht[0] 的表大小，则抛错 */
        return DICT_ERR;

    dictht n; /* 新 hash 表 */
    unsigned long realsize = _dictNextPower(size); /* 重新计算扩容后的值，必须为 2 的 N 次方幂 */

    /* Rehashing to the same table size is not useful. */
    if (realsize == d->ht[0].size) return DICT_ERR; /* 重新计算的值如果和原来的 size 相等，则无效 */

    /* 分配新 Hash 表，并初始化所有指针为 NULL */
    n.size = realsize;
    n.sizemask = realsize-1;
    n.table = zcalloc(realsize*sizeof(dictEntry*));
    n.used = 0;

    /* 初始化的情况，而不是进行 rehash 操作，就用 ht[0] 来接收值 */
    if (d->ht[0].table == NULL) {
        d->ht[0] = n;
        return DICT_OK;
    }

    /* 准备第二个 Hash 表，以便执行渐进式哈希操作 */
    d->ht[1] = n; /* 扩容后的新内存放入 ht[1] 中 */
    d->rehashidx = 0; /* 非默认的 -1，表示需进行 rehash */
    return DICT_OK;
}

redis中的key可能有成千上万，如果一次性扩容，会对性能造成巨大的影响，所以redis使用渐进式扩容，每次执行插入，删除，查找，修改等操作前，都先判断当前字典的rehash操作是否在进行，如果是在进行中，就对当前节点进行rehash操作，只执行一次。除此之外，当服务器空闲时，也会调用incrementallyRehash函数进行批量操作，每次100个节点，大概一毫秒。将rehash操作进行分而治之。

渐进式rehash源码

int dictRehash(dict *d, int n) {  
    int empty_visits = n*10; /* 最大访问的空桶的数量，n*10 */
    if (!dictIsRehashing(d)) return 0; /* dict 没有正在进行 rehash 时，直接返回 */

    while(n-- && d->ht[0].used != 0) { /* n 为最多迁移元素数量 */
        dictEntry *de, *nextde;

        assert(d->ht[0].size > (unsigned long)d->rehashidx); /* 为防止 rehashidx 越界，当 rehashidx 大于 ht[0] 的数组大小时，不继续执行 */
        while(d->ht[0].table[d->rehashidx] == NULL) { /* 当 rehashidx 位置的桶为空时，继续向下遍历，直到桶不为空或者达到最大访问空桶的数量 */
            d->rehashidx++;
            if (--empty_visits == 0) return 1; //最大访问空桶数量-1,若减完，则退出
        }
        de = d->ht[0].table[d->rehashidx]; 

        while(de) { // 遍历桶中元素，移动元素至新表
            uint64_t h;

            nextde = de->next;
            h = dictHashKey(d, de->key) & d->ht[1].sizemask; 
            de->next = d->ht[1].table[h]; // 头插法
            d->ht[1].table[h] = de;
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        }
        d->ht[0].table[d->rehashidx] = NULL; // ht[0] 对应桶置为空 
        d->rehashidx++;
    }

    if (d->ht[0].used == 0) { // 检查是否已经 rehash 完成
        zfree(d->ht[0].table);
        d->ht[0] = d->ht[1];
        _dictReset(&d->ht[1]);
        d->rehashidx = -1;
        return 0;
    }
    return 1;
}

5.2.3 查找元素

更新和删除操作大同小异，不在赘述

5.3 应用场景

1.总长度超过512字节或者单个元素长度大于64的Hash
2.总长度超过512字节或者单个元素长度大于64的set

六、压缩列表

redis使用字节数据表示压缩列表，尽最大可能节省空间。

其中，coding字段表示content的编码，其长度是动态变化的。如下表

encoding字段第1个字节的前2位，可以判断content字段存储的是整数或者字节数组。当content存储的是字节数组时，后续字节标识字节数组的实际长度；当content存储的是整数时，可根据第3、第4位判断整数的具体类型。而当encoding字段标识当前元素存储的是0～12的立即数时，数据直接存储在encoding字段的最后4位，此时没有content字段。

举个例子

6.1 数据结构

因为解码过程比较繁琐，每次解码都需要性能损耗，为此定义了结构体zlentry，用于表示解码后的压缩列表元素

typedef struct zlentry {  
    //previous_entry_length 长度
    unsigned int prevrawlensize; /* Bytes used to encode the previous entry len*/
    // previous_entry_length
    unsigned int prevrawlen;     /* Previous entry len. */
    //encoding 长度
    unsigned int lensize;        /* Bytes used to encode this entry type/len.
                                    For example strings have a 1, 2 or 5 bytes
                                    header. Integers always use a single byte.*/
    // 内容的长度
    unsigned int len;            /* Bytes used to represent the actual entry.
                                    For strings this is just the string length
                                    while for integers it is 1, 2, 3, 4, 8 or
                                    0 (for 4 bit immediate) depending on the
                                    number range. */
    //首部长度
    unsigned int headersize;     /* prevrawlensize + lensize. */
    //编码
    unsigned char encoding;      /* Set to ZIP_STR_* or ZIP_INT_* depending on
                                    the entry encoding. However for 4 bits
                                    immediate integers this can assume a range
                                    of values and must be range-checked. */
    // 当前元素的首地址
    unsigned char *p;            /* Pointer to the very start of the entry, that
                                    is, this points to prev-entry-len field. */
} zlentry;

6.2 解码

解码分为两步，解码previousentrylength和解码coding

1.解码previousentrylength

#define ZIP_DECODE_PREVLEN(ptr, prevlensize, prevlen) do { 
    ZIP_DECODE_PREVLENSIZE(ptr, prevlensize); 
    // prevlensize=1时，则 ptr 的第一个字节标识上一个节点的长度
    if ((prevlensize) == 1) {
        (prevlen) = (ptr)[0]; 
    } else if ((prevlensize) == 5) {            
        assert(sizeof((prevlen)) == 4);        
        // 如果 prevlensize = 5，取后面 4 个字节作为上一节点的长度
        memcpy(&(prevlen), ((char*)(ptr)) + 1, 4);          
        memrev32ifbe(&prevlen);                                               
    }                                                                          
} while(0);

2.解码coding

#define ZIP_DECODE_LENGTH(ptr, encoding, lensize, len) do {
    // 获取当前的编码类型
    ZIP_ENTRY_ENCODING((ptr), (encoding));
    // 如果编码类型为字节数组
    if ((encoding) < ZIP_STR_MASK) {
        // encoding == 00000000
        if ((encoding) == ZIP_STR_06B) {
            // 存储元素的长度数值所需要的字节数设置为 1
            (lensize) = 1;
            // 元素长度为 (ptr)[0] 和 111111 做位运算
            (len) = (ptr)[0] & 0x3f; 
        // encoding == 10000000
        } else if ((encoding) == ZIP_STR_14B) { 
            // 存储元素的长度数值所需要的字节数设置为 2
            (lensize) = 2; 
            // 元素长度为 高八位：(ptr)[0] 和 111111 做位运算 低八位：(ptr)[1]
            (len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1];
        // encoding == 11000000
        } else if ((encoding) == ZIP_STR_32B) {
            // 存储元素的长度数值所需要的字节数设置为 5
            (lensize) = 5; 
            // 元素长度为后 4 位
            (len) = ((ptr)[1] << 24) |
                    (ptr)[2] << 16) |
                    (ptr)[3] <<  8) |
                    ((ptr)[4]);
        } else {
            panic("Invalid string encoding 0x%02X", (encoding)); 
        } 
    } else { 
        // 数值类型长度存储为 1 字节
        (lensize) = 1;
        // 元素长度 
        (len) = zipIntSize(encoding);            
    }            
} while(0);

6.3 编码

static unsigned int zipPrevEncodeLength(unsigned char *p, unsigned int len) {

    // 仅返回编码 len 所需的字节数量
    if (p == NULL) {
        return (len < ZIP_BIGLEN) ? 1 : sizeof(len)+1;

    // 写入并返回编码 len 所需的字节数量
    } else {

        // 1 字节
        if (len < ZIP_BIGLEN) {
            p[0] = len;
            return 1;

        // 5 字节
        } else {
            // 添加 5 字节长度标识
            p[0] = ZIP_BIGLEN;
            // 写入编码
            memcpy(p+1,&len,sizeof(len));
            // 如果有必要的话，进行大小端转换
            memrev32ifbe(p+1);
            // 返回编码长度
            return 1+sizeof(len);
        }
    }
}

6.4 连锁更新

重新回顾一下 如果前驱节点的长度小于254，那么preventrylen成员需要用1字节长度来保存这个长度值。 如果前驱节点的长度大于等于254，那么preventrylen成员需要用5字节长度来保存这个长度值。 举个例子

比如有这么连续的四个节点，大小都是253字节，当最前面加入一个大于254字节的节点，会导致后面的节点因为因为previousentrylength从一个字节变成五个字节而频繁扩容，每次扩容缩容都需要分配空间和复制数据，对性能损耗巨大。

因为连锁更新发生的概率十分十分低，所以redis并没有采取相关的措施去避免 最后附一张连锁更新的流程图

6.5 应用场景

1.所有字符串元素的长度都小于 64 字节并且保存的元素数量小于512个的列表（list）
2.所有字符串元素的长度都小于 64 字节并且保存的元素数量小于512个的哈希表（Hash）
3.所有字符串元素的长度都小于 64 字节并且保存的元素数量小于512个的有序集合（Sorted Set）

七、Stream流

7.1 Stream流结构

redis中Stream流是Redis5.0以后新加入的数据结构，由生产者，消息，消费者，消费组四个部分组成

生产者负责向消息队列中生产消息，消费者消费某个消息流 对于消费组，有以下三点

1.每个消费组可以消费消息队列中的所有消息，且消费组之间独立。
2.每个消费组里有多个消费者，消息队列中的一条消息只能被其中的一个消费者消费。
3.消费组和消费者都维护了已消费待确认的队列

7.2 listpack结构

listpack可以理解为一个字符串序列化队列，可以存储字符串或者整型
c语言中没有定义listpack的结构体，因为listpack本身可以理解为是一个字符串数组。

其中encode编码字段，决定了后面的content的内容形式，具体如下表所示

backlen所占用的每个字节的第一个bit用于标识；0代表结束，1代表尚未结束，每个字节只有7 bit有效，用于从后向前遍历，能够快速找到上一个元素的首字符。

7.2.1 listpack基本操作

该结构查找效率低下，所以只适合在结尾增删，这刚好符合消息队列的操作。

7.2.1.1 增删改节点

listpack中增删改操作都是用的同一个方法lpInsert,实现了在任意位置插入元素。

删除操作转换为用空元素替换操作

代码比较多，主要介绍一下流程

插入元素的流程也比较简单，就和在数组中插入一个元素类似，不做过多的介绍。

7.2.1.2 遍历

类似数组的遍历，从前向后遍历。

7.3 RAX结构

我们经常会用到前缀树来查找一个单词，查找时间复杂度是O(len(单词的长度))

比如上图这样一棵前缀树，包含了两个单词，App，Apear

这是一种典型的空间换时间的方式，但是每个节点存储一个字母，是不是有点浪费了呢，Rax的出现就解决了这个问题。

下图为使用Rax结构保存App和Apear结构

其中使用带中括号表示的是非压缩节点，其它节点为压缩节点。

另外，非压缩节点是按照字典序排序的

7.3.1 数据结构

typedef struct rax {  
    // 头结点
    raxNode *head;
    // 元素数量(key的数量)
    uint64_t numele;
    // 节点数量
    uint64_t numnodes;
} rax;

typedef struct raxNode {  
    // 当前节点是否包含key
    uint32_t iskey:1;     /* Does this node contain a key? */
    // 当前key对应的value是否为null
    uint32_t isnull:1;    /* Associated value is NULL (don't store it). */
    // 是否压缩
    uint32_t iscompr:1;   /* Node is compressed. */
    // 压缩节点长度或者非压缩节点个数
    uint32_t size:29;
    unsigned char data[];
}

压缩节点

非压缩节点

压缩节点与非压缩节点最大的不同，除了iscomper标志字段不同外，压缩节点只有最后一个字符有子节点，而非压缩节点每个字符都有子节点。

7.3.2 Stream底层实现

如果Stream底层将消息都存放在listpack中，会存在性能问题。当查询消息的时候，需要遍历listpack，插入消息的时候，需要重新分配一块很大的空间。

7.3.2.1 初始化

rax *raxNew(void) {  
    rax *rax = rax_malloc(sizeof(*rax));
    if (rax == NULL) return NULL;
    rax->numele = 0;
    rax->numnodes = 1;
    rax->head = raxNewNode(0,0);
    if (rax->head == NULL) {
        rax_free(rax);
        return NULL;
    } else {
        return rax;
    }
}

7.3.2.2 查找元素

/**
 * 根据key获取对应的value
 * @param rax 待查找的rax
 * @param s 待查找的key
 * @param len s的长度
 * @return
 */
void *raxFind(rax *rax, unsigned char *s, size_t len) {  
    raxNode *h;

    debugf("### Lookup: %.*s\n", (int)len, s);
    int splitpos = 0;
    size_t i = raxLowWalk(rax,s,len,&h,NULL,&splitpos,NULL);
    if (i != len || (h->iscompr && splitpos != 0) || !h->iskey)
        return raxNotFound;
    return raxGetData(h); //返回对应的value
}

可以看到，主要的代码在raxLowWalk方法中

/**
 *
 * @param rax 待查找的rax
 * @param s 待查找的key
 * @param len s的长度
 * @param stopnode 终止的节点 要么匹配完成，要么没找到。。
 * @param plink 父节点指向stopnode的指针的地址
 * @param splitpos 压缩节点的匹配位置
 * @param ts 记录路径
 * @return
 */
static inline size_t raxLowWalk(rax *rax, unsigned char *s, size_t len, raxNode **stopnode, raxNode ***plink, int *splitpos, raxStack *ts) {  
    // 从根节点开始查找
    raxNode *h = rax->head;
    raxNode **parentlink = &rax->head;

    // 当前匹配字符位置
    size_t i = 0; /* Position in the string. */
    // 当前匹配节点位置
    size_t j = 0; /* Position in the node children (or bytes if compressed).*/

    // 当前节点有子节点且s字符串没有遍历完
    while(h->size && i < len) {
        debugnode("Lookup current node",h);
        unsigned char *v = h->data;

        if (h->iscompr) {
            //压缩节点判断是否完全匹配
            for (j = 0; j < h->size && i < len; j++, i++) {
                if (v[j] != s[i]) break;
            }
            // 没有遍历完字符串，退出
            if (j != h->size) break;
        } else {
            /* Even when h->size is large, linear scan provides good
             * performances compared to other approaches that are in theory
             * more sounding, like performing a binary search. */
            // 非压缩节点
            for (j = 0; j < h->size; j++) {
                if (v[j] == s[i]) break;
            }
            // 未在非压缩节点找到字符串
            if (j == h->size) break;
            // 压缩节点可以匹配
            i++;
        }

        // 记录路径
        if (ts) raxStackPush(ts,h); /* Save stack of parent nodes. */

        raxNode **children = raxNodeFirstChildPtr(h);
        if (h->iscompr) j = 0; /* Compressed node only child is at index 0. */
        // 移动到第j个子节点
        memcpy(&h,children+j,sizeof(h));
        parentlink = children+j;
        j = 0; /* If the new node is compressed and we do not
                  iterate again (since i == l) set the split
                  position to 0 to signal this node represents
                  the searched key. */
    }
    debugnode("Lookup stop node is",h);
    if (stopnode) *stopnode = h;
    if (plink) *plink = parentlink;
    if (splitpos && h->iscompr) *splitpos = j;
    return i;
}

这里的步骤比较简单

1.初始化变量
2.从根节点查找直到当前节点无子节点或者s字符串遍历完毕 如果是压缩节点，节点中字符需要和s中的字符完全匹配 如果是非压缩节点，需要找到至少一个与S中字符匹配的字符
3.如果匹配成功，就查找子节点。

7.3.2.3 添加元素

向rax中添加key-value对有两种方式，覆盖和不覆盖原有key

对应的方法分别为

int raxInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) {  
    return raxGenericInsert(rax,s,len,data,old,1);
}

int raxTryInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) {  
    return raxGenericInsert(rax,s,len,data,old,0);
}

两者都是调用同一个方法raxGenericInsert。 该接口方法主要分为以下几步

1.查找key是否存在

i = raxLowWalk(rax,s,len,&h,&parentlink,&j,NULL);  

2 key存在的情况下，直接更新节点数据

       // 如果之前节点没有数据，就分配一个空间
        if (!h->iskey || (h->isnull && overwrite)) {
            h = raxReallocForData(h,data);
            if (h) memcpy(parentlink,&h,sizeof(h));
        }
        if (h == NULL) {
            errno = ENOMEM;
            return 0;
        }

        /* Update the existing key if there is already one. */
        // 更新数据
        if (h->iskey) {
            if (old) *old = raxGetData(h);
            if (overwrite) raxSetData(h,data);
            errno = 0;
            return 0; /* Element already exists. */
        }

3 key不存在的情况下，最后停留在某个压缩节点上

key不存在时，就分为多种情况，这里借用redis源码注解的例子

原来有个rax树，长这样

有以下几种插入情况

1)插入 ANNIENTARE

2）插入ANNIBALI

3）插入 AGO

和第一种类似，只是右边的节点变成了非压缩节点

4）插入 CIAO

5)插入 ANNI

上面列举了五种情况，可以分为两类：

第一种是新插入的key是当前节点的一部分，这时我们只需要拆分压缩节点，并设置新的key即可

第二种是新插入的key与压缩节点的某个位置不匹配，这时我们需要在拆分后的相应位置的非压缩节点中，插入新key的不匹配字符，之后将新key的剩余部分，插入到这个非压缩节点的子节点中

源码中细节较多，不再细讲 感兴趣的同学可以查看

int raxGenericInsert(rax  ＊rax,  unsigned  char  ＊s,  size_t  len,  void  ＊data,  void＊＊old, int overwrite)  

7.3.2.4 添加元素删除节点

删除接口用于删除rax中某个key，依旧拿redis源码中注释的例子 原来有这样一个rax，其有两个key，foo和foobar

也有以下几种情况 1）删除foo

2）假设原来的rax是这样的，其有两个key，foobar,footer

7.3 stream消息

stream流就像是一个消息链表，依赖于Rax结构和listpack结构，本节主要介绍消息流的增删查操作。

typedef struct stream {  
    // 指向rax树
    rax *rax;              
    // 元素个数
    uint64_t length;       
    // 指向最后一个消息
    streamID last_id;      
    // 消费组
    rax *cgroups;          
} stream;

结构如下图所示

7.3.1 基本操作

7.3.1.1 添加消息

redis提供streamAppendItem 函数，向stream中添加一个新的消息

/**
 *
 * @param s 待插入的数据流
 * @param argv 消息内容
 * @param numfields 消息数量
 * @param added_id 消息id
 * @param use_id 调用方定义的id
 * @return
 */
int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id)  

代码比较多就不贴了，画一张流程图描述细节

7.3.1.2 新增消费组

消费组也是保存在rax树中，以消费组的名称为key，消费组的streamCG结构为value。

treamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id) {  
    //当前消息流没有消费组，就新建一个
    if (s->cgroups == NULL) s->cgroups = raxNew();

    // 查找是否有重名消费组，有就直接返回
    if (raxFind(s->cgroups,(unsigned char*)name,namelen) != raxNotFound)
        return NULL;

    // 新建消费组
    streamCG *cg = zmalloc(sizeof(*cg));
    cg->pel = raxNew();
    cg->consumers = raxNew();
    cg->last_id = *id;
    // 将消费组插入到消费组树中
    raxInsert(s->cgroups,(unsigned char*)name,namelen,cg,NULL);
    return cg;
}

7.3.1.3 新建消费者

redis没有提供为消费组中新增消费者的方法。在查询消费者的时候，如果不存在，就会新增。

7.3.1.4 删除消息

删除消息通过将listpack中消息的标志位设为已删除，并不是真正的删除。 如果整个listpack的消息都被删除了，才会从rax中释放该节点。

void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {  
    // 当前消息所在的listpack
    unsigned char *lp = si->lp;
    int64_t aux;

    // 标记删除位
    int flags = lpGetInteger(si->lp_flags);
    flags |= STREAM_ITEM_FLAG_DELETED;
    lp = lpReplaceInteger(lp,&si->lp_flags,flags);


    //修改有效的消息数量
    unsigned char *p = lpFirst(lp);
    aux = lpGetInteger(p);

    // 如果只有待删除的消息，就直接释放listpack
    if (aux == 1) {
        /* If this is the last element in the listpack, we can remove the whole
         * node. */
        lpFree(lp);
        raxRemove(si->stream->rax,si->ri.key,si->ri.key_len,NULL);
    } else {
    // 更新统计信息
        lp = lpReplaceInteger(lp,&p,aux-1);
    // 查找删除节点p
        p = lpNext(lp,p);
        aux = lpGetInteger(p);
        lp = lpReplaceInteger(lp,&p,aux+1);

        // 更新listpack指针，可能因为扩容缩容而变化
        if (si->lp != lp)
            raxInsert(si->stream->rax,si->ri.key,si->ri.key_len,lp,NULL);
    }

7.3.1.5.查找消费组

主要是利用rax查询接口

streamCG *streamLookupCG(stream *s, sds groupname) {  
    if (s->cgroups == NULL) return NULL;
    streamCG *cg = raxFind(s->cgroups,(unsigned char*)groupname,
                           sdslen(groupname));
    return (cg == raxNotFound) ? NULL : cg;
}

7.3.1.6 查找消费者

streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int create) {  
    // 在消费者的rax中查找指定的消费者
    streamConsumer *consumer = raxFind(cg->consumers,(unsigned char*)name,
                               sdslen(name));
    if (consumer == raxNotFound) {
        if (!create) return NULL;
        // 如果没有找到，新建一个消费者，插入到消费者rax树中
        consumer = zmalloc(sizeof(*consumer));
        consumer->name = sdsdup(name);
        consumer->pel = raxNew();
        raxInsert(cg->consumers,(unsigned char*)name,sdslen(name),
                  consumer,NULL);
    }
    consumer->seen_time = mstime();
    return consumer;
}

八、Redis线程模型演进

8.1 redis3.0

redis内部使用文件事件处理器file event handler,这个文件处理器是单线程的，所以我们经常说的redis是单线程模型。

客户端 Socket01 向 Redis 的 Server Socket 请求建立连接，此时 Server Socket 会产生一个 AEREADABLE 事件，IO 多路复用程序监听到 server socket 产生的事件后，将该事件压入队列中。文件事件分派器从队列中获取该事件，交给连接应答处理器。连接应答处理器会创建一个能与客户端通信的 Socket01，并将该 Socket01 的 AEREADABLE 事件与命令请求处理器关联。

假设此时客户端发送了一个 set key value 请求，此时 Redis 中的 Socket01 会产生 AEREADABLE 事件，IO 多路复用程序将事件压入队列，此时事件分派器从队列中获取到该事件，由于前面 Socket01 的 AEREADABLE 事件已经与命令请求处理器关联，因此事件分派器将事件交给命令请求处理器来处理。命令请求处理器读取 Scket01 的 set key value 并在自己内存中完成 set key value 的设置。操作完成后，它会将 Scket01 的 AE_WRITABLE 事件与令回复处理器关联。

如果此时客户端准备好接收返回结果了，那么 Redis 中的 Socket01 会产生一个 AEWRITABLE 事件，同样压入队列中，事件分派器找到相关联的命令回复处理器，由命令回复处理器对 socket01 输入本次操作的一个结果，比如 ok，之后解除 Socket01 的 AEWRITABLE 事件与命令回复处理器的关联。

8.2 redis4.0

redis4.0开始引入了多线程，除了主线程，redis有后台线程进行一些边缘的缓慢的操作，比如释放无用连接，rehash迁移等操作。

8.3 redis6.0

在redis6.0中，真正引入了多线程

九、总结

本文介绍了redis基本数据类型以及redis5.0以后的新特性。工作中我们也常常使用redis进行各种逻辑的处理，而了解其源码可以避免踩很多坑。另外，redis底层的设计也有很多值得学习的地方，比如更高效的使用内存和提升运算的时间复杂度，了解这些可以帮助我们在性能优化中有更多的思路。

pdf版本下载地址： https://pan.baidu.com/s/1kcdXOMgtxjEWOecQzdXjOA?pwd=xkgb 提取码: xkgb

一.引言

引言：“操作系统的线程状态和java的线程状态有什么关系？”这是校招时被问到的一个问题。当时只顾着看博文、面经等零散的资料，没有形成系统的知识体系，一时语塞，答的不是很对。在网上也没找到足够细致的讲解博文，于是整理出了这篇内容。

Java的线程状态牵扯到了同步语义，要探讨Java的线程状态的，必不可免要回顾其锁机制。因此本文的主要分为两大块：一是Synchronized源码粗析，分析了各类锁的进入、释放、升级过程，并大致说明了monitor原理；二是介绍了线程的实现方式和Java线程状态转换的部分细节。

P.S. 本文内容较啰嗦，时间不充裕的同学可以直接看2.6小结及3.3小结。

二. Synchronized锁

Java采用synchronized关键字、以互斥同步的方式的解决线程安全问题，那么什么是线程安全呢？这里引用《Java并发编程实战》作者Brian Goetz给出的定义：

当多个线程同时访问一个对象时，如果不用考虑这些线程在运行时环境下的调度和交替执行，也不需要进行额外的同步，或者在调用方进行任何其他的协调操作，调用这个对象的行为都可以获得正确的结果，那就称这个对象是线程安全的。 —— Brian Goetz

2.1 Synchronized的使用

先写过个demo，大致过一下synchronized的使用，包含同步代码块、实例方法和静态方法。

  public synchronized void test1(){
  }

  public void test2(){
    synchronized(new Test()){
    }
  }

  public static synchronized void test3(){
  }

反编译可查看字节码：

  public synchronized void test1();
    descriptor: ()V
    flags: ACC_PUBLIC, ACC_SYNCHRONIZED    // here

  public void test2();
    descriptor: ()
    flags: ACC_PUBLIC
    Code:
      stack=2, locals=3, args_size=1
         0: new           #2                  // class com/easy/helloworld/Test
         3: dup
         4: invokespecial #3                  // Method "<init>":()V
         7: dup
         8: astore_1
         9: monitorenter                   // here
        10: aload_1
        11: monitorexit                    // here
        12: goto          20
        15: astore_2
        16: aload_1
        17: monitorexit                    // here
        18: aload_2
        19: athrow
        20: return

  public static synchronized void test3();
    descriptor: ()V
    flags: ACC_PUBLIC, ACC_STATIC, ACC_SYNCHRONIZED   // here

可以观察到：

• 同步代码：通过moniterenter、moniterexit 关联到到一个monitor对象，进入时设置Owner为当前线程，计数+1、退出-1。除了正常出口的 monitorexit，还在异常处理代码里插入了 monitorexit。
• 实例方法：隐式调用moniterenter、moniterexit
• 静态方法：隐式调用moniterenter、moniterexit

2.2 Moniterenter、Moniterexit

monitorenter和monitorexit这两个jvm指令，主要是基于 Mark Word和Object monitor来实现的。

在 JVM 中，对象在内存中分为三块区域：

• 对象头：由Mark Word和Klass Point构成。

  • Mark Word（标记字段）：用于存储对象自身的运行时数据，例如存储对象的HashCode，分代年龄、锁标志位等信息，是synchronized实现轻量级锁和偏向锁的关键。 64位JVM的Mark Word组成如下：
  • Klass Point（类型指针）：对象指向它的类元数据的指针，虚拟机通过这个指针来确定这个对象是哪个类的实例。

• 实例数据：这部分主要是存放类的数据信息，父类的信息。

• 字节对齐：为了内存的IO性能，JVM要求对象起始地址必须是8字节的整数倍。对于不对齐的对象，需要填充数据进行对齐。

在JDK 1.6之前,synchronized只有传统的锁机制，直接关联到monitor对象，存在性能上的瓶颈。在JDK 1.6后，为了提高锁的获取与释放效率，JVM引入了两种锁机制：偏向锁和轻量级锁。它们的引入是为了解决在没有多线程竞争或基本没有竞争的场景下因使用传统锁机制带来的性能开销问题。这几种锁的实现和转换正是依靠对象头中的Mark Word。

2.3 偏向锁

引入偏向锁的目的：在没有多线程竞争的情况下，尽量减少不必要的轻量级锁的执行。轻量级锁的获取及释放依赖多次CAS原子指令，而偏向锁只依赖一次CAS原子指令。但在多线程竞争时，需要进行偏向锁撤销步骤，因此其撤销的开销必须小于节省下来的CAS开销，否则偏向锁并不能带来收益。JDK 1.6中默认开启偏向锁，可以通过-XX:-UseBiasedLocking来禁用偏向锁。

2.3.1 进入偏向锁

关于HotSpot虚拟机中获取锁的入口，网上主要有两种看法：一为interpreterRuntime.cpp#monitorenter#1608；二为bytecodeInterpreter.cpp#1816。在HotSpot的中，有两处地方对monitorenter指令进行解析：一个是bytecodeInterpreter.cpp#1816 ，另一个在templateTablex8664.cpp#3667。其中，bytecodeInterpreter是JVM中的字节码解释器， templateInterpreter为模板解释器。HotSpot对运行效率有着极其执着的追求，显然会倾向于用模板解释器来实现。R大的读书笔记中有说明，HotSpot中只用到了模板解释器，并没有用到字节码解释器。因此，本文认为montorenter的解析入口为templateTablex8664.cpp#3667。

但模板解释器templateInterpreter都是汇编代码，不易读，且实现逻辑与字节码解释器bytecodeInterpreter大体一致。因此本文的源码都以bytecodeInterpreter来说明，借此窥探synchronized的实现原理。在看代码之前，先介绍几个在偏向锁中会被大量应用的概念，以便后续理解：

prototype_header：JVM中的每个类有一个类似mark word的prototype_header，用来标记该class的epoch和偏向开关等信息。

匿名偏向状态：锁对象mark word标志位为101，且存储的Thread ID为空时的状态(即锁对象为偏向锁，且没有线程偏向于这个锁对象)。

Atomic::cmpxchg_ptr：CAS函数。这个方法有三个参数，依次为exchange_value、dest、compare_value。如果dest的值为compare_value则更新为exchange_value，并返回compare_value。否则，不更新并返回实际原值。

接下来开始源码实现分析，HotSpot中偏向锁的具体实现可参考bytecodeInterpreter.cpp#1816，代码如下：

CASE(_monitorenter): {  
  //锁对象
  oop lockee = STACK_OBJECT(-1);
  // derefing's lockee ought to provoke implicit null check
  CHECK_NULL(lockee);
  // 步骤1
  // 在栈中找到第一个空闲的Lock Record
  // 会找到栈中最高的
  BasicObjectLock* limit = istate->monitor_base();
  BasicObjectLock* most_recent = (BasicObjectLock*) istate->stack_base();
  BasicObjectLock* entry = NULL;
  while (most_recent != limit ) {
    if (most_recent->obj() == NULL) entry = most_recent;
    else if (most_recent->obj() == lockee) break;
    most_recent++;
  }
  // entry不为null，代表还有空闲的Lock Record
  if (entry != NULL) {
    // 将Lock Record的obj指针指向锁对象
    entry->set_obj(lockee);
    int success = false;
    uintptr_t epoch_mask_in_place = (uintptr_t)markOopDesc::epoch_mask_in_place;
    // markoop即对象头的mark word
    markOop mark = lockee->mark();
    intptr_t hash = (intptr_t) markOopDesc::no_hash;
    // 步骤2
    // implies UseBiasedLocking
    // 如果为偏向模式，即判断标识位是否为101
    if (mark->has_bias_pattern()) {
      ...
      // 一顿操作
      anticipated_bias_locking_value =
        (((uintptr_t)lockee->klass()->prototype_header() | thread_ident) ^ (uintptr_t)mark) &
        ~((uintptr_t) markOopDesc::age_mask_in_place);
      // 步骤3
      if  (anticipated_bias_locking_value == 0) {
        // already biased towards this thread, nothing to do
        // 偏向的是自己，啥都不做
        if (PrintBiasedLockingStatistics) {
          (* BiasedLocking::biased_lock_entry_count_addr())++;
        }
        success = true;
      }
      // class的prototype_header不是偏向模式
      else if ((anticipated_bias_locking_value & markOopDesc::biased_lock_mask_in_place) != 0) {
        // 尝试撤销偏向
                ...
      }
      // epoch过期，重新偏向
      else if ((anticipated_bias_locking_value & epoch_mask_in_place) !=0) {
        // try rebias
                ...
        success = true;    
      }
      else {
        // try to bias towards thread in case object is anonymously biased
        // 尝试偏向该线程，只有匿名偏向能成功
        // 构建了匿名偏向的mark word
        markOop header = (markOop) ((uintptr_t) mark & ((uintptr_t)markOopDesc::biased_lock_mask_in_place |(uintptr_t)markOopDesc::age_mask_in_place |epoch_mask_in_place));
        if (hash != markOopDesc::no_hash) {
          header = header->copy_set_hash(hash);
        }
        // 用「或」操作设置thread ID
        markOop new_header = (markOop) ((uintptr_t) header | thread_ident);、
        // 只有匿名偏向才能成功
        if (Atomic::cmpxchg_ptr((void*)new_header, lockee->mark_addr(), header) == header) {
          // cas修改成功    
          if (PrintBiasedLockingStatistics)
            (* BiasedLocking::anonymously_biased_lock_entry_count_addr())++;
        }
        else {
          // 失败说明存在竞争，进入monitorenter
          CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
        }
        success = true;
      }
    }
    // 步骤4
    // traditional lightweight locking
    // false走轻量级锁逻辑
    if (!success) {
      // 构造一个无锁状态的Displaced Mark Word，并将lock record指向它
      markOop displaced = lockee->mark()->set_unlocked();
      entry->lock()->set_displaced_header(displaced);
      bool call_vm = UseHeavyMonitors;
      if (call_vm || Atomic::cmpxchg_ptr(entry, lockee->mark_addr(), displaced) != displaced) {
        // 如果CAS替换不成功，代表锁对象不是无锁状态，这时候判断下是不是锁重入
        // Is it simple recursive case?
        if (!call_vm && THREAD->is_lock_owned((address) displaced->clear_lock_bits())) {
          // 如果是锁重入，则直接将Displaced Mark Word设置为null
          // 轻量级锁重入是使用lock record的数量来计入的
          entry->lock()->set_displaced_header(NULL);
        } else {
          CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
        }
      }
    }
    UPDATE_PC_AND_TOS_AND_CONTINUE(1, -1);
  } else {
    // 没拿到lock record，重新执行
    istate->set_msg(more_monitors);
    UPDATE_PC_AND_RETURN(0); // Re-execute
  }
}

偏向锁流程：

步骤 1、从当前线程的栈中找到一个空闲的Lock Record，并指向当前锁对象。

步骤 2、获取对象的markOop数据mark，即对象头的Mark Word；

步骤 3、判断锁对象的mark word是否是偏向模式，即低3位是否为101。若不是，进入步骤4。若是，计算anticipated_bias_locking_value，判断偏向状态：

步骤 3.1、anticipated_bias_locking_value若为0，代表偏向的线程是当前线程且mark word的epoch等于class的epoch，这种情况下直接执行同步代码块，什么都不用做。

步骤 3.2、判断class的prototype_header是否为非偏向模式。若为非偏向模式，CAS尝试将对象恢复为无锁状态。无论cas是否成功都会进入轻量级锁逻辑。

步骤 3.3、如果epoch偏向时间戳已过期，则需要重偏向。利用CAS指令将锁对象的mark word替换为一个偏向当前线程且epoch为类的epoch的新的mark word。

步骤 3.4、CAS将偏向线程改为当前线程，如果当前是匿名偏向（即对象头中的bit field存储的Thread ID为空）且无并发冲突，则能修改成功获取偏向锁，否则进入锁升级的逻辑。

步骤 4、走到一步会进行轻量级锁逻辑。构造一个无锁状态的mark word，然后存储到Lock Record。设置为无锁状态的原因是：轻量级锁解锁时是将对象头的mark wordcas替换为Lock Record中的Displaced Mark Word，所以设置为无锁状态。如果是锁重入，则将Lock Record的Displaced Mark Word设置为null，放到栈帧中，起到计数作用。

以上是偏向锁加锁的大致流程，如果当前锁已偏向其他线程||epoch值过期||class偏向模式关闭||获取偏向锁的过程中存在并发冲突，都会进入到InterpreterRuntime::monitorenter方法， 在该方法中会进行偏向锁撤销和升级。流程如下图所示：

Issue：有的同学可能会问了，对象一开始不是无锁状态吗，为什么上述偏向锁逻辑没有判断无锁状态的锁对象（001）？

只有匿名偏向的对象才能进入偏向锁模式。JVM启动时会延时初始化偏向锁，默认是4000ms。初始化后会将所有加载的Klass的prototype header修改为匿名偏向样式。当创建一个对象时，会通过Klass的prototype_header来初始化该对象的对象头。简单的说，偏向锁初始化结束后，后续所有对象的对象头都为匿名偏向样式，在此之前创建的对象则为无锁状态。而对于无锁状态的锁对象，如果有竞争，会直接进入到轻量级锁。这也是为什么JVM启动前4秒对象会直接进入到轻量级锁的原因。

为什么需要延迟初始化？

JVM启动时必不可免会有大量sync的操作，而偏向锁并不是都有利。如果开启了偏向锁，会发生大量锁撤销和锁升级操作，大大降低JVM启动效率。

因此，我们可以明确地说，只有锁对象处于匿名偏向状态，线程才能拿到到我们通常意义上的偏向锁。而处于无锁状态的锁对象，只能进入到轻量级锁状态。

2.3.2 偏向锁的撤销

偏向锁的 撤销（revoke）是一个很特殊的操作，为了执行撤销操作，需要等待全局安全点，此时所有的工作线程都停止了执行。偏向锁的撤销操作并不是将对象恢复到无锁可偏向的状态（注意区分偏向锁撤销和释放这两个概念，撤销的触发见上图），而是在偏向锁的获取过程中，发现竞争时，直接将一个被偏向的对象升级到被加了轻量级锁的状态。这个操作的具体完成方式如下：

IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorenter(JavaThread* thread, BasicObjectLock* elem))  
  ...
  Handle h_obj(thread, elem->obj());
  assert(Universe::heap()->is_in_reserved_or_null(h_obj()),
         "must be NULL or an object");
    // 开启了偏向锁
  if (UseBiasedLocking) {
    // Retry fast entry if bias is revoked to avoid unnecessary inflation
    ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);
  } else {
    ObjectSynchronizer::slow_enter(h_obj, elem->lock(), CHECK);
  }
  ...

如果开启了JVM偏向锁，则会进入到ObjectSynchronizer::fast_enter方法中。

void ObjectSynchronizer::fast_enter(Handle obj, BasicLock* lock, bool attempt_rebias, TRAPS) {  
 //再次校验
 if (UseBiasedLocking) {
    if (!SafepointSynchronize::is_at_safepoint()) {
      //不在安全点的执行
      BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);
      if (cond == BiasedLocking::BIAS_REVOKED_AND_REBIASED) {
        return;
      }
    } else {
      assert(!attempt_rebias, "can not rebias toward VM thread");    
      //批量撤销,底层调用bulk_revoke_or_rebias_at_safepoint
      BiasedLocking::revoke_at_safepoint(obj);
    }
    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");
 }
 slow_enter (obj, lock, THREAD) ;
}

主要看BiasedLocking::revoke_and_rebias方法。这个方法的主要作用像它的方法名：撤销或者重偏向。第一个参数封装了锁对象和当前线程，第二个参数代表是否允许重偏向，这里是true。

BiasedLocking::Condition BiasedLocking::revoke_and_rebias(Handle obj, bool attempt_rebias, TRAPS) {  
  assert(!SafepointSynchronize::is_at_safepoint(), "must not be called while at safepoint");
  markOop mark = obj->mark(); //获取锁对象的对象头
  if (mark->is_biased_anonymously() && !attempt_rebias) {
    // 如果锁对象为匿名偏向状态且不允许重偏向下，进入该分支。在一个非全局安全点进行偏向锁撤销
    markOop biased_value       = mark;
    // 创建一个匿名偏向的markword
    markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());
    // 通过cas重新设置偏向锁状态
    markOop res_mark = (markOop) Atomic::cmpxchg_ptr(unbiased_prototype, obj->mark_addr(), mark);
    if (res_mark == biased_value) {// 如果CAS成功，返回偏向锁撤销状态
      return BIAS_REVOKED;
    }
  } else if (mark->has_bias_pattern()) {
    // 锁为偏向模式（101）会走到这里 
    Klass* k = obj->klass(); 
    markOop prototype_header = k->prototype_header();
    // 如果对应class关闭了偏向模式
    if (!prototype_header->has_bias_pattern()) {
      markOop biased_value       = mark;
      // CAS更新对象头markword为非偏向锁
      markOop res_mark = (markOop) Atomic::cmpxchg_ptr(prototype_header, obj->mark_addr(), mark);
      assert(!(*(obj->mark_addr()))->has_bias_pattern(), "even if we raced, should still be revoked");
      return BIAS_REVOKED; // 返回偏向锁撤销状态
    } else if (prototype_header->bias_epoch() != mark->bias_epoch()) {
      // 如果epoch过期，则进入当前分支
      if (attempt_rebias) {
        // 如果允许重偏
        assert(THREAD->is_Java_thread(), "");
        markOop biased_value       = mark;
        markOop rebiased_prototype = markOopDesc::encode((JavaThread*) THREAD, mark->age(), prototype_header->bias_epoch());
        // 通过CAS操作， 将本线程的 ThreadID 、时间戳、分代年龄尝试写入对象头中
        markOop res_mark = (markOop) Atomic::cmpxchg_ptr(rebiased_prototype, obj->mark_addr(), mark);
        if (res_mark == biased_value) { //CAS成功，则返回撤销和重新偏向状态
          return BIAS_REVOKED_AND_REBIASED;
        }
      } else {
        // 如果不允许尝试获取偏向锁，进入该分支取消偏向
        // 通过CAS操作更新分代年龄
        markOop biased_value       = mark;
        markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());
        markOop res_mark = (markOop) Atomic::cmpxchg_ptr(unbiased_prototype, obj->mark_addr(), mark);
        if (res_mark == biased_value) { //如果CAS操作成功，返回偏向锁撤销状态
          return BIAS_REVOKED;
        }
      }
    }
  }
  //执行到这里有以下两种情况：
  //1.对象不是偏向模式
  //2.上面的cas操作失败
  HeuristicsResult heuristics = update_heuristics(obj(), attempt_rebias);
  if (heuristics == HR_NOT_BIASED) {
    // 非偏向从这出去
    // 轻量级锁、重量级锁
    return NOT_BIASED;
  } else if (heuristics == HR_SINGLE_REVOKE) {
    // 撤销单个线程
    // Mark，最常见的执行分支
    // Mark，最常见的执行分支
    // Mark，最常见的执行分支
    Klass *k = obj->klass();
    markOop prototype_header = k->prototype_header();
    if (mark->biased_locker() == THREAD &&
        prototype_header->bias_epoch() == mark->bias_epoch()) {
      // 偏向当前线程且不过期
      // 这里撤销的是偏向当前线程的锁，调用Object#hashcode方法时也会走到这一步
      // 因为只要遍历当前线程的栈就能拿到lock record了，所以不需要等到safe point再撤销。
      ResourceMark rm;
      if (TraceBiasedLocking) {
        tty->print_cr("Revoking bias by walking my own stack:");
      }
      BiasedLocking::Condition cond = revoke_bias(obj(), false, false, (JavaThread*) THREAD);
      ((JavaThread*) THREAD)->set_cached_monitor_info(NULL);
      assert(cond == BIAS_REVOKED, "why not?");
      return cond;
    } else {
      // 下面代码最终会在safepoint调用revoke_bias方法撤销偏向
      VM_RevokeBias revoke(&obj, (JavaThread*) THREAD);
      VMThread::execute(&revoke);
      return revoke.status_code();
    }
  }
  assert((heuristics == HR_BULK_REVOKE) ||
         (heuristics == HR_BULK_REBIAS), "?");
   //批量撤销、批量重偏向的逻辑
  VM_BulkRevokeBias bulk_revoke(&obj, (JavaThread*) THREAD,
                                (heuristics == HR_BULK_REBIAS),
                                attempt_rebias);
  VMThread::execute(&bulk_revoke);
  return bulk_revoke.status_code();
}

这块代码注释写的算是比较清楚，只简单介绍下最常见的情况：锁已经偏向线程A，此时线程B尝试获取锁。这种情况下会走到Mark标记的分支。如果需要撤销的是当前线程，只要遍历当前线程的栈就能拿到lock record，可以直接调用revoke_bias，不需要等到safe point再撤销。在调用Object#hashcode时，也会走到该分支将为偏向锁的锁对象直接恢复为无锁状态。若不是当前线程，会被push到VM Thread中等到safepoint的时候再执行。

VMThread内部维护了一个VMOperationQueue类型的队列，用于保存内部提交的VM线程操作VM_operation。GC、偏向锁的撤销等操作都是在这里被执行。

撤销调用的revoke_bias方法的代码就不贴了。大致逻辑是：

步骤 1、查看偏向的线程是否存活，如果已经死亡，则直接撤销偏向锁。JVM维护了一个集合存放所有存活的线程，通过遍历该集合判断某个线程是否存活。

步骤 2、偏向的线程是否还在同步块中，如果不在，则撤销偏向锁。如果在同步块中，执行步骤3。这里是否在同步块的判断基于上文提到的偏向锁的重入计数方式：在偏向锁的获取中，每次进入同步块的时候都会在栈中找到第一个可用（即栈中最高的）的Lock Record，将其obj字段指向锁对象。每次解锁的时候都会把最低的Lock Record移除掉，所以可以通过遍历线程栈中的Lock Record来判断是否还在同步块中。轻量级锁的重入也是基于Lock Record的计数来判断。

步骤 3、升级为轻量级锁。将偏向线程所有相关Lock Record的Displaced Mark Word设置为null，再将最高位的Lock Record的Displaced Mark Word 设置为无锁状态，然后将对象头指向最高位的Lock Record。这里没有用到CAS指令，因为是在safepoint，可以直接升级成轻量级锁。

2.3.3 偏向锁的释放

偏向锁的释放可参考bytecodeInterpreter.cpp#1923，这里也不贴了。偏向锁的释放只要将对应Lock Record释放就好了，但这里的释放并不会将mark word里面的thread ID去掉，这样做是为了下一次更方便的加锁。而轻量级锁则需要将Displaced Mark Word替换到对象头的mark word中。如果CAS失败或者是重量级锁则进入到InterpreterRuntime::monitorexit方法中。

2.3.4 批量重偏向与撤销

从上节偏向锁的加锁解锁过程中可以看出，当只有一个线程反复进入同步块时，偏向锁带来的性能开销基本可以忽略，但是当有其他线程尝试获得锁时，就需要等到safe point时将偏向锁撤销为无锁状态或升级为轻量级/重量级锁。因此，JVM中增加了一种批量重偏向/撤销的机制以减少锁撤销的开销，而mark word中的epoch也是在这里被大量应用，这里不展开说明。但无论怎么优化，偏向锁的撤销仍有一定不可避免的成本。如果业务场景存在大量多线程竞争，那偏向锁的存在不仅不能提高性能，而且会导致性能下降（偏向锁并不都有利，jdk15默认不开启）。

2.4 轻量级锁

引入轻量级锁的目的：在多线程交替执行同步块的情况下，尽量避免重量级锁使用的操作系统互斥量带来的开销，但是如果多个线程在同一时刻进入临界区，会导致轻量级锁膨胀升级重量级锁，所以轻量级锁的出现并非是要替代重量级锁。

2.4.1 进入轻量级锁

轻量级锁在上文或多或少已经涉及到，其获取流程入口为bytecodeInterpreter.cpp#1816。前大半部分都是偏向锁逻辑，还有一部分为轻量级锁逻辑。在偏向锁逻辑中，cas失败会执行到InterpreterRuntime::monitorenter。在轻量级锁逻辑中，如果当前线程不是轻量级锁的重入，也会执行到InterpreterRuntime::monitorenter。我们再看看InterpreterRuntime::monitorenter方法：

IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorenter(JavaThread* thread, BasicObjectLock* elem))  
  ...
  Handle h_obj(thread, elem->obj());
  assert(Universe::heap()->is_in_reserved_or_null(h_obj()),
         "must be NULL or an object");
  if (UseBiasedLocking) {
    // Retry fast entry if bias is revoked to avoid unnecessary inflation
    ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);
  } else {
    ObjectSynchronizer::slow_enter(h_obj, elem->lock(), CHECK);
  }
  ...
IRT_END  

fast_enter的流程在偏向锁的撤销小节中已经分析过，主要逻辑为revoke_and_rebias：如果当前是偏向模式且偏向的线程还在使用锁，会将锁的mark word改为轻量级锁的状态，并将偏向的线程栈中的Lock Record修改为轻量级锁对应的形式（此时Lock Record是无锁状态），且返回值不是BIAS_REVOKED_AND_REBIASED，会继续执行slow_enter。

我们直接看slow_enter的流程:

void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {  
  // 步骤1
  markOop mark = obj->mark();
  assert(!mark->has_bias_pattern(), "should not see bias pattern here");
  // 步骤2
  // 如果为无锁状态
  if (mark->is_neutral()) {
    // 步骤3
    // 设置mark word到栈 
    lock->set_displaced_header(mark);
    // CAS更新指向栈中Lock Record的指针
    if (mark == (markOop) Atomic::cmpxchg_ptr(lock, obj()->mark_addr(), mark)) {
      TEVENT (slow_enter: release stacklock) ;
      return ;
    }
    // Fall through to inflate() ... cas失败走下面锁膨胀方法
  } else if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {
    // 步骤4
    // 为轻量级锁且owner为当前线程
    assert(lock != mark->locker(), "must not re-lock the same lock");
    assert(lock != (BasicLock*)obj->mark(), "don't relock with same BasicLock");
    // 设置Displaced Mark Word为null，重入计数用
    lock->set_displaced_header(NULL);
    return;
  }
  // 步骤5
  // 走到这一步说明已经是存在多个线程竞争锁了，需要膨胀或已经是重量级锁
  lock->set_displaced_header(markOopDesc::unused_mark());
  // 进入、膨胀到重量级锁的入口
  // 膨胀后再调用monitor的enter方法竞争锁
  ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
}

步骤 1、markOop mark = obj->mark()方法获取对象的markOop数据mark；

步骤 2、mark->is_neutral()方法判断mark是否为无锁状态，标识位001；

步骤 3、如果mark处于无锁状态，把mark保存到BasicLock对象(Lock Record的属性)的displaced_header字段；

步骤 3.1、通过CAS尝试将Mark Word更新为指向BasicLock对象的指针，如果更新成功，表示竞争到锁，则执行同步代码，否则执行步骤4；

步骤 4、如果是重入，则设置Displaced Mark Word为null。

步骤 5、到这说明有多个线程竞争轻量级锁，轻量级锁需要膨胀升级为重量级锁；

结合上文偏向锁的流程，可以整理得到如下的流程图：

2.4.2 轻量级锁的释放

轻量级锁释放的入口在bytecodeInterpreter.cpp#1923。

轻量级锁释放时需要将Displaced Mark Word替换回对象头的mark word中。如果CAS失败或者是重量级锁则进入到InterpreterRuntime::monitorexit方法中。monitorexit直接调用slow_exit方法释放Lock Record。直接看slow_exit：

IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorexit(JavaThread* thread, BasicObjectLock* elem))  
  Handle h_obj(thread, elem->obj());
  assert(Universe::heap()->is_in_reserved_or_null(h_obj()),
         "must be NULL or an object");
  if (elem == NULL || h_obj()->is_unlocked()) {
    THROW(vmSymbols::java_lang_IllegalMonitorStateException());
  }
  // 直接调用slow_exit
  ObjectSynchronizer::slow_exit(h_obj(), elem->lock(), thread);
  // Free entry. This must be done here, since a pending exception might be installed on
  // exit. If it is not cleared, the exception handling code will try to unlock the monitor again.
  elem->set_obj(NULL);
IRT_END

void ObjectSynchronizer::slow_exit(oop object, BasicLock* lock, TRAPS) {  
  fast_exit (object, lock, THREAD) ;
}

void ObjectSynchronizer::fast_exit(oop object, BasicLock* lock, TRAPS) {  
  ...
  // displaced header就是对象mark word的拷贝
  markOop dhw = lock->displaced_header();
  markOop mark ;
  if (dhw == NULL) {
     // 什么也不做
     // Recursive stack-lock. 递归堆栈锁
     // Diagnostics -- Could be: stack-locked, inflating, inflated. 
     ...
     return ;
  }
  mark = object->mark() ;
  // 此处为轻量级锁的释放过程，使用CAS方式解锁。
  // 如果对象被当前线程堆栈锁定，尝试将displaced header和锁对象中的MarkWord替换回来。
  // If the object is stack-locked by the current thread, try to
  // swing the displaced header from the box back to the mark.
  if (mark == (markOop) lock) {
     assert (dhw->is_neutral(), "invariant") ;
     if ((markOop) Atomic::cmpxchg_ptr (dhw, object->mark_addr(), mark) == mark) {
        TEVENT (fast_exit: release stacklock) ;
        return;
     }
  }
  //走到这里说明已经是重量级锁或者解锁时发生了竞争，膨胀后再调用monitor的exit方法释放
  ObjectSynchronizer::inflate(THREAD, object)->exit (true, THREAD) ;
}

最后执行的是如果是fast_exit方法。如果是轻量级锁，尝试cas替换mark word。若解锁时有竞争，会调用inflate方法进行重量级锁膨胀，升级到到重量级锁后再执行exit方法。

2.5 重量级锁

2.5.1 重量级锁的进入

重量级锁通过对象内部的监视器（monitor）实现，其依赖于底层操作系统的Mutex Lock实现，需要额外的用户态到内核态切换的开销。由上文分析，slow_enter获取轻量级锁未成功时，会在inflate中完成锁膨胀：

ObjectMonitor * ATTR ObjectSynchronizer::inflate (Thread * Self, oop object) {  
  ...
  for (;;) {
      const markOop mark = object->mark() ;
      assert (!mark->has_bias_pattern(), "invariant") ;  
      // mark是以下状态中的一种：
      // *  Inflated（重量级锁状态）     - 直接返回
      // *  Stack-locked（轻量级锁状态） - 膨胀
      // *  INFLATING（膨胀中）    - 忙等待直到膨胀完成
      // *  Neutral（无锁状态）      - 膨胀
      // *  BIASED（偏向锁）       - 非法状态，在这里不会出现

      // CASE: inflated
      if (mark->has_monitor()) {
          // 已经是重量级锁状态了，直接返回
          ObjectMonitor * inf = mark->monitor() ;
          ...
          return inf ;
      }
      // CASE: inflation in progress
      if (mark == markOopDesc::INFLATING()) {
         // 正在膨胀中，说明另一个线程正在进行锁膨胀，continue重试
         TEVENT (Inflate: spin while INFLATING) ;
         // 在该方法中会进行spin/yield/park等操作完成自旋动作 
         ReadStableMark(object) ;
         continue ;
      }
      // 当前是轻量级锁，后面分析
      // CASE: stack-locked
          if (mark->has_locker()) {
        ...
      }
      // 无锁状态
      // CASE: neutral
      // 分配以及初始化ObjectMonitor对象
      ObjectMonitor * m = omAlloc (Self) ;
      // prepare m for installation - set monitor to initial state
      m->Recycle();
      m->set_header(mark);
      // owner为NULL
      m->set_owner(NULL);
      m->set_object(object);
      m->OwnerIsThread = 1 ;
      m->_recursions   = 0 ;
      m->_Responsible  = NULL ;
      m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;       // consider: keep metastats by type/class
        // 用CAS替换对象头的mark word为重量级锁状态
      if (Atomic::cmpxchg_ptr (markOopDesc::encode(m), object->mark_addr(), mark) != mark) {
          // 不成功说明有另外一个线程在执行inflate，释放monitor对象
          m->set_object (NULL) ;
          m->set_owner  (NULL) ;
          m->OwnerIsThread = 0 ;
          m->Recycle() ;
          omRelease (Self, m, true) ;
          m = NULL ;
          continue ;
          // interference - the markword changed - just retry.
          // The state-transitions are one-way, so there's no chance of
          // live-lock -- "Inflated" is an absorbing state.
      }

      ...
      return m ;
}

inflate其中是一个for循环，主要是为了处理多线程同时调用inflate的情况。然后会根据锁对象的状态进行不同的处理：

1.已经是重量级状态，说明膨胀已经完成，返回并继续执行ObjectMonitor::enter方法。
2.如果是轻量级锁则需要进行膨胀操作。
3.如果是膨胀中状态，则进行忙等待。
4.如果是无锁状态则需要进行膨胀操作。

轻量级锁膨胀流程如下：

if (mark->has_locker()) {  
  // 步骤1
  // 当前轻量级锁状态，先分配一个ObjectMonitor对象，并初始化值
  ObjectMonitor * m = omAlloc (Self) ;          
  m->Recycle();
  m->_Responsible  = NULL ;
  m->OwnerIsThread = 0 ;
  m->_recursions   = 0 ;
  m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;   // Consider: maintain by type/class
  // 步骤2
  // 将锁对象的mark word设置为INFLATING (0)状态 
  markOop cmp = (markOop) Atomic::cmpxchg_ptr (markOopDesc::INFLATING(), object->mark_addr(), mark) ;
  if (cmp != mark) {
    omRelease (Self, m, true) ;
    continue ;       // Interference -- just retry
  }
  // 步骤3
  // 栈中的displaced mark word
  markOop dmw = mark->displaced_mark_helper() ;
  assert (dmw->is_neutral(), "invariant") ;
  // 设置monitor的字段
  m->set_header(dmw) ;
  // owner为Lock Record
  m->set_owner(mark->locker());
  m->set_object(object);
  ...
  // 步骤4
  // 将锁对象头设置为重量级锁状态
  object->release_set_mark(markOopDesc::encode(m));
  ...
  return m ;
}

步骤 1、调用omAlloc获取一个可用的ObjectMonitor对象。在omAlloc方法中会先从线程私有的monitor集合omFreeList中分配对象，如果omFreeList中已经没有monitor对象，则从JVM全局的gFreeList中分配一批monitor到omFreeList中；

步骤 2、通过CAS尝试将Mark Word设置为markOopDesc:INFLATING，标识当前锁正在膨胀中。如果CAS失败，说明同一时刻其它线程已经将Mark Word设置为markOopDesc:INFLATING，当前线程进行自旋等待膨胀完成。

步骤 3、如果CAS成功，设置monitor的各个字段：设置monitor的header字段为displaced mark word，owner字段为Lock Record，obj字段为锁对象等；

步骤 4、设置锁对象头的mark word为重量级锁状态，指向第一步分配的monitor对象；

2.5.2 monitor竞争

当锁膨胀inflate执行完并返回对应的ObjectMonitor时，并不表示该线程竞争到了锁，真正的锁竞争发生在ObjectMonitor::enter方法中。

void ATTR ObjectMonitor::enter(TRAPS) {  
  Thread * const Self = THREAD ;
  void * cur ;
  // 步骤1
  // owner为null，如果能CAS设置成功，则当前线程直接获得锁
  cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
  if (cur == NULL) {
     ...
     return ;
  }
  // 如果是重入的情况
  if (cur == Self) {
     // TODO-FIXME: check for integer overflow!  BUGID 6557169.
     _recursions ++ ;
     return ;
  }
  // 步骤2
  // 如果当前线程是之前持有轻量级锁的线程
  // 上节轻量级锁膨胀将owner指向之前Lock Record的指针
  // 这里利用owner判断是否第一次进入。
  if (Self->is_lock_owned ((address)cur)) {
    assert (_recursions == 0, "internal state error");
    // 重入计数重置为1
    _recursions = 1 ;
    // 设置owner字段为当前线程
    _owner = Self ;
    OwnerIsThread = 1 ;
    return ;
  }
  ...
  // 步骤3
  // 在调用系统的同步操作之前，先尝试自旋获得锁
  if (Knob_SpinEarly && TrySpin (Self) > 0) {    
     ...
     //自旋的过程中获得了锁，则直接返回
     Self->_Stalled = 0 ;
     return ;
  }
  ...
  { 
    ...
    // 步骤4
    for (;;) {
      jt->set_suspend_equivalent();
      // 在该方法中调用系统同步操作
      EnterI (THREAD) ;
      ...
    }
    Self->set_current_pending_monitor(NULL); 
  }
  ...
}

步骤 1、当前是无锁、锁重入，简单操作后返回。

步骤 2、当前线程是之前持有轻量级锁的线程，则为首次进入，设置recursions为1，owner为当前线程，该线程成功获得锁并返回。

步骤 3、先自旋尝试获得锁，尽可能减少同步操作带来的开销。

步骤 4、调用EnterI方法。

这里注意，轻量级锁膨胀成功时，会把owner字段设置为Lock Record的指针，并在竞争时判断。这么做的原因是，假设当前线程A持有锁对象的锁，线程B进入同步代码块，并把锁对象升级为重量级锁。但此时，线程A可能还在执行，并无法感知其持有锁对象的变化。因此，需要线程B在执行ObjectMonitor::enter时，将自己放入到阻塞等列等待。并需要线程A第二次进入、或者退出的时候对monitor进行一些操作，以此保证代码块的同步。

这里有个自旋操作，直接看TrySpin对应的方法：

// TrySpin对应的方法
int ObjectMonitor::TrySpin_VaryDuration (Thread * Self) {  
    // Dumb, brutal spin.  Good for comparative measurements against adaptive spinning.
    int ctr = Knob_FixedSpin ;  // 固定自旋次数
    if (ctr != 0) {
        while (--ctr >= 0) {
            if (TryLock (Self) > 0) return 1 ;
            SpinPause () ;
        }
        return 0 ;
    }
    // 上一次自旋次数
    for (ctr = Knob_PreSpin + 1; --ctr >= 0 ; ) {
      if (TryLock(Self) > 0) {  // 尝试获取锁
        // Increase _SpinDuration ...
        // Note that we don't clamp SpinDuration precisely at SpinLimit.
        // Raising _SpurDuration to the poverty line is key.
        int x = _SpinDuration ;
        if (x < Knob_SpinLimit) {
           if (x < Knob_Poverty) x = Knob_Poverty ;
           _SpinDuration = x + Knob_BonusB ;
        }
        return 1 ;
      }
      ...
      ...

从方法名和注释可以看出，这就是自适应自旋，和网上说的轻量级锁cas失败会自旋的说法并不一致。实际上，无论是轻量级锁cas自旋还是重量级锁cas自旋，都是在用户态尽可能减少同步操作带来的开销，并没有太多本质上的区别。 到此为止，我们可以再结合上述的内容，整理出如下的状态转换图：

2.5.3 monitor等待

ObjectMonitor竞争失败的线程，通过自旋执行ObjectMonitor::EnterI方法等待锁的释放，EnterI方法的部分逻辑实现如下：

void ATTR ObjectMonitor::EnterI (TRAPS) {  
        // 尝试自旋
    if (TrySpin (Self) > 0) {
        ...
        return ;
    }
    ...
    // 将线程封装成node节点中
    ObjectWaiter node(Self) ;
    Self->_ParkEvent->reset() ;
    node._prev   = (ObjectWaiter *) 0xBAD ;
    node.TState  = ObjectWaiter::TS_CXQ ;
    // 将node节点插入到_cxq队列的头部，cxq是一个单向链表
    ObjectWaiter * nxt ;
    for (;;) {
        node._next = nxt = _cxq ;
        if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;
        // CAS失败的话 再尝试获得锁，这样可以降低插入到_cxq队列的频率
        if (TryLock (Self) > 0) {
            ...
            return ;
        }
    }
        ...
}

EnterI大致原理：一个ObjectMonitor对象包括两个同步队列（_cxq和_EntryList） ，以及一个等待队列_WaitSet。cxq、EntryList 、WaitSet都是由ObjectWaiter构成的链表结构。其中，_cxq为单向链表，_EntryList为双向链表。

当一个线程尝试获得重量级锁且没有竞争到时，该线程会被封装成一个ObjectWaiter对象插入到cxq的队列的队首，然后调用park函数挂起当前线程，进入BLOCKED状态。当线程释放锁时，会根据唤醒策略，从cxq或EntryList中挑选一个线程unpark唤醒。如果线程获得锁后调用Object#wait方法，则会将线程加入到WaitSet中，进入WAITING或TIMED_WAITING状态。当被Object#notify唤醒后，会将线程从WaitSet移动到cxq或EntryList中去，进入BLOCKED状态。需要注意的是，当调用一个锁对象的wait或notify方法时，若当前锁的状态是偏向锁或轻量级锁则会先膨胀成重量级锁。

2.5.4 monitor释放

当某个持有锁的线程执行完同步代码块时，会进行锁的释放。在HotSpot中，通过退出monitor的方式实现锁的释放，并通知被阻塞的线程，具体实现位于ObjectMonitor::exit方法中。

void ATTR ObjectMonitor::exit(bool not_suspended, TRAPS) {  
   Thread * Self = THREAD ;
   // 如果_owner不是当前线程
   if (THREAD != _owner) {
     // 轻量级锁膨胀上来，还没调用过enter方法，_owner还指向之前轻量级锁Lock Record的指针。
     if (THREAD->is_lock_owned((address) _owner)) {
       assert (_recursions == 0, "invariant") ;
       _owner = THREAD ;
       _recursions = 0 ;
       OwnerIsThread = 1 ;
     } else {
       // 异常情况:当前不是持有锁的线程
       TEVENT (Exit - Throw IMSX) ;
       assert(false, "Non-balanced monitor enter/exit!");
       if (false) {
          THROW(vmSymbols::java_lang_IllegalMonitorStateException());
       }
       return;
     }
   }
   // 重入计数器还不为0，则计数器-1后返回
   if (_recursions != 0) {
     _recursions--;        // this is simple recursive enter
     TEVENT (Inflated exit - recursive) ;
     return ;
   }
   ...
   //这块开始是唤醒操作
   for (;;) {
     ...
     ...
     ObjectWaiter * w = NULL ;
     // 根据QMode的不同会有不同的唤醒策略，默认为0
     int QMode = Knob_QMode ;
     if (QMode == 2 && _cxq != NULL) {
          ...
          ...

步骤 1、处理owner不是当前线程的状况。这里特指之前持有轻量级锁的线程，由于没有调用过enter，owner指向仍为Lock Record的指针，以及其他异常情况。

步骤 2、重入计数器还不为0，则计数器-1后返回。

步骤 3、唤醒操作。根据不同的策略（由QMode指定），从cxq或EntryList中获取头节点，通过ObjectMonitor::ExitEpilog方法唤醒该节点封装的线程，唤醒操作最终由unpark完成。

根据QMode的不同(默认为0)，有不同的处理方式：

QMode = 0：暂时什么都不做；
QMode = 2且cxq非空：取cxq队列队首的ObjectWaiter对象，调用ExitEpilog方法，该方法会唤醒ObjectWaiter对象的线程，然后立即返回，后面的代码不会执行了；
QMode = 3且cxq非空：把cxq队列插入到EntryList的尾部；
QMode = 4且cxq非空：把cxq队列插入到EntryList的头部；

只有QMode=2的时候会提前返回，等于0、3、4的时继续执行：

1.如果EntryList的首元素非空，就取出来调用ExitEpilog方法，该方法会唤醒ObjectWaiter对象的线程，然后立即返回；
2.如果EntryList的首元素为空，就将cxq的所有元素放入到EntryList中，然后再从EntryList中取出来队首元素执行ExitEpilog方法，然后立即返回；
3.被唤醒的线程，继续竞争monitor；

2.6 本章小节

本章介绍了Synchronized的底层实现和锁升级过程。对于锁升级，再看看本文整理的图，一图胜千言：

这里有几个点可以注意一下:

1.HotSpot中，只用到了模板解释器，并没有用到字节码解释器，monitorenter的实际入口位于templateTablex8664.cpp#3667。本文的分析是基于字节码解释器的，因此部分结论不能作为实际执行情况。本章的内容只能作为Synchronized锁升级原理、各类锁的适用场景的一种窥探。
2.再次强调，无锁状态只能升级为轻量级锁，匿名偏向状态才能进入到偏向锁。
3.偏向锁并不都有利，其适用于单个线程重入的场景，原因为：偏向锁的撤销需要进入safepoint，开销较大。需要进入safepoint是由于，偏向锁的撤销需要对锁对象的lock record进行操作，而lock record要到其他线程的栈帧中遍历寻找。在非safepoint，栈帧是动态的，会引入更多的问题。目前看来，偏向锁存在的价值是为历史遗留的Collection类如Vector和HashTable等做优化，迟早药丸。Java 15中默认不开启。
4.执行Object类的hashcode方法，偏向锁撤销并且锁会膨胀为轻量级锁或者重量锁。执行Object类的wait/notify/notifyall方法，偏向锁撤销并膨胀成重量级锁。
5.轻量级锁适用于两个线程的交替执行场景：线程A进入轻量级锁，退出同步代码块并释放锁，会将锁对象恢复为无锁状态；线程B再进入锁，发现为无锁状态，会cas尝试获取该锁对象的轻量级锁。如果有竞争，则直接膨胀为重量级锁，没有自旋操作，详情看10。
6.唤醒策略依赖于QMode。重量级锁获取失败后，线程会加入cxq队列。当线程释放锁时，会从cxq或EntryList中挑选一个线程唤醒。线程获得锁后调用Object#wait方法，则会将线程加入到WaitSet中。当被Object#notify唤醒后，会将线程从WaitSet移动到cxq或EntryList中去。
7.重量级锁，会将线程放进等待队列，等待操作系统调度。而偏向锁和轻量级锁，未交由操作系统调度，依然处于用户态，只是采用CAS无锁竞争的方式获取锁。CAS通过Unsafe类中compareAndSwap方法，jni调用C++方法，通过汇编指令锁住cpu中的北桥信号。
8.许多文章声称一个对象关联到一个monitor，这个说法不够准确。如果对象已经是重量级锁了，对象头的确指向了一个monitor。但对于正在膨胀的锁，会先从线程私有的monitor集合omFreeList中分配对象。如果omFreeList中已经没有monitor对象，再从JVM全局的gFreeList中分配一批monitor到omFreeList中。
9.在编译期间还有锁消除和锁粗化这两步锁优化操作，本章没做介绍。
10.字节码实现中没有体现轻量级锁自旋逻辑。这可能是模板解释器中的实现，或者是jvm在不同平台、不同jvm版本的不同实现。但本文分析的字节码链路中没有发现该逻辑，倒是发现了重量级锁会自适应自旋竞争锁。因此个人对轻量级锁自适应自旋的说法存疑，至少hotspot jdk8u字节码实现中没有这个逻辑。但两者都是在用户态进行自适应自旋，以尽可能减少同步操作带来的开销，没有太多本质上的区别，并不需要特别关心。

三、线程的实现与状态转换

3.1 线程的实现

（1）内核线程实现

内核线程（Kernel-Level Thread，KLT）：由内核来完成线程切换，内核通过调度器对线程进行调度，并负责将线程的任务映射到各个处理器上。 程序一般不会直接去使用内核线程，而是使用内核线程的一种高级接口——轻量级进程（Light Weight Process，LWP），也就是通常意义上的线程。

优点：每个LWP都是独立的调度单元。一个LWP被阻塞，不影响其他LWP。

缺点：基于KLT，耗资源。线程的创建、析构、同步都需要进行系统调用，频繁的用户态、内核态切换。

（2) 用户线程实现（User Thread，UT）

广义：非内核线程，都可认为是用户线程。（包括LWT，虽然LWT的大多操作都要映射到KLT）

狭义：完全建立在用户空间的线程库上，系统内核不能感知线程存在的实现。UT也只感知到掌管这些UT的进程P。

优点：用户线程的创建、同步、销毁和调度完全在用户态中完成，不需要内核的帮助。

缺点：线程的创建、销毁、切换和调度都是用户必须考虑到问题。“阻塞如何处理”、“多处理器系统中如何将线程映射到其他处理器上”这类问题解决起来将会异常困难。

（3) 混合实现 混合模式下，即存在用户线程，也存在轻量级进程。用户线程的创建、切换、析构等操作依然廉价，可以支持大规模的用户线程并发，且可以使用内核线程提供的线程调度功能及处理器映射。

线程的实现依赖操作系统支持的线程模型。在主流的操作系统上，hotspot、classic、art等虚拟机默认是 1:1的线程模型。在Solaris平台上，hotspot支持1:1、N：M两种线程模型。

3.2 线程的转换

首先明确一点，当我们讨论一个线程的状态，指的是Thread 类中threadStatus的值。

private volatile int threadStatus = 0;  

该值映射后对应的枚举为：

public enum State {  
    NEW,
    RUNNABLE,
    BLOCKED,
    WAITING,
    TIMED_WAITING,
    TERMINATED;
}

也就是说，线程的具体状态，看threadStatus就行了。

NEW

先要创建Thread 类的对象，才能谈其状态。

Thread t = new Thread();  

这个时候，线程t就处于新建状态。但他还不是“线程”。

RUNNABLE

然后调用start()方法。

t.start();  

调用start()后，会执行一个native方法创建内核线程，以linux为例：

private native void start0();

// 最后走到这
hotspot/src/os/linux/vm/os_linux.cpp  
pthread_create(...);  

这时候才有一个真正的线程创建出来，并即刻开始运行。这个内核线程与线程t进行1：1的映射。这时候t具备运行能力，进入RUNNABLE状态。 RUNNABLE可以细分为READY和RUNNING，两者的区别只是是否等待到了资源并开始运行。
处于RUNNABLE且未运行的线程，会进入一个就绪队列中，等待操作系统的调度。处于就绪队列的线程都在等待资源，这个资源可以是cpu的时间片、也可以是系统的IO。 JVM并不关系READY和RUNNING这两种状态，毕竟上述的枚举类都不对RUNNABLE进行细分。

TERMINATED

当一个线程执行完毕（或者调用已经不建议的 stop 方法），线程的状态就变为 TERMINATED。进入TERMINATED后，线程的状态不可逆，无法再复活。

关于BLOCKED、WAITING、TIMED_WAITING

BLOCKED、WAITING、TIMED_WAITING都是带有同步语义的状态，我们先看一下wait和notify方法的底层实现。
wait方法的底层实现:

void ObjectSynchronizer::wait(Handle obj, jlong millis, TRAPS) {  
    ...
    ...
  //获得Object的monitor对象
  ObjectMonitor* monitor = ObjectSynchronizer::inflate(THREAD, obj());
  DTRACE_MONITOR_WAIT_PROBE(monitor, obj(), THREAD, millis);
  //调用monitor的wait方法
  monitor->wait(millis, true, THREAD);
    ...
}

  inline void ObjectMonitor::AddWaiter(ObjectWaiter* node) {
    ...
  if (_WaitSet == NULL) {
    //_WaitSet为null，就初始化_waitSet
    _WaitSet = node;
    node->_prev = node;
    node->_next = node;
  } else {
    //否则就尾插
    ObjectWaiter* head = _WaitSet ;
    ObjectWaiter* tail = head->_prev;
    assert(tail->_next == head, "invariant check");
    tail->_next = node;
    head->_prev = node;
    node->_next = head;
    node->_prev = tail;
  }
}

主要流程：通过object获得objectMonitor，将Thread封装成OjectWaiter对象，然后addWaiter将它插入waitSet中，进入waiting或timed_waiting状态。最后释放锁，并通过底层的park方法挂起线程；

notify方法的底层实现：

void ObjectSynchronizer::notify(Handle obj, TRAPS) {  
    ...
    ...
    ObjectSynchronizer::inflate(THREAD, obj())->notify(THREAD);
}
    //通过inflate方法得到ObjectMonitor对象
    ObjectMonitor * ATTR ObjectSynchronizer::inflate (Thread * Self, oop object) {
    ...
     if (mark->has_monitor()) {
          ObjectMonitor * inf = mark->monitor() ;
          assert (inf->header()->is_neutral(), "invariant");
          assert (inf->object() == object, "invariant") ;
          assert (ObjectSynchronizer::verify_objmon_isinpool(inf), "monitor is inva;lid");
          return inf 
      }
    ...
      }
    //调用ObjectMonitor的notify方法
    void ObjectMonitor::notify(TRAPS) {
    ...
    //调用DequeueWaiter方法移出_waiterSet第一个结点
    ObjectWaiter * iterator = DequeueWaiter() ;
    //将上面DequeueWaiter尾插入_EntrySet或cxq等操作
    ...
    ...
  }

通过object获得objectMonitor，调用objectMonitor的notify方法。这个notify最后会走到ObjectMonitor::DequeueWaiter方法，获取waitSet列表中的第一个ObjectWaiter节点。并根据不同的策略，将取出来的ObjectWaiter节点，加入到EntryList或cxq中。 notifyAll的实现类似于notify，主要差别在多了个for循环。

由这里以及上一章2.5.4 monitor释放小节中可以了解到，notify和notifyAll并不会立即释放所占有的ObjectMonitor对象，其真正释放ObjectMonitor的时间点是在执行monitorexit指令。

一旦释放ObjectMonitor对象了，entryList和cxq中的ObjectWaiter节点会依据QMode所配置的策略，通过ExitEpilog方法唤醒取出来的ObjectWaiter节点。被唤醒的线程，继续参与monitor的竞争。若竞争失败，重新进入BLOCKED状态，再回顾一下monitor的核心结构。

既然聊到了wait和notify，那顺便也看下join、sleep和park。

打开 Thread.join() 的源码：

public final synchronized void join(long millis) throws InterruptedException {  
    ...
  if (millis == 0) {
    while (isAlive()) {
      wait(0);
    }
  } else {
    while (isAlive()) {
      long delay = millis - now;
      if (delay <= 0) {
        break;
      }
      wait(delay);
      now = System.currentTimeMillis() - base;
    }
  }
}

join的本质仍然是 wait() 方法。在使用join时，JVM会帮我们隐式调用notify，因此我们不需要主动notify唤醒主线程(可见析构函数synchronizer.cpp#368)。 而sleep()方法最终是调用SleepEvent对象的park方法：

int os::sleep(Thread* thread, jlong millis, bool interruptible) {  
  //获取thread中的_SleepEvent对象
  ParkEvent * const slp = thread->_SleepEvent ;
  ...
  //如果是允许被打断
  if (interruptible) {
    //记录下当前时间戳，这是时间比较的基准
    jlong prevtime = javaTimeNanos();
    for (;;) {
      //检查打断标记，如果打断标记为true，则直接返回
      if (os::is_interrupted(thread, true)) {
        return OS_INTRPT;
      }
      //线程被唤醒后的当前时间戳
      jlong newtime = javaTimeNanos();
      //睡眠毫秒数减去当前已经经过的毫秒数
      millis -= (newtime - prevtime) / NANOSECS_PER_MILLISEC;
      //如果小于0，那么说明已经睡眠了足够多的时间，直接返回
      if (millis <= 0) {
        return OS_OK;
      }
      //更新基准时间
      prevtime = newtime;
      //调用_SleepEvent对象的park方法，阻塞线程
      slp->park(millis);
    }
  } else {
    //如果不能打断，除了不再返回OS_INTRPT以外，逻辑是完全相同的
    for (;;) {
      ...
      slp->park(millis);
      ...
    }
    return OS_OK ;
  }
}

Thread.sleep在jvm层面上是调用thread中SleepEvent对象的park()方法实现阻塞线程，在此过程中会通过判断时间戳来决定线程的睡眠时间是否达到了指定的毫秒。看到这里，对于sleep和wait的区别应该会有更深入的理解。

park、unpark方法也与同步语义无关。每个线程都与一个许可(permit)关联。unpark函数为线程提供permit，线程调用park函数则等待并消耗permit。park和unpark方法具体实现比较复杂，这里不展开。 到此为止，我们可以整理出如下的线程状态转换图。

3.3 本章小节

Java 将OS经典五种状态中的ready和running，统一为 RUNNABLE。将WAITING（即不可能得到 CPU 运行机会的状态）细分为了 BLOCKED、WAITING、TIMED_WAITING。本章的内容较为简短，因为部分的内容已囊括在第一章中。

这里提个会使人困惑的问题： 使用socket时，调用accept()，read() 等阻塞方法时，线程处于什么状态？

答案是java线程处于RUNNABLE状态，OS线程处于WAITING状态。因为在jvm层面，等待cpu时间片和等待io资源是等价的。

这里有几个点可以注意一下：

1.JVM线程状态不代表内核线程状态。
2.BLOCKED的线程一定处于entryList或cxq中，而处于WAITING和TIMED WAITING的线程，可能是由于执行了sleep或park进入该状态，不一定在waitSet中。也就是说，处于BLOCKED状态的线程一定是与同步相关。由这可延伸出，调用 jdk 的 lock并获取不到锁的线程，进入的是 WAITING 或 TIMED_WAITING 状态，而不是BLOCKED状态。

四、相关资料

本文主要参考资料：

死磕Synchronized底层实现--概论

死磕Synchronized底层实现--偏向锁

死磕Synchronized底层实现--轻量级锁

死磕Synchronized底层实现--重量级锁

​ 随着有赞实时计算业务场景全部以 Flink SQL 的方式接入，对有赞现有的引擎版本——Flink 1.10 的 SQL 能力提出了越来越多无法满足的需求以及可以优化的功能点。目前有赞的 Flink SQL 是在 Yarn 上运行，但是在公司应用容器化的背景下，可以统一使用公司 K8S 资源池，同时考虑到任务之间的隔离性以及任务的弹性调度，Flink SQL 任务 K8S 化是必须进行的，所以我们也希望通过这次升级直接利社区的 on K8S 能力，直接将Flink SQL 集群迁移到 K8S 上。特别是社区在 Flink 1.13 中 on Native K8S 能力的支持完善，为了紧跟社区同时提升有赞实时计算引擎的能力，经过一些列调研，我们决定将有赞实时计算引擎由 Flink 1.10 升级到 Flink 1.13.2。

二、有赞业务场景下的升级到 Flink 1.13 收益评估

​ 社区在发布 Flink 1.13 后相比于 Flink 1.10 有了很多的新特性和优化，有些新特性在有赞场景下可能并未用到，所以接下来将主要从以下几个方面介绍一下在有赞业务场景下升级到 Flink 1.13 的一些收益。

2.1 Flink SQL 相关收益

​ 由于目前几乎所有的实时计算任务都通过 Flink SQL 方式实现，所以升级后关于 Flink SQL 上的一些优化是我们十分关注的，其中下面几点在升级后在有赞的实时计算业务场景下有很大的收益的：

（1） Flink SQL 语法更为简洁，提高开发效率

​ Flink 1.10 之后，社区提出了新的 connector 属性 key，SQL开发更为简洁，可以提升实时用户的开发作业效率。

（2）时区和时间函数相关优化

​ 由于 Flink 1.10 的时间函数在时区问题的不完善，用户在使用 currenttimestamp 和 currentday 等函数时由于时区问题需要额外的转换。而在 Flink 1.13 中对时区和时间函数进行纠正和优化，包括：

• 相关时间函数考虑了时区问题

  • CURRENTTIMESTAMP/CURRENTTIME/CURRENT_DATE/NOW() 在 Flink 1.13 中考虑了时区问题，且为本地时区。
  • PROCTIME() 考虑了时区问题，且为本地时区。

  

• 支持了TIMESTAMP_LTZ类型

  例如 CURRENTTIMESTAMP 函数返回值为 TIMESTAMPLTZ 类型，而不是 TIMESTAMP 类型。

  

• 夏令时支持 Flink 支持在 TIMESTAMPLTZ 列上定义时间属性，Flink SQL 在window 处理时结合 TIMESTAMP 和 TIMESTAMPLTZ, 优雅地支持了夏令时。

（3）支持 Window TVF 语法标准化

​ 在官方的介绍中，关于 Window TVF 包含四部分内容：Window TVF 语法，近实时累计计算，Window 性能优化，多维数据分析。 ​ 其中 Window TVF 语法在Flink 1.13 中用 Table-Valued Function 进行了语法标准化，在新的语法中支持 TUMBLE 和 HOP 窗口，我们通过以下两个例子来展示这一特性在某些场景下的应用：

• 用户在 table-valued 窗口函数中可以访问窗口的起始和终止时间，从而使用户可以实现新的功能。例如，除了常规的基于窗口的聚合和 Join 之外，用户现在也可以实现基于窗口的 Top-K 聚合：

SELECT *  
  FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
    FROM (
      SELECT window_start, window_end, supplier_id, SUM(price) as price, COUNT(*) as cnt
      FROM TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
      GROUP BY window_start, window_end, supplier_id
    )
  ) WHERE rownum <= 3;

• 新增了CUMULATE WINDOW 窗口，它可以支持按特定步长扩展的窗口，直到达到最大窗口大小，例如计算一段时间内的 PV, UV 等指标：

  

  SELECT window_time, window_start, window_end, SUM(price) AS total_price 
    FROM TABLE(CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES))
  GROUP BY window_start, window_end, window_time;

• window1 统计的是第一个区间的数据；
• window2 统计的是第一区间和第二个区间的数据；
• window3 统计的是第一区间，第二个区间和第三个区间的数据。

​ 累积窗口可以对迟到的数据进行处理，比如某一个数据是 window1 的迟到数据，无法被 window1 统计进去，但是触发 window2 时，会把 window1 迟到的数据统计进去，而且 window2 会复用 window1 的统计结果，而不是重新计算一遍。

（4）Flink On Hive 的能力

​ 目前在有赞已经开始有部分实时业务方希望 Flink 能够支持 Hive，比如 Flink-Hive 近实时的数仓中间层【小时表可更快产出】，以及 Flink 实时任务和离线数据对比功能。而在 Flink 1.12 中，已经支持生产级别 Flink On Hive 任务运行（社区 Commitor 说），所以基于这次 Flink 1.13 引擎版本升级，能够支持 Flink on hive 生产功能。因此本次升级可以解决部分实时业务方， Flink On Hive 的业务需求，下面是 Flink 1.13具体 Hive 相关功能：

• 支持 Hive 的 SQL 语法，支持常用的 Hive DML，DQL语法。
• Hive 写入：FLIP-115 完善扩展了 FileSystem connector 的基础能力和实现，Table/SQL 层的 sink 可以支持各种格式（CSV、Json、Avro、Parquet、ORC），而且支持 Hive table 的所有格式。
• FLIP-123 通过 Hive Dialect 为用户提供语法兼容，可以直接迁移 Hive 脚本到 Flink 中执行等等。

（5）其他功能及需求

• 支持查看 SQL 的执行计划
• 解决复杂 Quary 语句无法推断出 primary key 的问题
• SQL Connectors 中的 Metadata 处理。如果可以将某些 source（和 format）的元数据作为额外字段暴露给用户，对于需要将元数据与记录数据一起处理的用户来说很有意义。一个常见的例子是 Kafka，用户可能需要访问 offset、partition 或 topic 信息、读写 kafka 消息中的 key 或 使用消息 metadata 中的时间戳进行时间相关的操作。

CREATEF TABLE table (  
id BIGINT,  
name STRING,  
event_time TIMESTAMP(3) METADATA FROM 'timestamp', -- access Kafka 'timestamp' metadata  
headers MAP METADATA -- access Kafka 'headers' metadata  
) WITH (
'connector' = 'kafka',  
'topic' = 'test-topic',  
'format' = 'avro'  
);

2.2 Flink on K8S 相关收益

​ 在 on K8S 层面考虑升级到 Flink 1.13 主要有以下几个方面收益：

（1） Flink 1.13 on K8S 更成熟稳定

​ 相比于Flink 1.11 和 Flink 1.12，在Flink 1.13 版本中 on K8S 模式上更加丰富，更为成熟稳定。而且社区后续肯定是在 Native K8S 或者 Application Level K8S 上面发力。目前 社区在Flink 1.13 中关于 K8S 已经有了下面一些优化和新特性：

• 基于 Kubernetes 的高可用 (HA) 方案 Flink 可以利用 Kubernetes 提供的内置功能来实现 JobManager 的 failover，而不用依赖 ZooKeeper。为了实现不依赖于 ZooKeeper 的高可用方案，社区在 Flink 1.12（FLIP-144）中实现了基于 Kubernetes 的高可用方案。

• 引入 Application 模式 按照 application 粒度来启动一个集群，属于这个 application 的所有 job 在这个集群中运行。核心是 Job Graph 的生成以及作业的提交不在客户端执行，而是转移到 JM 端执行，这样网络下载上传的负载也会分散到集群中，不再有上述 client 单点上的瓶颈。

（2）实时离线弹性扩缩容

​ 目前有赞的离线任务已经实现了较好的弹性扩缩容，当 Flink SQL 任务 K8S 化之后，可以和离线任务之间实现更好的弹性扩缩容，节省集群资源成本，这是十分有意义的。

2.3 状态保留和恢复相关收益

（1）基于Savepoint跨集群迁移的能力

​ 在 Flink 1.10 版本中，Savepoint 中 meta 数据和 state 数据存放的是绝对路径，这就造成了不能进行集群迁移，否则会造成任务状态丢失。而在 Flink 1.10 以后 savepoint 中 meta 数据和 state 数据保存在同一目录，方便整体转移和复用；把 state 引用改成了相对路径，这样即使迁移后路径发生变化依然可用。

（2）生产可用的 Unaligned Checkpoint

​ 用户现在使用 Unaligned Checkpoint 时也可以扩缩容应用。如果用户需要因为性能原因不能使用 Savepoint 而必须使用 Retained checkpoint 时，这一功能会非常方便。 ​
收益：这一特性极大提升我们 checkpoint 的性能，同时也优化了在反压场景下 checkpoint超时失败的问题，解决目前一些大状态任务经常 checkpoint 超时的问题。同时也符合我们利用 checkpoint 来做重启状态恢复的场景。

（3）优化失败 Checkpoint 的异常和失败原因的汇报

​ Flink 1.13 现在提供了失败或被取消的 Checkpoint 的统计，从而使用户可以更简单的判断 Checkpoint 失败的原因，而不需要去查看日志。Flink 1.13 之前的版本只有在 Checkpoint 成功的时候才会汇报指标（例如持久化数据的大小、触发时间等）。

2.4 支持 upsert kafka 和 更丰富 Format 格式

（1）支持 upsert kafka

​ 在 Flink 1.12 中支持了 Upsert kafka，这一特性在有赞的实时计算业务场景中可以在某些数据链路中保障数据一致性。对于公司现有的一些场景，Upsert-kafka可以解决一些典型场景的数据重复问题：比如下图展示的在有赞很常见的一条实时链路，上游数据可以能是 MySQL binlog 或者NSQ -> Kafka 进行数据同步，然后下游对 Kafka 数据进行按照 key 聚合，将聚合数据存到 mysql , tidb 等等。这是很容易产生的问题就是在中间环节写入 Kafka 时很可能因为容错恢复等一些原因造成数据重复，特别是在 checkpoint 时间比较大时，造成的重复的数据量会很大，在现有的解决方案中，往往需要业务方在写入Kafka 时进行幂等操作，比如存入ZanKV等方式进行幂等。但是现有的方式问题就是现在的幂等方式性能有限，同时不能做到完全幂等。

​ 而接入 Upsert Kafka 连接器支持以 upsert 方式从 Kafka topic 中读取数据并将数据写入 Kafka topic。 作为 source，upsert-kafka 连接器生产 changelog 流，其中每条数据记录代表一个更新或删除事件。更准确地说，数据记录中的 value 被解释为同一 key 的最后一个 value 的 UPDATE，如果有这个 key（如果不存在相应的 key，则该更新被视为 INSERT）。

​ 作为 sink，upsert-kafka 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中，实现像 Hbase 一样的幂等写入。

（2）支持更丰富的 Format 格式

​ 在 Flink1.10 版本中对 Source 和 Sink 的 Format 支持是有限的，这也造成了我们业务方有些任务需要 Source 段支持更多的格式，比如 Kafka 支持 Raw、本地调试功能中Filesystem 需要支持 Json 等，这在 Flink 1.10 版本中是无法做到的。但是如果升级到 Flink 1.13 则可以完美解决这些问题。

2.5 其他相关收益

（1）查看JM和TM的内存相关指标

​ Flink 1.12 在 WebUI 上暴露了 JobManager 内存相关的指标和配置参数（FLIP-104）。对于 TaskManager 的指标页面也进行了更新，为 Managed Memory、Network Memory 和 Metaspace 添加了新的指标，以反映自 Flink 1.10（FLIP-102）开始引入的 TaskManager 内存模型的更改。

（2）WebUI界面查看被压情况

​ Flink 1.13 带来了一个改进的背压度量系统(使用任务邮箱计时而不是线程堆栈采样)，以及一个重新设计的作业数据流图形表示，用颜色编码和繁忙度和背压比率表示。

（3）CPU 火焰图查看

​ 可以直观的看 CPU 的火焰图来确定以下指标：当前哪些方法在消耗 CPU 的资源、各个方法消耗的 CPU 的资源的多少对比、堆栈上的哪些调用会导致执行特定的方法。

（4）Web UI 支持历史异常

​ Flink Web UI 现在可以展示导致作业失败的 n 次历史异常，从而提升在一个异常导致多个后续异常的场景下的调试体验。用户可以在异常历史中找到根异常。

三、Flink 1.13 升级过程实践与踩坑

​ 实时计算平台 Flink 引擎从 Flink 1.10 升级到 Flink 1.13 的主要工作将主要集中在自定义 connector 的升级、SQL 语法升级转换、任务迁移验证等几个方面的实践和踩坑来介绍此次升级过程。

3.1 自定义 connector 升级

​ 目前有赞的实时计算平台的数据流如下图所示，包括有赞自研的 NSQ、Kafka、Mysql、TiDB、Clickhouse、Habse 等大数据组件。那么此次升级需要将一些官方没有提供以及一些已经定制化的 connector 升级，其中包括 NSQ connector，定制的无用户名密码的 jdbc connector, clickhouse connector, 定制的高可用 hbase connector等。

​ 本次升级 connector 的主要工作是在 Flink 1.10 中 DataStream 和 Table connector 都统一是用到的是 Row 这种数据结构。而Flink 1.11 在 FLIP-95 对 TableSource 和 TableSink API 进行了重构，新增了 Flink SQL 内部数据结构 RowData, 在一些场景的序列化有一定的提升。为此，我们需要对上述四种定制或者自定义的 table connector 进行升级重构，对于无用户名密码的 jdbc connector 的链接方式采用的是连接池构建链接的方式，但是采用链接池的方式构建链接时，如果对于 Flink 任务长时间没有数据流入则链接会被释放掉，如果再次过来数据用原来的链接去写入数据时会抛出链接被关闭的异常，导致任务出现频繁的重启：

​ 为解决上述问题，需要在 flush 前检查链接是否有效，如果连接失效需要重新构建链接：

3.2 UDF 兼容

​ 在 Flink 1.10 版本有赞实时计算平台根据业务需求提供了很多通用的 UDF, 如 Dubbo 调用，JSON 转换，动态过滤条件。同时用户也自定义了一部分 UDF。所以在升级的过程中需要保证 UDF 的兼容性。好在 Flink 本身对 UDF 做了良好的兼容性，我们只需要将 maven 中 flink-table-common 改成对应的 Flink 版本即可。其中要注意的一点是，在 Flink 1.13 版本中如果 UDF 的参数是 Object 需要加上注解 @DataTypeHint(inputGroup = InputGroup.ANY) 帮助 Flink 做类型推荐。

3.3 SQL 语法转换实践

​ 在 Flink 1.13 SQL 用法中相比于 Flink 1.10 的 SQL 用法主要有以下几部分存在差异：建表语句的配置项简化 、时间函数的优化导致类型不匹配、存在 upsert 操作的的建表语句中需要指定 primary key。为保证任务可以平滑的从 Flink 1.10 升级到 Flink 1.13，我们对目前集群已有的数百个 Flink 1.10 语法的 SQL 任务进行转换，自动生成 Flink 1.13 版本的语法。`

（1）建表语句的配置项转换

​ 在Flink 1.13 中社区提出了新的 connector 属性 key，SQL 开发更为简洁，如下图分别展示了 Kafka 作为数据源时在 Flink 1.10 语法中的 connector 属性配置以及转换后在 Flink 1.13 语法中的属性配置。

​ 从上图的对比可以看出 Flink 1.13 语法中的 connector 属性配置 相比于 Flink 1.10 语法更为简洁易懂。虽然 Flink 本身对老版本的 SQL connector 的配置依然兼容，但是为了让用户使用新版的语法，我们对 用户在 Flink 1.10 的任务 SQL 进行配置了转换。

​ 值得注意的是:在一些 connector 的属性配置中，一些属性的 key 进行了改变，以 Kafka connector 为例，其中在 Flink 1.10 中 format.fail-on-not-json-record = false 要对应 Flink 1.13 中的 json.ignore-parse-errors = true 表示的是按照 JSON 格式解析数据失败则跳过。同样 connector.startup-mode = earliest-offset 和 scan.startup.mode = earliest-offset 都表示从 consumer 的最早的点位开始消费，但是配置的 key 已经改变了，这是大家在做新老版本语法转换需要注意的事情。

（2）时间函数类型逻辑转换及时间数据类型转换

在 Flink 1.13 中对一些时间函数进行了优化正如上一章的第一节所介绍的，那么在现有的 Flink 1.10 SQL 业务中，有些用户用到了相关的时间函数比如最常见的 currenttimestamp 函数，那么我们要对任务进行平滑升级时需要对使用 currenttimestamp 等时间函数进行相应的逻辑转换，主要是时区变更的转化和类型不匹配的转换。

• 时间函数时区逻辑转换以 currenttimestamp 函数为例，在 Flink 1.10 版本中 currenttimestamp 未考虑时区是 UTC+0 时间，而升级 Flink 1.13 之后 current_timestamp 考虑时区，且是本地时区时间。因此在之前的任务中，有些任务为了解决时区问题在任务中加了8小时或者减了16小时（前一天时间）。那么针对已经进行了时区转换的任务，我们需要将对应的 8 小时 时差减去，因此关于这一点我们对 SQL 任务进行匹配分析，对已经做了时区转换的任务逻辑减去 8 小时的时差。
• 时间函数类型转换还是以 currenttimestamp 函数为例，在 Flink 1.10 版本中 currenttimestamp 返回值类型为 timestamp 而在 Flink 1.13 中 currenttimestamp 返回值类型为 timestampltz 的格式。那么在 Flink 1.10 中的 current_timestamp 一些函数使用在 Flink 1.13中会报错，比如 TIMESTAMPDIFF 和 TIMESTAMPADD。简单举个例子

TIMESTAMPDIFF(MINUTE, (current_timestamp - INTERVAL '10' MINUTE), TO_TIMESTAMP(FROM_UNIXTIME(orderTime / 1000, 'yyyy-MM-dd HH:mm:ss')))  

​ 上述语句在 Flink 1.13 中会因为 TIMESTAMPDIFF 函数中一个是 timestampltz 格式 一个是 timestamp 而出现异常，为此需要转换成同一种类型，比如将后面的时间转为 timestampltz 类型，才能应用 TIMESTAMPDIFF、TIMESTAMPADD 等函数。

TIMESTAMPDIFF(MINUTE, (current_timestamp - INTERVAL '10' MINUTE), TO_TIMESTAMP_LTZ(mainOrderInfo.orderTime , 3))  

​ 如果需要将 CURRENTTIMESTAMP 的 TIMESTAMPLTZ 类型转为 TIMESTAMP 类型，可以使用下面的方式进行转换：

TO_TIMESTAMP(CAST(CURRENT_TIMESTAMP AS STRING))  

​ 因此在升级到 Flink 1.13 中关于时间函数的使用转换是尤为需要注意的，否则会因为逻辑不对造成数据不准确，或者任务异常无法启动。

（3）Primary key 自动生成

​ 在 Flink 1.10 以后对于存在 upsert 操作时比如写 mysql，tidb 时出现了聚合等操作需要在建表语句中指定 primary key, 这也是为了解决在 Flink 1.10 中对于一个复杂的 SQL 语句无法通过优化器 从Quary 语句中自动推断出 primary key而产生异常的问题。因此，为了平滑升级，我们需要对 upsert 流的建表语句中指定 primary key，否则会提示异常:

please declare primary key for sink table when query contains update/delete record.

​ 我们采用优化器推断 Quary 语句推断的方式实现了一套 primary key 自动生成的逻辑，然后判断任务是为 upsert 流 来为需要添加 primary key 的建表语句自动生成对应的 primary key。当然对于一些过于复杂的 SQL 任务如果生成失败会进行提示，联系用户自己去手动添加 primary key, 我们的 primary key 生成逻辑满足 95%的任务的 primary key 的自动生成。

3.4 任务平滑迁移实践与踩坑

​ 在 Flink 1.10 SQL 任务升级到 Flink 1.13 版本的过程中，我们除了做了语法转换之外，还有批量按照 Flink 1.13 语法检查，数据准确性验证，批量重启等工作。整个工作过程如下流程图所示：

其中有几点需要关注的是：

• 在迁移之前我们对各种任务构建了测试任务，并在第二天将测试任务的数据与老版本的实时任务和离线任务进行数据准确性验证；
• 同时关于 SQL 转换后关于 current_timestamp 这种时间函数的逻辑转换以及 primary key 的自动生成，需要在 SQL 转换后让用户进行 double check，反正升级后数据不准产生问题。
• 任务迁移尽量选择流量较小的时间段，防止重启异常时对业务产生很大的数据延迟影响。同时按照任务优先级的高低，以及根据实时任务血缘确定任务的重启顺序，比如在有赞的实时计算任务中，我们会优先重启低优先级和数据链路中下游的任务，在保证任务升级重启稳定运行一段时间后再去重启高优先级的任务，反正一些未发现的异常对升级后的任务产生大的影响。

3.5 其他踩坑和注意

​ 关于本次有赞实时计算平台引擎升级到 Flink 1.13 过程中也遇到过一些问题和踩过一些坑，一些问题已经在对应的实践中提及过了，那么还有遇到其他的一些升级过程中遇到一些问题在这里可以分享一下：

（1）任务升级后从之前版本的 checkpoint 文件恢复失败

当我们升级 Flink 1.13 后的任务想通过之前的任务的 checkpoint 文件进行状态恢复时，会偶尔出现下面的异常：

​ 通过社区邮件和源码阅读发现根本原因是在 Flink 1.11之后 BaseRowSerializer 改名成 RowDataSerializer了，即使用 state-processor-API 也没办法处理当前不存在的类。目前关于这一个问题社区也没有专门去处理的 Jira。

​ 这种问题并不是所有的任务重启时从之前的状态文件恢复都会出现的，所以面对这种问题的比较好的办法就是升级重启的时间尽量选择在流量小的时间段，对于一些按天维度做聚合的任务最好在凌晨的时候重启，这样出现问题也不会对第二天的数据有很大的影响，同时对于恢复异常的任务做好数据重放的处理。

（2）Mysql 维表关联出现类型转换异常报错

在升级 Flink 1.13 过程中，我们发现有几个 mysql 维表关联的任务升级重启后抛出如下异常：

​ 在1.13中由于对 Table connector 数据类型统一为 RowData，在维表关联时如果业务方的 mysql 的字段类型定义为 BIGINT，当 mysql 中是 BIGINT UNSIGNED 时，如果用Flink 的BIGINT 去转成 mysql 的 BIGINT UNSIGNED 时会出现上述的报错。因为最终维表关联的数据要转换成RowData格式，所以不能将mysql 的 BIGINT UNSIGNED 与Flink 的 BIGINT 进行相互转换。

​ 为了解决上述问题，在 Flink 1.11 中提出的一个Jira : FLINK-18580 ,官方建议在Flink 的建维表时将 BIGINT 定义为 DECIMAL(20,0)。

（3）执行多条 insert 语句任务异常

​ 在 Flink 1.10 中我们底层真正执行 SQL 的是 executeSql() 方法，对于 Flink 1.10 版本去调用该方法不会出现任何异常，且每条 insert 语句均有输出。但是升级到 Flink 1.13 之后，如果依然采用 executeSql() 方法去执行一个任务内的多条 insert 语句时会出现问题，我们发现只有第一条 insert 语句是有结果的，同时集群上出现多个相同的 job 被提交。如下面例子所示：

insert into max_realtimet select guangBusinessId,st_hour,'orderCountHourAll',orderCountHourAll from order_hour_cnt_all_view;

insert into max_realtime select guangBusinessId,st_hour,'orderPaidAmountHourAll',orderPaidAmountHourAll from order_hour_cnt_all_view;

insert into max_realtime select guangBusinessId,st_hour,'orderPaidUserCountHourAll',orderPaidUserCountHourAll from order_hour_cnt_all_view;  

​ 在 Flink 1.10的任务中 print 的结果是正常的：

​ 但是在 Flink 1.13 中可以明显只有第一条 insert 语句的输出:

​ 通过官方文档的解释我们发现执在 Flink 1.13 版本中 executeSql() 方法每执行一条 insert 语句的会立即提交一个 Flink 作业，并返回一个与提交的作业相关联的 TableResult 实例。这也验证了为什么发现确实当启动一个多 insert 语句的任务时在集群会起来了多个 job。 ​ 为此需要采用 StatementSet 将 insert 语句添加到 StatementSet 中，最后执行 StatementSet.execute()，如下代码所示：

StatementSet statementSet = streamTableEnv.createStatementSet();  
sqls.forEach(sql -> {  
  if(isInsertSql(sql)){
    statementSet.addInsertSql(sql);
  }else{
    streamTableEnv.executeSql(sql);
  }
});
statementSet.execute();  

​ 上述是我们从 Flink 1.10 升级到 Flink 1.13 中间遇到的一些问题，因为在 Flink 1.10 以后社区的代码架构改动还是很大的，中间踩了一些坑，也遇到一些问题，其实好多问题在社区邮件和社区的 jira 里面都给出了好的解决方案，我们更多的介绍了实践过程中踩过的一些坑来分享。

四、总结

​ 目前有赞实时计算平台已经将 Flink 引擎从 Flink 1.10 升级到了 Flink 1.13，并将所有的 Flink SQL 任务平滑迁移升级到 Flink 1.13 版本中，并成功运行了近三个月。随着有赞更多的业务场景不断接入实时任务，目前 Flink SQL 任务接近整体实时任务体量的 60%，实时任务 SQL 化是我们的目标，因此升级到 Flink 1.13 后对于 Flink SQL 开发的简化以及特性增加与性能优化对我们来说是十分有价值的。

​ 同时随着实时集群任务体量的增大，对资源的管控以及弹性扩缩容的需求也越来越大。而社区在 Flink on K8S 的投入也在不断增加，后续肯定是在 Native K8S 或者 Application Level K8S 有更多的优化，为此升级 Flink 1.13 之后我们将所有 Flink SQL 任务全部迁移到 K8S 集群，采用 Flink on Native 的 Application 模式运行任务，实现整个集群容器化，为后续的实时任务弹性扩缩容做好准备，目前我们已经完成 Flink on Native 的 Application 模式任务的测试阶段。后面将紧跟 Flink 社区的发展，为有赞的更多业务场景提供更多实践的可能。