在最近的學習中，在GitHub上找了一些布局練習，我發現了我自己對布局超級不熟悉（很難受）。

在以前的學習CSS過程中，感覺使用CSS就記住各個屬性的屬性值以及作用就OK了，但是實際上呢？呵呵一笑。不說了，太傷心了，進入正題吧！

最近我使用float和margin布局，加深了我對這兩個一起使用的理解。（新生可以看一下，大神請忽略）

float屬性

float: left | right | none | inherit

當然最常用的還是前面兩個：向左浮動和向右浮動

浮動最主要的特點：脫標。

脫離標準流，處於更加高級的層面，影響父元素和後面元素的布局，這裏就不具體介紹了。

margin屬性

這裏主要講margin-left和margin-right

margin-left: 設置元素的左外邊距。
margin-right: 設置元素的右外邊距。

總的來說，這個兩個屬性的字面理解還是很容易的，但是越簡單的東西越不要小看。

重點

以下代碼：

html:

<div class="box">
    <div class="zi_box1">1</div>
    <div class="zi_box2">2</div>
    <div class="zi_box3">3</div>
    <div class="clear"></div>
</div>

CSS:

.box {
            background-color: #555555;
            width: 600px;
            height: 200px;
        }
        .zi_box1 {
            float: left;
            background-color: #c23232;
            width: 200px;
            height: 100px;
        }
        .zi_box2 {
            float: left;
            background-color: chartreuse;
            width: 200px;
            height: 100px;
        }

        .zi_box3 {
            float: left;
            background-color: blue;
            width: 200px;
            height: 100px;
        }
        .clear {
            clear: both;
        }

最後實現的效果圖：

三個子盒子充滿父盒子，因為但他們寬度可以在父盒子裏面撐開。

如果父盒子撐不開呢？

加大一個子盒子的寬度，序號為3的盒子

zi_box3 {
    width: 300px;
}

效果圖如下：

那麼第三個盒子則會另外起一行。

結合margin使用時

在第一代碼的基礎上，增加一個margin值

zi_box1 {
    margin-left: 20px;
}

這時候，由於三個盒子的寬度加上margin值大於父盒子的寬度，所以盒子3就會另起一行

反之，給盒子3設置一個外邊距的值，盒子1和盒子2不設置外邊距，是不是盒子3也會另外起一行呢？答案是肯定的，因為他們的寬度已經超過父盒子的值了。

實現三列布局

在不改變DOM的順序的情況下，使盒子3，盒子1，盒子2的順序呢？是不是就可以充分使用margin這個屬性了。最開始白痴的我（很少練習布局吧，大神就不要噴我了，我只是個菜鳥）

白痴代碼

.zi_box1 {
     margin-left: 200px;       
}
.zi_box2 {
     margin-left: 200px;       
}

.zi_box3 {
     margin0left: -400px;
}
//這裏很天真的想法，以為每個元素是單獨行動

這樣寫的效果圖：

我當時就傻了，這是什麼玩意。

但是在最後的摸索中，我知道原因了，最最最最重要的就是DOM的執行順序

造成這樣的原因就是：盒子1先解析，margin-left: 200px,那麼這樣盒子3也就去了第二行; 再盒子2解析，margin-left：200px，那麼盒子2也去了第二行，因為第一行已經有600px這麼寬的長度了。最後解析盒子3，margin-left：-400px,盒子向前移動400px,不就造成了這樣的效果圖嘛。

這樣想的，就是指考慮片面的，而不是全局的

實現三列布局的最終代碼

.zi_box1 {
     margin-left: 200px;       
}
.zi_box2 {
     margin-left: 0px;      
}

.zi_box3 {
     margin0left: -600px;
}

效果圖

可以簡單的這樣理解

盒子1向右移動200px，那麼盒子2和盒子3也會向右移動200px,具體的效果圖如下

那麼盒子3移動到前面去，是不是需要600px的距離啊（是不是很容易懂，嘻嘻），當然這隻是我的片面理解，也不完全是對的。
在這種思維模式下，還要注意一點：當超出的部分盒子還是會遵守float的規則的。

那麼float: right和margin-right是一樣的道理。

這是我的第一篇博客，寫的太菜，不要笑我喲。

喜歡我的話，點個關注吧！

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※別再煩惱如何寫文案,掌握八大原則!

※教你寫出一流的銷售文案?

※超省錢租車方案

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※產品缺大量曝光嗎?你需要的是一流包裝設計!

上一篇我們介紹了數據塑形，HATEOAS和內容協商，並在制器方法中完成了對應功能的添加；本章我們將介紹日誌和測試相關的概念，並添加對應的功能

一、全局日誌

在第一章介紹項目結構時，有提到.NET Core啟動時默認加載了日誌服務，且在appsetting.json文件配置了一些日誌的設置，根據設置的日誌等級的不同可以進行不同級別的信息的显示，但它無法做到輸出固定格式的log信息至本地磁盤或是數據庫，所以需要我們自己手動實現，而我們可以藉助日誌框架實現。

ps：在第7章節中我們記錄的是數據處理層方法調用的日誌信息，這裏記錄的則是ASP.NET Core WebAPI層級的日誌信息，兩者有所差異

1、引入日誌框架

.NET程序中常用的日誌框架有log4net，serilog 和Nlog，這裏我們使用Serilog來實現相關功能，在BlogSystem.Core層使用NuGet安裝Serilog.AspNetCore，同時還需要搜索Serilog.Skins安裝希望支持的功能，這裏我們希望添加對文件和控制台的輸出，所以選擇安裝的是Serilog.Skins.File和Serilog.Skins.Console

需要注意的是Serilog是不受appsetting.json的日誌設置影響的，且它可以根據命名空間重寫記錄級別。還有一點需要注意的是需要手動對Serilog對象進行資源的釋放，否則在系統運行期間，無法打開日誌文件。

2、系統添加

在BlogSystem.Core項目中添加一個Logs文件夾，並在Program類中進行Serilog對象的添加和使用，如下：

3、全局添加

1、這個時候其實系統已經使用Serilog替換了系統自帶的log對象，如下圖，Serilog會根據相關信息進行高亮显示：

2、這個時候問題就來了，我們怎麼才能進行全局的添加呢，總不能一個方法一個方法的添加吧？還記得之前我們介紹AOP時提到的過濾器Filter嗎?ASP.NET Core中一共有五類過濾器，分別是：

• 授權過濾器Authorization Filter：優先級最高，用於確定用戶是否獲得授權。如果請求未被授權，則授權過濾器會使管道短路；
• 資源過濾器Resource Filter：授權后運行，會在Authorization之後，Model Binding之前執行，可以實現類似緩存的功能；
• 方法過濾器Action Filter：在控制器的Action方法執行之前和之後被調用，可以更改傳遞給操作的參數或更改從操作返回的結果；
• 異常過濾器Exception Filter：當Action方法執行過程中出現了未處理的異常，將會進入這個過濾器進行統一處理；
• 結果過濾器Result Filter：執行操作結果之前和之後運行，僅在action方法成功執行后才運行；

過濾器的具體執行順序如下：

3、這裏我們可以藉助異常過濾器實現全局日誌功能的添加；在在BlogSystem.Core項目添加一個Filters文件夾，添加一個名為ExceptionFilter的類，繼承IExceptionFilter接口，這裡是參考老張的哲學的簡化版本，實現如下：

using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.AspNetCore.Mvc.Filters;
using Microsoft.Extensions.Logging;
using Serilog;
using System;

namespace BlogSystem.Core.Filters
{
    public class ExceptionsFilter : IExceptionFilter
    {
        private readonly ILogger<ExceptionsFilter> _logger;

        public ExceptionsFilter(ILogger<ExceptionsFilter> logger)
        {
            _logger = logger;
        }

        public void OnException(ExceptionContext context)
        {
            try
            {
                //錯誤信息
                var msg = context.Exception.Message;
                //錯誤堆棧信息
                var stackTraceMsg = context.Exception.StackTrace;
                //返回信息
                context.Result = new InternalServerErrorObjectResult(new { msg, stackTraceMsg });
                //記錄錯誤日誌
                _logger.LogError(WriteLog(context.Exception));
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
                throw;
            }
            finally
            {
                //記得釋放，否則運行時無法打開日誌文件
                Log.CloseAndFlush();
            }

        }

        //返回500錯誤
        public class InternalServerErrorObjectResult : ObjectResult
        {
            public InternalServerErrorObjectResult(object value) : base(value)
            {
                StatusCode = StatusCodes.Status500InternalServerError;
            }
        }

        //自定義格式內容
        public string WriteLog(Exception ex)
        {
            return $"【異常信息】：{ex.Message} \r\n 【異常類型】：{ex.GetType().Name} \r\n【堆棧調用】：{ex.StackTrace}";
        }
    }
}

4、在Startup類的ConfigureServices方法中進行異常處理過濾器的註冊，如下：

5、我們在控制器方法中拋出一個異常，分別查看效果如下，如果覺得信息太多，可調整日誌記錄級別：

二、系統測試

這裏我們從測試的類別出發，了解下測試相關的內容，並添加相關的測試（介紹內容大部分來自微軟官方文檔，為了更易理解，從個人習慣的角度進行了修改，如有形容不當之處，可在評論區指出）

1、測試說明及分類

1、自動測試是確保軟件應用程序按照作者期望執行操作的一種絕佳方式。軟件應用有多種類型的測試，包括單元測試、集成測試、Web測試、負載測試和其他測試。單元測試用於測試個人軟件的組件或方法，並不包括如數據庫、文件系統和網絡資源類的基礎結構測試。

當然我們可以使用編寫測試的最佳方法，如測試驅動開發(TDD)所指的先編寫單元測試，再編寫該單元測試要檢查的代碼，就好比先編寫書籍的大綱，再編寫書籍。其主要目的是為了幫助開發人員編寫更簡單，更具可讀性的高效代碼。兩者區別如下(來自Edison Zhou)

2、以深度（測試的細緻程度）和廣度（測試的覆蓋程度）區分， 測試分類如下(此處內容來自solenovex)：

Unit Test 單元測試：它可以測試一個類或者一個類的某個功能，但其覆蓋程度較低；

Integration Test 集成測試：它的細緻程度沒有單元測試高，但是有較好的覆蓋程度，它可以測試功能的組合，以及像數據庫或文件系統這樣的外部資源；

Subcutaneous Test 皮下測試 ：其作用區域為UI層的下一層，有較好的覆蓋程度，但是深度欠佳；

UI測試：直接從UI層進行測試，覆蓋程度很高，但是深度欠佳

3、在編寫單元測試時，盡量不要引入基礎結構依賴項，這些依賴項會降低測試速度，使測試更加脆弱，我們應當將其保留供集成測試使用。可以通過遵循显示依賴項原則和使用依賴項注入避免應用程序中的這些依賴項，還可以將單元測試保留在單獨的項目中與集成測試相分離，以確保單元測試項目沒有引用或依賴於基礎結構包。

總結下常用的單元測試和集成測試，單元測試會與外部資源隔離，以保證結果的一致性；而集成測試會依賴外部資源，且覆蓋面更廣。

2、測試的目的及特徵

1、為什麼需要測試？我們從以單元測試為例從4個方面進行說明：

• 時間人力成本：進行功能測試時，通常涉及打開應用程序，執行一系列需要遵循的步驟來驗證預期的行為，這意味着測試人員需要了解這些步驟或聯繫熟悉該步驟的人來獲取結果。對於細微的更改或者是較大的更改，都需要重複上述過程，而單元測試只需要按一下按鈕即可運行，無需測試人員了解整個系統，測試結果也取決於測試運行程序而非測試人員。
• 防止錯誤回歸：程序更改後有時會出現舊功能異常的問題，所以測試時不僅要測試新功能還要確保舊功能的正常運行。而單元測試可以確保在更改一行代碼后重新運行整套測試，確保新代碼不會破壞現有的功能。
• 可執行性：在給定某個輸入的情況下，特定方法的作用或行為可能不會很明顯。比如，輸入或傳遞空白字符串、null后，該方法會有怎樣的行為？而當我們使用一套命名正確的單元測試，並清楚的解釋給定的輸入和預期輸出，那麼它將可以驗證其有效性。
• 減少代碼耦合：當代碼緊密耦合時，會難以進行單元測試，所以以創建單元測試為目的時，會在一定程度上要求我們注意代碼的解耦

2、優質的測試需要符合哪些特徵，同樣以單元測試為例：

• 快速：成熟的項目會進行數千次的單元測試，所以應當花費非常少的時間來運行單元測試，一般來說在幾毫秒
• 獨立：單元測試應當是獨立的，可以單獨運行，不依賴文件系統或數據庫等外部因素
• 可重複：單元測試的結果應當保持一致，即運行期間不進行更改，返回的結果應該相同
• 自檢查：測試應當在沒有人工交互的情況下，自動檢測是否通過
• 及時：編寫單元測試不應該花費過多的時間，如果花費時間較長，應當考慮另外一種更易測試的設計

在具體的執行時，我們應當遵循一些最佳實踐規則，具體請參考微軟官方文檔單元測試最佳做法

3、xUnit框架介紹

常用的單元測試框架有MSTest、xUnit和NUnit，這裏我們以xUnit為例進行相關的說明

3.1、測試操作

首先我們要明確如何編寫測試代碼，一般來說，測試分為三個主要操作：

• Arrange：意為安排或準備，這裏可以根據需求進行對象的創建或相關的設置；
• Act：意為操作，這裏可以執行獲取生產代碼返回的結果或者是設置屬性；
• Assert：意為斷言，這裏可以用來判斷某些項是否按預期進行，即測試通過還是失敗

3.2、Assert類型

Assert時通常會對不同類型的返回值進行判斷，而在xUnit中是支持多種返回值類型的，常用的類型如下：

boolean：針對方法返回值為bool的結果，可以判斷結果是true或false

string：針對方法返回值為string的結果，可以判斷結果是否相等，是否以某字符串開頭或結尾，是否包含某些字符，並支持正則表達式

數值型：針對方法返回值為數值的結果，可以判斷數值是否相等，數值是否在某個區間內，數值是否為null或非null

Collection：針對方法返回值為集合的結果，可以針對集合內所有元素或至少一個元素判斷其是否包含某某字符，兩個集合是否相等

ObjectType：針對方法返回值為某種類型的情況，可以判斷是否為預期的類型，一個類是否繼承於另一個類，兩個類是否為同一實例

Raised event：針對事件是否執行的情況，可以判斷方法內部是否執行了預期的事件

3.3、常用特性

在xUnit中還有一些常用的特性，可作用於方法或類，如下：

[Fact]：用來標註該方法為測試方法

[Trait(“Name”,”Value”)]：用來對測試方法進行分組，支持標註多個不同的組名

[Fact(Skip=”忽略說明…”)]：用來修飾需要忽略測試的方法

3.4 、性能相關

在測試時我們應當注意性能上的問題，針對一個對象供多個方法使用的情況，我們可以使用共享上下文

• 針對一個對象供同一類中的多個方法使用時，可以將該對象提取出來，使用IClassFixture 對象將其注入到構造函數中
• 針對一個對象供多個測試類使用的情況，可以使用ICollectionFixture 對象和[CollectionDefinition(“…”)]定義該對象

需要注意在使用IClassFixture和ICollectionFixture對象時應當避免多個測試方法之間相互影響的情況

3.5、數據驅動測試

在進行測試方法時，通常我們會指定輸入值和輸出值，如希望多測試幾種情況，我們可以定義多個測試方法，但這顯然不是一個最佳的實現；在合理的情況下，我們可以將參數和數據分離，如何實現？

• 方法一：使用[Theory]替換[Fact]，將輸入輸出參數提取為方法參數，並使用多個[InlineData(“輸入參數”,”輸出參數)]來標註方法
• 方法二：使用[Theory]替換[Fact]，針對測試方法新增一個測試數據類，該類包含一個靜態屬性IEumerable<object[]>，將數據封裝為一個list后賦值給該屬性，並使用[MemberData(nameof(數據類的屬性),MemberType=typeof(數據類))]標註測試方法即可；
• 方法三：使用外部數據如數據庫數據/Excel數據/txt數據等，其實現原理與方法二相同，只是多了一個數據獲取封裝為list的步驟；
• 方法四：自定義一個Attribute，繼承自DataAttribute，實現其對應的方法，使用yield返回object類型的數組；使用時只需要在測試方法上方添加[Theory]和[自定義Attribute]即可

4、測試項目添加

4.1、添加測試項目

首先我們右鍵項目解決方案選擇添加一個項目，輸入選擇xUnit後進行添加，項目命名為BlogSystem.Core.Test，如下：

項目添加完成后我們需要添加對測試項目的引用，在解決方案中右擊依賴項選擇添加BlogSystem.Core；這裏我們預期對Controller進行測試，但後續有可能會添加其他項目的測試，所以我們建立一個Controller_Test文件夾保證項目結構相對清晰。

4.2、添加測試方法

在BlogSystem.Core.Test項目的Controller_Test文件夾下新建一個命名為UserController_Should的方法；在微軟的《單元測試的最佳做法》文檔中有提到，測試命名應該包括三個部分：①被測試方法的名稱②測試的方案③方案預期行為；實際使用時也可以對照測試的方法進行命名，這裏我們先不考慮最佳命名原則，僅對照測試方法進行命名，如下：

using Xunit;

namespace BlogSystem.Core.Test.Controller_Test
{
    public class UserController_Should
    {
        [Fact]
        public void Register_Test()
        {
            
        }
    }
}

4.3、方案選擇

1、在進行測試時，我們可以根據實際情況使用以下方案來進行測試：

• 方案一：直接new一個Controller對象，調用其Action方法直接進行測試；適用於Controller沒有其他依賴項的情況；
• 方案二：當有多個依賴項時，可以藉助工具來模擬實例化時的依賴項，如Moq就是一個很好的工具；當然這需要一定的學習成本；
• 方案三：模擬Http請求的方式來調用API進行測試；NuGet中的Microsoft.AspNetCore.TestHost就支持這類情況；
• 方案四：自定義方法實例化所有依賴項；將測試過程種需要用到的對象放到容器中並加載，其實現較為複雜；

這裏我們以測試UserController為例，其構造函數包含了接口服務實例和HttpContext對象實例，Action方法內部又有數據庫連接操作，從嚴格意義上來講測試這類方法已經脫離了單元測試的範疇，屬於集成測試，但這類測試一定程度上可以節省我們大量的重複勞動。這裏我們選擇方案三進行相關的測試。

2、如何使用TestHost對象？先來看看它的工作流程，首先它會創建一個IHostBuilder對象，並用它創建一個TestServer對象，TestServer對象可以創建HttpClient對象，該對象支持發送及響應請求，如下圖所示（來自solenovex）：

在嘗試使用該對象的過程中我們會發現一個問題，創建IHostBuilder對象時需要指明類似Startup的配置項，因為這裡是測試環境，所以實際上會與BlogSystem.Core中的配置類StartUp存在一定的差異，因而這裏我們需要為測試新建立一個Startup配置類。

4.4、方法實現

1、我們在測試項目中添加名為TestServerFixture 的類和名為TestStartup的類，TestServerFixture 用來創建HttpClient對象並做一些準備工作，TestStartup類為配置類。然後使用Nuget安裝Microsoft.AspNetCore.TestHost；TestServerFixture 和TestStartup實現如下：

using Autofac.Extensions.DependencyInjection;
using BlogSystem.Core.Helpers;
using BlogSystem.Model;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.TestHost;
using Microsoft.Extensions.Hosting;
using System;
using System.Net.Http;

namespace BlogSystem.Core.Test
{
    public static class TestServerFixture
    {
        public static IHostBuilder GetTestHost()
        {
            return Host.CreateDefaultBuilder()
           .UseServiceProviderFactory(new AutofacServiceProviderFactory())//使用autofac作為DI容器
           .ConfigureWebHostDefaults(webBuilder =>
           {
               webBuilder.UseTestServer()//建立TestServer——測試的關鍵
               .UseEnvironment("Development")
               .UseStartup<TestStartup>();
           });
        }

        //生成帶token的httpclient
        public static HttpClient GetTestClientWithToken(this IHost host)
        {
            var client = host.GetTestClient();
            client.DefaultRequestHeaders.Add("Authorization", $"Bearer {GenerateJwtToken()}");//把token加到Header中
            return client;
        }

        //生成JwtToken
        public static string GenerateJwtToken()
        {
            TokenModelJwt tokenModel = new TokenModelJwt { UserId = userData.Id, Level = userData.Level.ToString() };
            var token = JwtHelper.JwtEncrypt(tokenModel);
            return token;
        }

        //測試用戶的數據
        private static readonly User userData = new User
        {
            Account = "jordan",
            Id = new Guid("9CF2DAB5-B9DC-4910-98D8-CBB9D54E3D7B"),
            Level = Level.普通用戶
        };

    }
}

using Autofac;
using Autofac.Extras.DynamicProxy;
using BlogSystem.Common.Helpers;
using BlogSystem.Common.Helpers.SortHelper;
using BlogSystem.Core.AOP;
using BlogSystem.Core.Filters;
using BlogSystem.Core.Helpers;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc.Formatters;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.IdentityModel.Tokens;
using System;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;

namespace BlogSystem.Core.Test
{
    public class TestStartup
    {
        private readonly IConfiguration _configuration;

        public TestStartup(IConfiguration configuration)
        {
            _configuration = GetConfig(null);
            //傳遞Configuration對象
            JwtHelper.GetConfiguration(_configuration);
        }

        public void ConfigureServices(IServiceCollection services)
        {
            //控制器服務註冊
            services.AddControllers(setup =>
            {
                setup.ReturnHttpNotAcceptable = true;//開啟不存在請求格式則返回406狀態碼的選項
                var jsonOutputFormatter = setup.OutputFormatters.OfType<SystemTextJsonOutputFormatter>()?.FirstOrDefault();//不為空則繼續執行
                jsonOutputFormatter?.SupportedMediaTypes.Add("application/vnd.company.hateoas+json");
                setup.Filters.Add(typeof(ExceptionsFilter));//添加異常過濾器
            }).AddXmlDataContractSerializerFormatters()//開啟輸出輸入支持XML格式

            //jwt授權服務註冊
            services.AddAuthentication(options =>
            {
                options.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
                options.DefaultChallengeScheme = JwtBearerDefaults.AuthenticationScheme;
            }).AddJwtBearer(x =>
            {
                x.TokenValidationParameters = new TokenValidationParameters
                {
                    ValidateIssuerSigningKey = true, //驗證密鑰
                    IssuerSigningKey = new SymmetricSecurityKey(Encoding.ASCII.GetBytes(_configuration["JwtTokenManagement:secret"])),

                    ValidateIssuer = true, //驗證發行人
                    ValidIssuer = _configuration["JwtTokenManagement:issuer"],

                    ValidateAudience = true, //驗證訂閱人
                    ValidAudience = _configuration["JwtTokenManagement:audience"],

                    RequireExpirationTime = true, //驗證過期時間
                    ValidateLifetime = true, //驗證生命周期
                    ClockSkew = TimeSpan.Zero, //緩衝過期時間，即使配置了過期時間，也要考慮過期時間+緩衝時間
                };
            });

            //註冊HttpContext存取器服務
            services.AddHttpContextAccessor();

            //自定義判斷屬性隱射關係
            services.AddTransient<IPropertyMappingService, PropertyMappingService>();

            services.AddTransient<IPropertyCheckService, PropertyCheckService>();
        }

        //configureContainer訪問AutoFac容器生成器
        public void ConfigureContainer(ContainerBuilder builder)
        {
            //獲取程序集並註冊,採用每次請求都創建一個新的對象的模式
            var assemblyBll = Assembly.LoadFrom(Path.Combine(AppContext.BaseDirectory, "BlogSystem.BLL.dll"));
            var assemblyDal = Assembly.LoadFrom(Path.Combine(AppContext.BaseDirectory, "BlogSystem.DAL.dll"));

            builder.RegisterAssemblyTypes(assemblyDal).AsImplementedInterfaces().InstancePerDependency();

            //註冊攔截器
            builder.RegisterType<LogAop>();
            //對目標類型啟用動態代理，並注入自定義攔截器攔截BLL
            builder.RegisterAssemblyTypes(assemblyBll).AsImplementedInterfaces().InstancePerDependency()
           .EnableInterfaceInterceptors().InterceptedBy(typeof(LogAop));
        }

        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            else
            {
                app.UseExceptionHandler(builder =>
                {
                    builder.Run(async context =>
                    {
                        context.Response.StatusCode = 500;
                        await context.Response.WriteAsync("Unexpected Error!");
                    });
                });
            }

            app.UseRouting();

           //添加認證中間件
            app.UseAuthentication();

            //添加授權中間件
            app.UseAuthorization();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllers();
            });
        }

        private IConfiguration GetConfig(string environmentName)
        {
            var path = Microsoft.DotNet.PlatformAbstractions.ApplicationEnvironment.ApplicationBasePath;

            IConfigurationBuilder builder = new ConfigurationBuilder().SetBasePath(path)
               .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true);

            if (!string.IsNullOrWhiteSpace(environmentName))
            {
                builder = builder.AddJsonFile($"appsettings.{environmentName}.json", optional: true);
            }

            builder = builder.AddEnvironmentVariables();

            return builder.Build();
        }
    }
}

2、這裏對UserController中的註冊、登錄、獲取用戶信息方法進行測試，實際上這裏的斷言並不嚴謹，會產生什麼後果？請繼續往下看

using BlogSystem.Model.ViewModels;
using Microsoft.AspNetCore.Mvc;
using Microsoft.AspNetCore.TestHost;
using Microsoft.Extensions.Hosting;
using Newtonsoft.Json;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading.Tasks;
using Xunit;

namespace BlogSystem.Core.Test.Controller_Test
{
    public class UserController_Should
    {
        const string _mediaType = "application/json";
        readonly Encoding _encoding = Encoding.UTF8;


        /// <summary>
        /// 用戶註冊
        /// </summary>
        [Fact]
        public async Task Register_Test()
        {
            // 1、Arrange
            var data = new RegisterViewModel { Account = "test", Password = "123456", RequirePassword = "123456" };

            StringContent content = new StringContent(JsonConvert.SerializeObject(data), _encoding, _mediaType);

            using var host = await TestServerFixture.GetTestHost().StartAsync();//啟動TestServer

            // 2、Act
            var response = await host.GetTestClient().PostAsync($"http://localhost:5000/api/user/register", content);

            var result = await response.Content.ReadAsStringAsync();

            // 3、Assert
            Assert.DoesNotContain("用戶已存在", result);
        }

        /// <summary>
        /// 用戶登錄
        /// </summary>
        [Fact]
        public async Task Login_Test()
        {
            var data = new LoginViewModel { Account = "jordan", Password = "123456" };

            StringContent content = new StringContent(JsonConvert.SerializeObject(data), _encoding, _mediaType);

            var host = await TestServerFixture.GetTestHost().StartAsync();//啟動TestServer

            var response = await host.GetTestClientWithToken().PostAsync($"http://localhost:5000/api/user/Login", content);

            var result = await response.Content.ReadAsStringAsync();

            Assert.DoesNotContain("賬號或密碼錯誤！", result);
        }

        /// <summary>
        /// 獲取用戶信息
        /// </summary>
        [Fact]
        public async Task UserInfo_Test()
        {
            string id = "jordan";

            using var host = await TestServerFixture.GetTestHost().StartAsync();//啟動TestServer

            var client = host.GetTestClient();

            var response = await client.GetAsync($"http://localhost:5000/api/user/{id}");

            var result = response.StatusCode;

            Assert.True(Equals(HttpStatusCode.OK, result)|| Equals(HttpStatusCode.NotFound, result));
        }
    }
}

4.5、異常及解決

1、添加完上述的測試方法后，我們使用打開Visual Studio自帶的測試資源管理器，點擊運行所有測試，發現提示錯誤無法加載BLL？在原先的BlogSystem.Core的StartUp類中我們是加載BLL和DAL項目的dll來達到解耦的目的，所以做了一個將dll輸出到Core項目bin文件夾的動作，但是在測試項目的TestStarup類中，我們是無法加載到BLL和DAL的。我嘗試將BLL和DAL同時輸出到兩個路徑下，但未找到對應的方法，所以這裏我採用了最簡單的解決方法，測試項目添加了對DAL和BLL的引用。再次運行，如下圖，似乎成功了？？

2、我們在測試方法內部打上斷點，右擊測試方法，選擇調試測試，結果發現response參數為空，只應Assert不嚴謹導致看上去沒有問題；在各種查找后，我終於找到了解決辦法，在TestStarup類的ConfigureServices方法內部service.AddControllers方法最後加上這麼一句話即可解決 .AddApplicationPart(Assembly.LoadFrom(Path.Combine(AppContext.BaseDirectory, "BlogSystem.Core.dll")))

3、再次運行測試方法，成功！但是又發現了另外一個問題，這裏我們只是測試，但是數據庫中卻出現了我們測試添加的test賬號，如何解決？我們可以使用Microsoft.EntityFrameworkCore.InMemory庫 ，它支持使用內存數據庫進行測試，這裏暫未添加，有興趣的朋友可以自行研究。

本章完~

本人知識點有限，若文中有錯誤的地方請及時指正，方便大家更好的學習和交流。

本文部分內容參考了網絡上的視頻內容和文章，僅為學習和交流，視頻地址如下：

老張的哲學，系列教程一目錄：.netcore+vue 前後端分離

我想吃晚飯，ASP.NET Core搭建多層網站架構【12-xUnit單元測試之集成測試】

solenovex，使用 xUnit.NET 對 .NET Core 項目進行單元測試

solenovex，ASP.NET Core Web API 集成測試

微軟官方文檔，.NET Core 和 .NET Standard 中的單元測試

Edison Zhou，.NET單元測試的藝術

聲明

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※別再煩惱如何寫文案,掌握八大原則!

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※超省錢租車方案

※教你寫出一流的銷售文案?

※網頁設計最專業,超強功能平台可客製化

※產品缺大量曝光嗎?你需要的是一流包裝設計!

前言

在 Java 中通過鎖來控制多個線程對共享資源的訪問，使用 Java 編程語言開發的朋友都知道，可以通過 synchronized 關鍵字來實現鎖的功能，它可以隱式的獲取鎖，也就是說我們使用該關鍵字並不需要去關心鎖的獲取和釋放過程，但是在提供方便的同時也意味着其靈活性的下降。例如，有這樣的一個場景，先獲取鎖 A，然後再獲取鎖 B，當鎖 B 獲取到之後，釋放鎖 A 同時獲取鎖 C，當獲取鎖 C 后，再釋放鎖 B 同時獲取鎖 D，依次類推，像這種比較複雜的場景，使用 synchronized 關鍵字就比較難實現了。
在 Java SE 5 之後，新增加了 Lock 接口和一系列的實現類來提供和 synchronized 關鍵字一樣的功能，它需要我們显示的進行鎖的獲取和釋放，除此之外還提供了可響應中斷的鎖獲取操作以及超時獲取鎖等同步特性。JDK 中提供的 Lock 接口實現類大部分都是聚合一個同步器 AQS 的子類來實現多線程的訪問控制的，下面我們看看這個構建鎖和其它同步組件的基礎框架——隊列同步器 AQS（AbstractQueuedSynchronizer）。

AQS 基礎數據結構

同步隊列

隊列同步器 AQS（下文簡稱為同步器）主要是依賴於內部的一個 FIFO（first-in-first-out）雙向隊列來對同步狀態進行管理的，當線程獲取同步狀態失敗時，同步器會將當前線程和當前等待狀態等信息封裝成一個內部定義的節點 Node，然後將其加入隊列，同時阻塞當前線程；當同步狀態釋放時，會將同步隊列中首節點喚醒，讓其再次嘗試去獲取同步狀態。同步隊列的基本結構如下：

隊列節點 Node

同步隊列使用同步器中的靜態內部類 Node 用來保存獲取同步狀態的線程的引用、線程的等待狀態、前驅節點和後繼節點。

同步隊列中 Node 節點的屬性名稱和具體含義如下錶所示：

屬性類型和名稱	描述
volatile int waitStatus	當前節點在隊列中的等待狀態
volatile Node prev	前驅節點，當節點加入同步隊列時被賦值(使用尾部添加方式)
volatile Node next	後繼節點
volatile Thread thread	獲取同步狀態的線程
Node nextWaiter	等待隊列中的後繼節點，如果當前節點是共享的，則該字段是一個 SHARED 常量

每個節點線程都有兩種鎖模式，分別為 SHARED 表示線程以共享的模式等待鎖，EXCLUSIVE 表示線程以獨佔的方式等待鎖。同時每個節點的等待狀態 waitStatus 只能取以下錶中的枚舉值：

枚舉值	描述
SIGNAL	值為 -1，表示該節點的線程已經準備完畢，等待資源釋放
CANCELLED	值為 1，表示該節點線程獲取鎖的請求已經取消了
CONDITION	值為 -2，表示該節點線程等待在 Condition 上，等待被其它線程喚醒
PROPAGATE	值為 -3，表示下一次共享同步狀態獲取會無限進行下去，只在 SHARED 情況下使用
0	值為 0，初始狀態，初始化的默認值

同步狀態 state

同步器內部使用了一個名為 state 的 int 類型的變量表示同步狀態，同步器的主要使用方式是通過繼承，子類通過繼承並實現它的抽象方法來管理同步狀態，同步器給我們提供了如下三個方法來對同步狀態進行更改。

方法簽名	描述
protected final int getState()	獲取當前同步狀態
protected final void setState(int newState)	設置當前同步狀態
protected final boolean compareAndSetState(int expect, int update)	使用 CAS 設置當前狀態，該方法能夠保證狀態設置的原子性

在獨享鎖中同步狀態 state 這個值通常是 0 或者 1（如果是重入鎖的話 state 值就是重入的次數），在共享鎖中 state 就是持有鎖的數量。

獨佔式同步狀態獲取與釋放

同步器中提供了 acquire(int arg) 方法來進行獨佔式同步狀態的獲取，獲取到了同步狀態也就是獲取到了鎖，該方法源碼如下所示：

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

方法首先會調用 tryAcquire 方法嘗試去獲取鎖，查看方法的源碼可以發現，同步器並未對該方法進行實現（只是拋出一個不支持操作異常 UnsupportedOperationException），這個方法是需要後續同步組件的開發人員自己去實現的，如果方法返回 true 則表示當前線程成功獲取到鎖，調用 selfInterrupt() 中斷當前線程（PS：這裏留給大家一個問題：為什麼獲取了鎖以後還要中斷線程呢？），方法結束返回，如果方法返回 false 則表示當前線程獲取鎖失敗，也就是說有其它線程先前已經獲取到了鎖，此時就需要把當前線程以及等待狀態等信息添加到同步隊列中，下面來看看同步器在線程未獲取到鎖時具體是如何實現。
通過源碼發現，當獲取鎖失敗時，會執行判斷條件與操作的後半部分 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)，首先指定鎖模式為 Node.EXCLUSIVE 調用 addWaiter 方法，該方法源碼如下：

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

通過方法參數指定的鎖模式（共享鎖 or 獨佔鎖）和當前線程構造出一個 Node 節點，如果同步隊列已經初始化，那麼首先會進行一次從尾部加入隊列的嘗試，使用 compareAndSetTail 方法保證原子性，進入該方法源碼可以發現是基於 sun.misc 包下提供的 Unsafe 類來實現的。如果首次嘗試加入同步隊列失敗，會再次調用 enq 方法進行入隊操作，繼續跟進 enq 方法源碼如下：

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

通過其源碼可以發現和第一次嘗試加入隊列的代碼類似，只是該方法裏面加了同步隊列初始化判斷，使用 compareAndSetHead 方法保證設置頭節點的原子性，同樣它底層也是基於 Unsafe 類，然後外層套了一個 for (; 死循環，循環唯一的退出條件是從隊尾入隊成功，也就是說如果從該方法成功返回了就表示已經入隊成功了，至此，addWaiter 執行完畢返回當前 Node 節點。然後以該節點作為 acquireQueued 方法的入參繼續進行其它步驟，該方法如下所示：

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

可以看到，該方法本質上也是通過一個死循環（自旋）去獲取鎖並且支持中斷，在循環體外面定義兩個標記變量，failed 標記是否成功獲取到鎖，interrupted 標記在等待的過程中是否被中斷過。方法首先通過 predecessor 獲取當前節點的前驅節點，噹噹前節點的前驅節點是 head 頭節點時就調用 tryAcquire 嘗試獲取鎖，也就是第二個節點則嘗試獲取鎖，這裏為什麼要從第二個節點才嘗試獲取鎖呢？是因為同步隊列本質上是一個雙向鏈表，在雙向鏈表中，第一個節點並不存儲任何數據是虛節點，只是起到一個佔位的作用，真正存儲數據的節點是從第二個節點開始的。如果成功獲取鎖，也就是 tryAcquire 方法返回 true 后，將 head 指向當前節點並把之前找到的頭節點 p 從隊列中移除，修改是否成功獲取到鎖標記，結束方法返回中斷標記。
如果當前節點的前驅節點 p 不是頭節點或者前驅節點 p 是頭節點但是獲取鎖操作失敗，那麼會調用 shouldParkAfterFailedAcquire 方法判斷當前 node 節點是否需要被阻塞，這裏的阻塞判斷主要是為了防止長時間自旋給 CPU 帶來非常大的執行開銷，浪費資源。該方法源碼如下：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        /*
          * This node has already set status asking a release
          * to signal it, so it can safely park.
          */
        return true;
    if (ws > 0) {
        /*
          * Predecessor was cancelled. Skip over predecessors and
          * indicate retry.
          */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
          * waitStatus must be 0 or PROPAGATE.  Indicate that we
          * need a signal, but don't park yet.  Caller will need to
          * retry to make sure it cannot acquire before parking.
          */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

方法參數為當前節點的前驅節點以及當前節點，主要是靠前驅節點來判斷是否需要進行阻塞，首先獲取到前驅節點的等待狀態 ws，如果節點狀態 ws 為 SIGNAL，表示前驅節點的線程已經準備完畢，等待資源釋放，方法返回 true 表示可以阻塞，如果 ws > 0，通過上文可以知道節點只有一個狀態 CANCELLED（值為 1） 滿足該條件，表示該節點線程獲取鎖的請求已經取消了，會通過一個 do-while 循環向前查找 CANCELLED 狀態的節點並將其從同步隊列中移除，否則進入 else 分支，使用 compareAndSetWaitStatus 原子操作將前驅節點的等待狀態修改為 SIGNAL，以上這兩種情況都不需要進行阻塞方法返回 false。
當經過判斷後需要阻塞的話，也就是 compareAndSetWaitStatus 方法返回 true 時，會通過 parkAndCheckInterrupt 方法阻塞掛起當前線程，並返回當前線程的中斷標識。方法如下：

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

線程阻塞是通過 LockSupport 這個工具類實現的，深入其源碼可以發現它底層也是基於 Unsafe 類實現的。如果以上兩個方法都返回 true 的話就更新中斷標記。這裏還有一個問題就是什麼時候會將一個節點的等待狀態 waitStatus 修改為 CANCELLED 節點線程獲取鎖的請求取消狀態呢？細心的朋友可能已經發現了，在上文貼出的 acquireQueued 方法源碼中的 finally 塊中會根據 failed 標記來決定是否調用 cancelAcquire 方法，這個方法就是用來將節點狀態修改為 CANCELLED 的，方法的具體實現留給大家去探索。至此 AQS 獨佔式同步狀態獲取鎖的流程就完成了，下面通過一個流程圖來看看整體流程：

下面再看看獨佔式鎖釋放的過程，同步器使用 release 方法來讓我們進行獨佔式鎖的釋放，其方法源碼如下：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

首先調用 tryRelease 方法嘗試進行鎖釋放操作，繼續跟進該方法發現同步器只是拋出了一個不支持操作異常 UnsupportedOperationException，這裏和上文獨佔鎖獲取中 tryAcquire 方法是一樣的套路，需要開發者自己定義鎖釋放操作。

通過其 JavaDoc 可以得知，如果返回 false，則表示釋放鎖失敗，方法結束。該方法如果返回 true，則表示當前線程釋放鎖成功，需要通知隊列中等待獲取鎖的線程進行鎖獲取操作。首先獲取頭節點 head，如果當前頭節點不為 null，並且其等待狀態不是初始狀態（0），則解除線程阻塞掛起狀態，通過 unparkSuccessor 方法實現，該方法源碼如下：

private void unparkSuccessor(Node node) {
    /*
      * If status is negative (i.e., possibly needing signal) try
      * to clear in anticipation of signalling.  It is OK if this
      * fails or if status is changed by waiting thread.
      */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
      * Thread to unpark is held in successor, which is normally
      * just the next node.  But if cancelled or apparently null,
      * traverse backwards from tail to find the actual
      * non-cancelled successor.
      */
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

首先獲取頭節點的等待狀態 ws，如果狀態值為負數（Node.SIGNAL or Node.PROPAGATE），則通過 CAS 操作將其改為初始狀態（0），然後獲取頭節點的後繼節點，如果後繼節點為 null 或者後繼節點狀態為 CANCELLED（獲取鎖請求已取消），就從隊列尾部開始尋找第一個狀態為非 CANCELLED 的節點，如果該節點不為空則使用 LockSupport 的 unpark 方法將其喚醒，該方法底層是通過 Unsafe 類的 unpark 實現的。這裏需要從隊尾查找非 CANCELLED 狀態的節點的原因是，在之前的獲取獨佔鎖失敗時的入隊 addWaiter 方法實現中，該方法如下：

假設一個線程執行到了上圖中的 ① 處，② 處還沒有執行，此時另一個線程恰好執行了 unparkSuccessor 方法，那麼就無法通過從前向後查找了，因為節點的後繼指針 next 還沒賦值呢，所以需要從后往前進行查找。至此，獨佔式鎖釋放操作就結束了，同樣的，最後我們也通過一個流程圖來看看整個鎖釋放的過程：

獨佔式可中斷同步狀態獲取

同步器提供了 acquireInterruptibly 方法來進行可響應中斷的獲取鎖操作，方法實現源碼如下：

public final void acquireInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg);
}

方法首先檢查當前線程的中斷狀態，如果已中斷，則直接拋出中斷異常 InterruptedException 即響應中斷，否則調用 tryAcquire 方法嘗試獲取鎖，如果獲取成功則方法結束返回，獲取失敗調用 doAcquireInterruptibly 方法，跟進該方法如下：

private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

仔細觀察可以發現該方法實現源碼和上文中 acquireQueued 方法的實現基本上類似，只是這裏把入隊操作 addWaiter 放到了方法裏面了，還有一個區別就是當在循環體內判斷需要進行中斷時會直接拋出異常來響應中斷，兩個方法的對比如下：

其它步驟和獨佔式鎖獲取一致，流程圖大體上和不響應中斷的鎖獲取差不多，只是在最開始多了一步線程中斷狀態檢查和循環是會拋出中斷異常而已。

獨佔式超時獲取同步狀態

同步器提供了 tryAcquireNanos 方法可以超時獲取同步狀態（也就是鎖），該方法提供了之前 synchronized 關鍵字不支持的超時獲取的特性，通過該方法我們可以在指定時間段 nanosTimeout 內獲取鎖，如果獲取到鎖則返回 true，否則，返回 false。方法源碼如下：

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}

首先會調用 tryAcquire 方法嘗試獲取一次鎖，如果獲取鎖成功則立即返回，否則調用 doAcquireNanos 方法進入超時獲取鎖流程。通過上文可以得知，同步器的 acquireInterruptibly 方法在等待獲取同步狀態時，如果當前線程被中斷了，會拋出中斷異常 InterruptedException 並立刻返回。超時獲取鎖的流程其實是在響應中斷的基礎上增加了超時獲取的特性，doAcquireNanos 方法的源碼如下：

private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

由以上方法實現源碼可以看出，針對超時獲取這裏主要實現思路是：先使用當前時間加上參數傳入的超時時間間隔 deadline 計算出超時的時間點，然後每次進行循環的時候使用超時時間點 deadline 減去當前時間得到剩餘的時間 nanosTimeout，如果剩餘時間小於 0 則證明當前獲取鎖操作已經超時，方法結束返回 false，反如果剩餘時間大於 0。
可以看到在裏面執行自旋的時候和上面獨佔式同步獲取鎖狀態 acquireQueued 方法那裡是一樣的套路，即噹噹前節點的前驅節點為頭節點時調用 tryAcquire 嘗試獲取鎖，如果獲取成功則返回。

除了超時時間計算那裡不同外，還有個不同的地方就是在超時獲取鎖失敗之後的操作，如果當前線程獲取鎖失敗，則判斷剩餘超時時間 nanosTimeout 是否小於 0，如果小於 0 則表示已經超時方法立即返回，反之則會判斷是否需要進行阻塞掛起當前線程，如果通過 shouldParkAfterFailedAcquire 方法判斷需要掛起阻塞當前線程，還要進一步比較超時剩餘時間 nanosTimeout 和 spinForTimeoutThreshold 的大小，如果小於等於 spinForTimeoutThreshold 值（1000 納秒）的話，將不會使當前線程進行超時等待，而是再次進行自旋過程。
加後面這個判斷的主要原因在於，在非常短（小於 1000 納秒）的時間內的等待無法做到十分精確，如果這時還進行超時等待的話，反而會讓我們指定 nanosTimeout 的超時從整體上給人感覺反而不太精確，因此，在剩餘超時時間非常短的情況下，同步器會再次自旋進行超時獲取鎖的過程，獨佔式超時獲取鎖整個過程如下所示：

共享式同步狀態獲取與釋放

共享鎖顧名思義就是可以多個線程共用一個鎖，在同步器中使用 acquireShared 來獲取共享鎖（同步狀態），方法源碼如下：

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

首先通過 tryAcquireShared 嘗試獲取共享鎖，該方法是一個模板方法在同步器中只是拋出一個不支持操作異常，需要開發人員自己去實現，同時方法的返回值有三種不同的類型分別代表三種不同的狀態，其含義如下：

1. 小於 0 表示當前線程獲取鎖失敗
2. 等於 0 表示當前線程獲取鎖成功，但是之後的線程在沒有鎖釋放的情況下獲取鎖將失敗，也就是說這個鎖是共享模式下的最後一把鎖了
3. 大於 0 表示當前線程獲取鎖成功，並且還有剩餘的鎖可以獲取

當方法 tryAcquireShared 返回值小於 0 時，也就是獲取鎖失敗，將會執行方法 doAcquireShared，繼續跟進該方法：

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

方法首先調用 addWaiter 方法封裝當前線程和等待狀態為共享模塊的節點並將其添加到等待同步隊列中，可以發現在共享模式下節點的 nextWaiter 屬性是固定值 Node.SHARED。然後循環獲取當前節點的前驅節點，如果前驅節點是頭節點的話就嘗試獲取共享鎖，如果返回值大於等於 0 表示獲取共享鎖成功，則調用 setHeadAndPropagate 方法，更新頭節點同時如果有可用資源，則向後傳播，喚醒後繼節點，接下來會檢查一下中斷標識，如果已經中斷則中斷當前線程，方法結束返回。如果返回值小於 0，則表示獲取鎖失敗，需要掛起阻塞當前線程或者繼續自旋獲取共享鎖。下面看看 setHeadAndPropagate 方法的具體實現：

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);
    /*
        * Try to signal next queued node if:
        *   Propagation was indicated by caller,
        *     or was recorded (as h.waitStatus either before
        *     or after setHead) by a previous operation
        *     (note: this uses sign-check of waitStatus because
        *      PROPAGATE status may transition to SIGNAL.)
        * and
        *   The next node is waiting in shared mode,
        *     or we don't know, because it appears null
        *
        * The conservatism in both of these checks may cause
        * unnecessary wake-ups, but only when there are multiple
        * racing acquires/releases, so most need signals now or soon
        * anyway.
        */
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

首先將當前獲取到鎖的節點設置為頭節點，然後方法參數 propagate > 0 時表示之前 tryAcquireShared 方法的返回值大於 0，也就是說當前還有剩餘的共享鎖可以獲取，則獲取當前節點的後繼節點並且後繼節點是共享節點時喚醒節點去嘗試獲取鎖，doReleaseShared 方法是同步器共享鎖釋放的主要邏輯。

同步器提供了 releaseShared 方法來進行共享鎖的釋放，方法源碼如下所示：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

首先調用 tryReleaseShared 方法嘗試釋放共享鎖，方法返回 false 代表鎖釋放失敗，方法結束返回 false，否則就表示成功釋放鎖，然後執行 doReleaseShared 方法，進行喚醒後繼節點並檢查它是否可以向後傳播等操作。繼續跟進該方法如下：

private void doReleaseShared() {
        /*
        * Ensure that a release propagates, even if there are other
        * in-progress acquires/releases.  This proceeds in the usual
        * way of trying to unparkSuccessor of head if it needs
        * signal. But if it does not, status is set to PROPAGATE to
        * ensure that upon release, propagation continues.
        * Additionally, we must loop in case a new node is added
        * while we are doing this. Also, unlike other uses of
        * unparkSuccessor, we need to know if CAS to reset status
        * fails, if so rechecking.
        */
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

可以看到和獨佔式鎖釋放不同的是，在共享模式下，狀態同步和釋放可以同時執行，其原子性由 CAS 來保證，如果頭節點改變了也會繼續循環。每次共享節點在共享模式下喚醒時，頭節點都會指向它，這樣就可以保證可以獲取到共享鎖的所有後續節點都可以喚醒了。

如何自定義同步組件

在 JDK 中基於同步器實現的一些類絕大部分都是聚合了一個或多個繼承了同步器的類，使用同步器提供的模板方法自定義內部同步狀態的管理，然後通過這個內部類去實現同步狀態管理的功能，其實這從某種程度上來說使用了 模板模式。比如 JDK 中可重入鎖 ReentrantLock、讀寫鎖 ReentrantReadWriteLock、信號量 Semaphore 以及同步工具類 CountDownLatch 等，其源碼部分截圖如下：

通過上文可以知道，我們基於同步器可以分別自定義獨佔鎖同步組件和共享鎖同步組件，下面以實現一個在同一個時刻最多只允許 3 個線程訪問，其它線程的訪問將被阻塞的同步工具 TripletsLock 為例，很顯然這個工具是共享鎖模式，主要思路就是去實現一個 JDk 中的 Lock 接口來提供面向使用者的方法，比如，調用 lock 方法獲取鎖，使用 unlock 來對鎖進行釋放等，在 TripletsLock 類內部有一個自定義同步器 Sync 繼承自同步器 AQS，用來對線程的訪問和同步狀態進行控制，當線程調用 lock 方法獲取鎖時，自定義同步器 Sync 先計算出獲取到鎖后的同步狀態，然後使用 Unsafe 類操作來保證同步狀態更新的原子性，由於同一時刻只能 3 個線程訪問，這裏我們可以將同步狀態 state 的初始值設置為 3，表示當前可用的同步資源數量，當有線程成功獲取到鎖時將同步狀態 state 減 1，有線程成功釋放鎖時將同步狀態加 1，同步狀態的取值範圍為 0、1、2、3，同步狀態為 0 時表示沒有可用同步資源，這個時候如果有線程訪問將被阻塞。下面來看看這個自定義同步組件的實現代碼：

/**
 * @author mghio
 * @date: 2020-06-13
 * @version: 1.0
 * @description:
 * @since JDK 1.8
 */
public class TripletsLock implements Lock {

  private final Sync sync = new Sync(3);

  private static final class Sync extends AbstractQueuedSynchronizer {
    public Sync(int state) {
      setState(state);
    }

    Condition newCondition() {
      return new ConditionObject();
    }

    @Override
    protected int tryAcquireShared(int reduceCount) {
      for (; ;) {
        int currentState = getState();
        int newState = currentState - reduceCount;
        if (newState < 0 || compareAndSetState(currentState, newState)) {
          return newState;
        }
      }
    }

    @Override
    protected boolean tryReleaseShared(int count) {
      for (; ;) {
        int currentState = getState();
        int newState = currentState + count;
        if (compareAndSetState(currentState, newState)) {
          return true;
        }
      }
    }
  }

  @Override
  public void lock() {
    sync.acquireShared(1);
  }

  @Override
  public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
  }

  @Override
  public boolean tryLock() {
    return sync.tryAcquireShared(1) > 0;
  }

  @Override
  public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(timeout));
  }

  @Override
  public void unlock() {
    sync.releaseShared(1);
  }

  @Override
  public Condition newCondition() {
    return sync.newCondition();
  }
}

下面啟動 20 個線程測試看看自定義同步同步工具類 TripletsLock 是否達到我們的預期。測試代碼如下：

/**
 * @author mghio
 * @date: 2020-06-13
 * @version: 1.0
 * @description:
 * @since JDK 1.8
 */
public class TripletsLockTest {
  private final Lock lock = new TripletsLock();
  private final DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

  @Test
  public void testTripletsLock() {
    // 啟動 20 個線程
    for (int i = 0; i < 20; i++) {
      Thread worker = new Runner();
      worker.setDaemon(true);
      worker.start();
    }

    for (int i = 0; i < 20; i++) {
      second(2);
      System.out.println();
    }
  }

  private class Runner extends Thread {
    @Override
    public void run() {
      for (; ;) {
        lock.lock();
        try {
          second(1);
          System.out.println(dateFormat.format(new Date()) + " ----> " + Thread.currentThread().getName());
          second(1);
        } finally {
          lock.unlock();
        }
      }
    }
  }

  private static void second(long seconds) {
    try {
      TimeUnit.SECONDS.sleep(seconds);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

測試結果如下：

從以上測試結果可以發現，同一時刻只有三個線程可以獲取到鎖，符合預期，這裏需要明確的是這個鎖獲取過程是非公平的。

總結

本文主要是對同步器中的基礎數據結構、獨佔式與共享式同步狀態獲取與釋放過程做了簡要分析，由於水平有限如有錯誤之處還請留言討論。隊列同步器 AbstractQueuedSynchronizer 是 JDK 中很多的一些多線程併發工具類的實現基礎框架，對其深入學習理解有助於我們更好的去使用其特性和相關工具類。

參考文章

Java併發編程的藝術
Java Synchronizer – AQS Learning
從 ReentrantLock 的實現看 AQS 的原理及應用
The java.util.concurrent Synchronizer Framework

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※別再煩惱如何寫文案,掌握八大原則!

※教你寫出一流的銷售文案?

※超省錢租車方案

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※產品缺大量曝光嗎?你需要的是一流包裝設計!

在Asp.NET Core中如何優雅的管理用戶機密數據

背景

回顧

在軟件開發過程中，使用配置文件來管理某些對應用程序運行中需要使用的參數是常見的作法。在早期VB/VB.NET時代，經常使用.ini文件來進行配置管理；而在.NET FX開發中，我們則傾向於使用web.config文件，通過配置appsetting的配置節來處理；而在.NET Core開發中，我們有了新的基於json格式的appsetting.json文件。

無論採用哪種方式，其實配置管理從來都是一件看起來簡單，但影響非常深遠的基礎性工作。尤其是配置的安全性，貫穿應用程序的始終，如果沒能做好安全性問題，極有可能會給系統帶來不可控的風向。

源代碼比配置文件安全么？

有人以為把配置存放在源代碼中，可能比存放在明文的配置文件中似乎更安全，其實是“皇帝的新裝”。

在前不久，筆者的一位朋友就跟我說了一段故事：他說一位同事在離職后，直接將曾經寫過的一段代碼上傳到github的公共倉庫，而這段代碼中包含了某些涉及到原企業的機密數據，還好被github的安全機制提前發現而及時終止了該行為，否則後果不堪設想。

於是，筆者順手查了一下由於有意或無意泄露企業機密，造成企業損失的案例，發現還真不少。例如大疆前員工通過 Github 泄露公司源代碼，被罰 20 萬、獲刑半年 這起案件，也是一個典型的案例。

該員工離職后，將包含關鍵配置信息的源代碼上傳到github的公共倉庫，被黑客利用，使得大量用戶私人數據被黑客獲取，該前員工最終被刑拘。

圖片來源： http://www.digitalmunition.com/WhyIWalkedFrom3k.pdf

大部分IT公司都會在入職前進行背景調查，而一旦有案底，可能就已經與許多IT公司無緣；即便是成為創業者，也可能面臨無法跟很多正規企業合作的問題。

小結

所以，安全性問題不容小覷，哪怕時間再忙，也不要急匆匆的就將數據庫連接字符串或其他包含敏感信息的內容輕易的記錄在源代碼或配置文件中。在這個點上，一旦出現問題，往往都是非常嚴重的問題。

如何實現

在.NET FX時代，我們可以使用對web.config文件的關鍵配置節進行加密的方式，來保護我們的敏感信息，在.NET Core中，自然也有這些東西，接下來我將簡述在開發環境和生產環境下不同的配置加密手段，希望能夠給讀者帶來啟迪。

開發環境

在開發環境下，我們可以使用visual studio 工具提供的用戶機密管理器，只需0行代碼，即可輕鬆完成關鍵配置節的處理。

機密管理器概述

根據微軟官方文檔 的描述：

ASP.NET Core 機密管理器工具提供了開發過程中在源代碼外部保存機密的另一種方法 。 若要使用機密管理器工具，請在項目文件中安裝包 Microsoft.Extensions.Configuration.SecretManager 。 如果該依賴項存在並且已還原，則可以使用 dotnet user-secrets 命令來通過命令行設置機密的值。 這些機密將存儲在用戶配置文件目錄中的 JSON 文件中（詳細信息隨操作系統而異），與源代碼無關。

機密管理器工具設置的機密是由使用機密的項目的 UserSecretsId 屬性組織的。 因此，必須確保在項目文件中設置 UserSecretsId 屬性，如下面的代碼片段所示。 默認值是 Visual Studio 分配的 GUID，但實際字符串並不重要，只要它在計算機中是唯一的。

<PropertyGroup>
   <UserSecretsId>UniqueIdentifyingString</UserSecretsId>
</PropertyGroup> 

Secret Manager工具允許開發人員在開發ASP.NET Core應用程序期間存儲和檢索敏感數據。敏感數據存儲在與應用程序源代碼不同的位置。由於Secret Manager將秘密與源代碼分開存儲，因此敏感數據不會提交到源代碼存儲庫。但機密管理器不會對存儲的敏感數據進行加密，因此不應將其視為可信存儲。敏感數據作為鍵值對存儲在JSON文件中。最好不要在開發和測試環境中使用生產機密。查看引文。

存放位置

在windows平台下，機密數據的存放位置為：

%APPDATA%\Microsoft\UserSecrets\\secrets.json

而在Linux/MacOs平台下，機密數據的存放位置為：

 ~/.microsoft/usersecrets/<user_secrets_id>/secrets.json 

在前面的文件路徑中， “將替換UserSecretsId為 .csproj文件中指定的值。

在Windows環境下使用機密管理器

在windows下，如果使用Visual Studio2019作為主力開發環境，只需在項目右鍵單擊，選擇菜單【管理用戶機密】，即可添加用戶機密數據。

在管理用戶機密數據中，添加的配置信息和傳統的配置信息沒有任何區別。

{
“ConnectionStrings”: {
“Default”: “Server=xxx;Database=xxx;User ID=xxx;Password=xxx;”
}
}

我們同樣也可以使用IConfiguration的方式、IOptions 的方式，進行配置的訪問。

在非Windows/非Visual Studio環境下使用機密管理器

完成安裝dotnet-cli后，在控制台輸入

dotnet user-secrets init 

前面的命令將在UserSecretsId .csproj 文件的PropertyGroup中添加 .csproj一個元素。 UserSecretsId是對項目是唯一的Guid值。

 <PropertyGroup>  
 	<TargetFramework>netcoreapp3.1</TargetFramework>
    <UserSecretsId>79a3edd0-2092-40a2-a04d-dcb46d5ca9ed</UserSecretsId> 
 </PropertyGroup> 

設置機密

 dotnet user-secrets set "Movies:ServiceApiKey" "12345" 

列出機密

 dotnet user-secrets list 

刪除機密

 dotnet user-secrets remove "Movies:ConnectionString" 

清除所有機密

 dotnet user-secrets clear 

生產環境

機密管理器為開發者在開發環境下提供了一種保留機密數據的方法，但在開發環境下是不建議使用的，如果想在生產環境下，對機密數據進行保存該怎麼辦？

按照微軟官方文檔的說法，推薦使用Azure Key Vault 來保護機密數據，但。。我不是貴雲的用戶（當然，買不起貴雲不是貴雲太貴，而是我個人的問題[手動狗頭]）。

其次，與Azure Key Valut類似的套件，例如其他雲，差不多都有，所以都可以為我們所用。

但。。如果您如果跟我一樣，不想通過第三方依賴的形式來解決這個問題，那不如就用最簡單的辦法，例如AES加密。

使用AES加密配置節

該方法與平時使用AES對字符串進行加密和解密的方法並無區別，此處從略。

使用數據保護Api（DataProtect Api實現）

在平時開發過程中，能夠動手擼AES加密是一種非常好的習慣，而微軟官方提供的數據保護API則將這個過程進一步簡化，只需調Api即可完成相應的數據加密操作。

關於數據保護api， Savorboard 大佬曾經寫過3篇博客討論這個技術問題，大家可以參考下面的文章來獲取信息。

ASP.NET Core 數據保護（Data Protection 集群場景）【上】

ASP.NET Core 數據保護（Data Protection 集群場景）【中】

ASP.NET Core 數據保護（Data Protection 集群場景）【下】

(接下來我要貼代碼了，如果沒興趣，請出門左拐，代碼不能完整運行，查看代碼）

首先，注入配置項

 public static IServiceCollection AddProtectedConfiguration(this IServiceCollection services, string directory)
        {
            services
                .AddDataProtection()
                .PersistKeysToFileSystem(new DirectoryInfo(directory))
                .UseCustomCryptographicAlgorithms(new ManagedAuthenticatedEncryptorConfiguration
                {
                    EncryptionAlgorithmType = typeof(Aes),
                    EncryptionAlgorithmKeySize = 256,
                    ValidationAlgorithmType = typeof(HMACSHA256)
                });
            ;

            return services;
        }

其次，實現對配置節的加/解密。（使用AES算法的數據保護機制）

public class ProtectedConfigurationSection : IConfigurationSection
    {
        private readonly IDataProtectionProvider _dataProtectionProvider;
        private readonly IConfigurationSection _section;
        private readonly Lazy<IDataProtector> _protector;

        public ProtectedConfigurationSection(
            IDataProtectionProvider dataProtectionProvider,
            IConfigurationSection section)
        {
            _dataProtectionProvider = dataProtectionProvider;
            _section = section;

            _protector = new Lazy<IDataProtector>(() => dataProtectionProvider.CreateProtector(section.Path));
        }

        public IConfigurationSection GetSection(string key)
        {
            return new ProtectedConfigurationSection(_dataProtectionProvider, _section.GetSection(key));
        }

        public IEnumerable<IConfigurationSection> GetChildren()
        {
            return _section.GetChildren()
                .Select(x => new ProtectedConfigurationSection(_dataProtectionProvider, x));
        }

        public IChangeToken GetReloadToken()
        {
            return _section.GetReloadToken();
        }

        public string this[string key]
        {
            get => GetProtectedValue(_section[key]);
            set => _section[key] = _protector.Value.Protect(value);
        }

        public string Key => _section.Key;
        public string Path => _section.Path;

        public string Value
        {
            get => GetProtectedValue(_section.Value);
            set => _section.Value = _protector.Value.Protect(value);
        }

        private string GetProtectedValue(string value)
        {
            if (value == null)
                return null;

            return _protector.Value.Unprotect(value);
        }
    }

再次，在使用前，先將待加密的字符串轉換成BASE64純文本，然後再使用數據保護API對數據進行處理，得到處理后的字符串。

private readonly IDataProtectionProvider _dataProtectorTokenProvider;
public TokenAuthController( IDataProtectionProvider dataProtectorTokenProvider)
{
}
[Route("encrypt"), HttpGet, HttpPost]
public string Encrypt(string section, string value)
{
     var protector = _dataProtectorTokenProvider.CreateProtector(section);
     return protector.Protect(value);
}

再替換配置文件中的對應內容。

{
  "ConnectionStrings": {
    "Default": "此處是加密后的字符串"
  }
}

然後我們就可以按照平時獲取IOptions 的方式來獲取了。

問題

公眾號【DotNET騷操作】號主【周傑】同學提出以下觀點：

1、在生產環境下，使用AES加密，其實依然是一種不夠安全的行為，充其量也就能忽悠下產品經理，畢竟幾條簡單的語句，就能把機密數據dump出來。

也許在這種情況下，我們應該優先考慮accessKeyId/accessSecret，盡量通過設置多級子賬號，通過授權Api的機制來管理機密數據，而不是直接暴露類似於數據庫連接字符串這樣的關鍵配置信息。另外，應該定期更換數據庫的密碼，盡量將類似的問題可能造成的風險降到最低。數據保護api也提供的類似的機制，使得開發者能夠輕鬆的管理機密數據的時效性問題。

2、配置文件放到CI/CD中，發布的時候在CI/CD中進行組裝，然後運維只是負責管理CI/CD的賬戶信息，而最高機密數據，則由其他人負責配置。

嗯，我完全同意他的第二種做法，另外考慮到由於運維同樣有可能會有意無意泄露機密數據，所以如果再給運維配備一本《刑法》，並讓他日常補習【侵犯商業秘密罪】相關條款，這個流程就更加閉環了。

結語

本文簡述了在.NET Core中，如何在開發環境下使用用戶機密管理器、在生產環境下使用AES+IDataProvider的方式來保護我們的用戶敏感數據。由於時間倉促，如有考慮不周之處，還請各位大佬批評指正。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※別再煩惱如何寫文案,掌握八大原則!

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※超省錢租車方案

※教你寫出一流的銷售文案?

※網頁設計最專業,超強功能平台可客製化

※產品缺大量曝光嗎?你需要的是一流包裝設計!

前言

從如何判定對象消亡的角度出發，垃圾收集算法可以劃分為“引用計數式垃圾收集”（Reference Counting GC）和“追蹤式垃圾收集”（Tracing GC）兩大類，這兩類也常被稱作“直接垃圾收集”和“間接垃圾收集”。由於主流Java虛擬機中使用 的都是“追蹤式垃圾收集”，所以後續介紹的垃圾收集算法都是屬於追蹤式的垃圾收集。

分代式收集理論

當前商業虛擬機的垃圾收集器，大多數都遵循了“分代收集”的理論進行設計。
主要簡歷在兩個分代假說之上：
1、弱分代假說：絕大多數對象都是“朝生夕滅”的。
2、強分代假說：熬過越多此垃圾收集過程的對象就越難以消亡。
這兩個分代假說奠定了多款常用的垃圾收集器的一致設計原則：收集器應該將Java堆劃分出不同的區域，然後將回收對象依據其年齡（對象熬過垃圾收集過程的次數）分配到不同的區域之中存儲。
把分代收集理論具體放到現在商用的Java虛擬機里，設計者一般至少會把Java堆劃分為新生代（Young Generation） 和 老年代（Old Generation兩個區域。在新生代中，每次垃圾收集時都有大批對象死去，而每次回收后存活的少量對象，將會逐步晉陞到老年代中存放。

標記-清除算法

標記-清除算法，分為“標記”和“清除”兩個階段：首先標記所有需要回收的對象，標記完成后，統一回收掉所有被標記的對象，也可以反過來，標記存活的對象，統一回收所有未被標記的對象。
這個算法有兩個主要的缺點：
第一個是執行效率不穩定，如果Java堆中有大部分是需要回收的對象，這個會進行大量標記和清除動作，導致標記和清除兩個過程的執行效率隨着對象數量增長而降低。
第二個是內存碎片化問題，標記、清除之後會產生大量不連續的內存碎片，空間碎片太多會導致當需要大對象時找不到足夠的連續內存，而提前觸發另一次垃圾收集動作。
因為這兩個缺點的原因，才會產生後續一些針對於修復這兩個缺點的算法。
標記清除算法示意圖：

標記複製算法

標記複製算法也被簡稱Wie複製算法，為了解決標記清除算法面對大量可回收對象時執行效率低的問題，而產生的一種稱為“半區複製”的垃圾收集算法。
原理是：將可用內存按容量劃分為大小相等的兩塊，每次只使用其中的一塊當這一塊內存用完了，就將還存活着的對象複製到另外一塊上面，然後再把已使用過的內存空間一次清理掉。
這種算法不用考慮空間碎片化，只需要移動堆指針，按順序分配即可，實現簡單，運行高效，但缺點也是顯而易見的，就是將可用內存縮小了原來的一半。
標記複製算法示意圖：

由於新生代里的對象“朝生夕滅”，針對這個特點，又產生了一種更優化的半區複製分代策略，稱為“Appel式回收”。具體做法是把新生代分為一塊較大的Eden空間和兩塊較小的Survivor空間，每次分配內存只是用Eden和其中一塊Survivor。當發生垃圾收集時，將Eden和Survivor中任然存活的對象一次性複製到另外一塊Survivor空間上，然後直接清理掉Eden和Survivor空間。
HotSpot虛擬機默認Eden和Survivor的大小比例是8：1，也就是說每次可利用的空間為新生代的90%，只有10%的空間會暫時“浪費”。
如果另外一塊兒Survivor沒有足夠的空間存放存活的對象了，這些對象將通過分配擔保機制直接進入到老年代。

標記整理算法

標記複製算法在對象存活率較高時就要進行較多的複製操作，效率將會降低。更關鍵的是，如果不浪費50%的空間，就需要有額外的空間進行分配擔保，以應對被使用的內存中所有對象都100%存活的極端情況，所以在老年代一般不能直接選用這種算法。
針對老年代對象的存亡特徵，產生了另外一種有針對性的“標記整理”算法。標記的過程和“標記-清除”算法一樣，也是判斷對象是否屬於垃圾的過程。但後續步驟是讓所有存活的對象都向內存空間一端移動，然後直接清理掉邊界以外的內存。
標記整理算法示意圖：

在這種算法中，在移動存活對象，尤其是在老年代這種每次回收都有大量對象存活區域，移動存活對象並更新所有引用這些對象的地方將會是一種極為負重的操作，而且這種移動操作必須在暫停用戶應用程序才能進行（也就是“Stop The World”）。但是不移動又會造成內存空間碎片化。所以各有利弊，從垃圾收集的停頓時間來看，不移動對象停頓時間更短，但從整個程序的吞吐量來看，移動對象會更划算。所以要依情況而定。
還有一種“和稀泥”的解決方案，就是平時採用標記清除算法，直到內存空間碎片化程度已經大到影響對象分配時，再採用標記整理算法收集一次，以獲得規整的內存空間。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※教你寫出一流的銷售文案?

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※回頭車貨運收費標準

※別再煩惱如何寫文案,掌握八大原則!

※超省錢租車方案

※產品缺大量曝光嗎?你需要的是一流包裝設計!

Python的垃圾回收機制包括了兩大部分：

• 引用計數(大部分在 Include/object.h 中定義)
• 標記清除+隔代回收(大部分在 Modules/gcmodule.c 中定義)

1. 引用計數機制

python中萬物皆對象，他的核心結構是：PyObject

typedef __int64 ssize_t;

typedef ssize_t         Py_ssize_t;

typedef struct _object {
    _PyObject_HEAD_EXTRA
    Py_ssize_t ob_refcnt;   // Py_ssize_t __int64
    struct _typeobject *ob_type;
} PyObject;

typedef struct {
    PyObject ob_base;
    Py_ssize_t ob_size; /* Number of items in variable part */
} PyVarObject;

PyObject是每個對象的底層數據結構，其中ob_refcnt就是作為引用計數。當一個對象有新的引用時, 它的ob_refcnt就會增加，當引用它的對象被刪除，它的ob_refcnt就會減少,當引用技術為0時，該對象的生命結束了。

1. 引用計數+1的情況
   • 對象被創建 eg: a=2
   • 對象被引用 eg: b=a
   • 對象被作為參數，傳入到一個函數中，例如func(a)
   • 對象作為一個元素，存儲在容器中，例如list1=[a, b]
2. 引用計數-1的情況
   • 對象的別名被显示的銷毀 eg: del a
   • 對象的別名被賦予新的對象 eg: a=34
   • 一個對象離開它的作用域， 例如f函數執行完畢時，func函數中的局部變量（全局變量不會）
   • 對象所在的容器被銷毀，或者從容器中刪除

如何查看對象的引用計數

import sys
a = 'hello'
sys.getrefcount(a)   
// 注意: getrefcount(a) 傳入a時, a的引用計數會加1

1.1 什麼時候觸發回收

當一個對象的引用計數變為了 0, 會直接進入釋放空間的流程

/* cpython/Include/object.h */
static inline void _Py_DECREF(const char *filename, int lineno,
                              PyObject *op)
{
    _Py_DEC_REFTOTAL;
    if (--op->ob_refcnt != 0) {
#ifdef Py_REF_DEBUG
        if (op->ob_refcnt < 0) {
            _Py_NegativeRefcount(filename, lineno, op);
        }
#endif
    }
    else {
    	/* // _Py_Dealloc 會找到對應類型的 descructor, 並且調用這個 descructor
        destructor dealloc = Py_TYPE(op)->tp_dealloc;
        (*dealloc)(op);
        */
        _Py_Dealloc(op);
    }
}

2. 常駐內存對象

引用計數機制所帶來的維護引用計數的額外操作，與python運行中所進行的內存分配、釋放、引用賦值的次數是成正比的，這一點，相對於主流的垃圾回收技術，比如標記–清除(mark--sweep)、停止–複製(stop--copy)等方法相比是一個弱點，因為它們帶來額外操作只和內存數量有關，至於多少人引用了這塊內存則不關心。因此為了與引用計數搭配、在內存的分配和釋放上獲得最高的效率，python設計了大量的內存池機制，比如小整數對象池、字符串的intern機制，列表的freelist緩衝池等等，這些大量使用的面向特定對象的內存池機制正是為了彌補引用計數的軟肋。

2.1 小整數對象池

#ifndef NSMALLPOSINTS
#define NSMALLPOSINTS           257
#endif
#ifndef NSMALLNEGINTS
#define NSMALLNEGINTS           5
#endif

#if NSMALLNEGINTS + NSMALLPOSINTS > 0
/* Small integers are preallocated in this array so that they
   can be shared.
   The integers that are preallocated are those in the range
   -NSMALLNEGINTS (inclusive) to NSMALLPOSINTS (not inclusive).
*/
static PyLongObject small_ints[NSMALLNEGINTS + NSMALLPOSINTS];

Py_INCREF(op)  增加對象引用計數

Py_DECREF(op)  減少對象引用計數, 如果計數位0, 調用_Py_Dealloc

_Py_Dealloc(op) 調用對應類型的 tp_dealloc 方法

小整數對象池就是一個PyLongObject 數組, 大小=257+5=262, 範圍是[-5, 257) 注意左閉右開.

python對小整數的定義是[-5, 257)， 這些整數對象是提前建立好的，不會被垃圾回收，在一個python程序中，所有位於這個範圍內的整數使用的都是同一個對象

2.2 大整數對象池

疑惑：《Python源碼剖析》提到的整數對象池block_list應該已經不存在了（因為PyLongObject為變長對象）。Python2中的PyIntObject實際是對c中的long的包裝。所以Python2也提供了專門的緩存池，供大整數輪流使用，避免每次使用不斷的malloc分配內存帶來的效率損耗，可參考劉志軍老師的講解。既然沒有池了，malloc/free會帶來的不小性能損耗。Guido認為Py3.0有極大的優化空間，在字符串和整形操作上可以取得很好的優化結果。

/* Allocate a new int object with size digits.
   Return NULL and set exception if we run out of memory. */

#define MAX_LONG_DIGITS \
    ((PY_SSIZE_T_MAX - offsetof(PyLongObject, ob_digit))/sizeof(digit))

PyLongObject *
_PyLong_New(Py_ssize_t size)
{
    PyLongObject *result;
    /* Number of bytes needed is: offsetof(PyLongObject, ob_digit) +
       sizeof(digit)*size.  Previous incarnations of this code used
       sizeof(PyVarObject) instead of the offsetof, but this risks being
       incorrect in the presence of padding between the PyVarObject header
       and the digits. */
    if (size > (Py_ssize_t)MAX_LONG_DIGITS) {
        PyErr_SetString(PyExc_OverflowError,
                        "too many digits in integer");
        return NULL;
    }
    result = PyObject_MALLOC(offsetof(PyLongObject, ob_digit) +
                             size*sizeof(digit));
    if (!result) {
        PyErr_NoMemory();
        return NULL;
    }
    return (PyLongObject*)PyObject_INIT_VAR(result, &PyLong_Type, size);
}

result = PyObject_MALLOC(offsetof(PyLongObject, ob_digit) + size*sizeof(digit));

每一個大整數，均創建一個新的對象。id(num)均不同。

2.4 字符串的intern機制

Objects/unicodeobject.c
Objects/codeobject.c

PyStringObject對象的intern機制之目的是：對於被intern之後的字符串，比如“Ruby”，在整個Python的運行期間，系統中都只有唯一的一個與字符串“Ruby”對應的PyStringObject對象。這樣當判斷兩個PyStringObject對象是否相同時，如果它們都被intern了，那麼只需要簡單地檢查它們對應的PyObject*是否相同即可。這個機制既節省了空間，又簡化了對PyStringObject對象的比較，嗯，可謂是一箭雙鵰哇。

摘自：《Python源碼剖析》 — 陳儒

Python3中PyUnicodeObject對象的intern機制和Python2的PyStringObject對象intern機制一樣，主要為了節省內存的開銷，利用字符串對象的不可變性，對存在的字符串對象重複利用

In [50]: a = 'python'

In [51]: b = 'python'

In [52]: id(a)
Out[52]: 442782398256

In [53]: id(b)
Out[53]: 442782398256

In [54]: b = 'hello python'

In [55]: a = 'hello python'

In [56]: id(a)
Out[56]: 442808585520

In [57]: id(b)
Out[57]: 442726541488

什麼樣的字符串會使用intern機制?

intern機制跟編譯時期有關，相關代碼在Objects/codeobject.c

/* Intern selected string constants */
static int
intern_string_constants(PyObject *tuple)
{
    int modified = 0;
    Py_ssize_t i;

    for (i = PyTuple_GET_SIZE(tuple); --i >= 0; ) {
        PyObject *v = PyTuple_GET_ITEM(tuple, i);
        if (PyUnicode_CheckExact(v)) {
            if (PyUnicode_READY(v) == -1) {
                PyErr_Clear();
                continue;
            }
            if (all_name_chars(v)) {
                PyObject *w = v;
                PyUnicode_InternInPlace(&v);
                if (w != v) {
                    PyTuple_SET_ITEM(tuple, i, v);
                    modified = 1;
                }
            }
        }
        /*....*/
}
    
/* all_name_chars(s): true iff s matches [a-zA-Z0-9_]* */
static int
all_name_chars(PyObject *o)
{
    const unsigned char *s, *e;

    if (!PyUnicode_IS_ASCII(o))
        return 0;

    s = PyUnicode_1BYTE_DATA(o);
    e = s + PyUnicode_GET_LENGTH(o);
    for (; s != e; s++) {
        if (!Py_ISALNUM(*s) && *s != '_')
            return 0;
    }
    return 1;
}

可見 all_name_chars 決定了是否會 intern，簡單來說就是 ascii 字母，数字和下劃線組成的字符串會被緩存。但是不僅如此。2.5還會說

/* This dictionary holds all interned unicode strings.  Note that references
   to strings in this dictionary are *not* counted in the string's ob_refcnt.
   When the interned string reaches a refcnt of 0 the string deallocation
   function will delete the reference from this dictionary.

   Another way to look at this is that to say that the actual reference
   count of a string is:  s->ob_refcnt + (s->state ? 2 : 0)
*/
static PyObject *interned = NULL;
/*省略*/
void
PyUnicode_InternInPlace(PyObject **p)
{
    PyObject *s = *p;
    PyObject *t;
#ifdef Py_DEBUG
    assert(s != NULL);
    assert(_PyUnicode_CHECK(s));
#else
    if (s == NULL || !PyUnicode_Check(s))
        return;
#endif
    /* If it's a subclass, we don't really know what putting
       it in the interned dict might do. */
    if (!PyUnicode_CheckExact(s))
        return;
    // [1]
    if (PyUnicode_CHECK_INTERNED(s))
        return;
    if (interned == NULL) {
        interned = PyDict_New();
        if (interned == NULL) {
            PyErr_Clear(); /* Don't leave an exception */
            return;
        }
    }
    Py_ALLOW_RECURSION
    // [2]
    t = PyDict_SetDefault(interned, s, s);
    Py_END_ALLOW_RECURSION
    if (t == NULL) {
        PyErr_Clear();
        return;
    }
    // [3]
    if (t != s) {
        Py_INCREF(t);
        Py_SETREF(*p, t);
        return;
    }
    // [4]
    /* The two references in interned are not counted by refcnt.
       The deallocator will take care of this */
    Py_REFCNT(s) -= 2;
    _PyUnicode_STATE(s).interned = SSTATE_INTERNED_MORTAL;
}

通過函數我們可以得知，python中維護這一個interned變量的指針，這個變量指向PyDict_New創建的對象，而PyDict_New實際上創建了一個PyDictObject對象，是Python中dict類型的對象。實際上intern機制就是維護一個字典，這個字典中記錄著被intern機制處理過的字符串對象，[1]處PyUnicode_CHECK_INTERNED宏檢查字符串對象的state.interned是否被標記，

如果字符串對象的state.interned被標記了，就直接返回；[2]處嘗試把沒有被標記的字符串對象s作為key-value加入interned字典中；[3]處表示字符串對象s已經在interned字典中（對應的value值是字符串對象t），（通過Py_SETREF宏來改變p指針的指向），且原字符串對象p會因引用計數為零被回收。Py_SETREF宏在Include/object.h定義着：

/* Safely decref `op` and set `op` to `op2`.
 *
 * As in case of Py_CLEAR "the obvious" code can be deadly:
 *
 *     Py_DECREF(op);
 *     op = op2;
 *
 * The safe way is:
 *
 *      Py_SETREF(op, op2);
 *
 * That arranges to set `op` to `op2` _before_ decref'ing, so that any code
 * triggered as a side-effect of `op` getting torn down no longer believes
 * `op` points to a valid object.
 *
 * Py_XSETREF is a variant of Py_SETREF that uses Py_XDECREF instead of
 * Py_DECREF.
 */

#define Py_SETREF(op, op2)                      \
    do {                                        \
        PyObject *_py_tmp = (PyObject *)(op);   \
        (op) = (op2);                           \
        Py_DECREF(_py_tmp);                     \
    } while (0)

[4]中把新加入interned字典中的字符串對象做減引用操作，並把state.interned標記成SSTATE_INTERNED_MORTAL。SSTATE_INTERNED_MORTAL表示字符串對象被intern機制處理，但會隨着引用計數被回收；interned標記還有另外一種SSTATE_INTERNED_IMMORTAL，表示被intern機制處理但對象不可銷毀，會與Python解釋器同在。PyUnicode_InternInPlace只能創建SSTATE_INTERNED_MORTAL狀態的字符串，要想創建SSTATE_INTERNED_IMMORTAL狀態的字符串需要通過另外一個接口，強制改變intern的狀態

void
PyUnicode_InternImmortal(PyObject **p)
{
    PyUnicode_InternInPlace(p);
    if (PyUnicode_CHECK_INTERNED(*p) != SSTATE_INTERNED_IMMORTAL) {
        _PyUnicode_STATE(*p).interned = SSTATE_INTERNED_IMMORTAL;
        Py_INCREF(*p);
    }
}

為什麼引用Py_REFCNT(s) -= 2;要-2呢？

PyDict_SetDefault(PyObject *d, PyObject *key, PyObject *defaultobj)
{
    PyDictObject *mp = (PyDictObject *)d;
    PyObject *value;
    Py_hash_t hash;

    /*...*/
    if (ix == DKIX_EMPTY) {
        /*...*/
        Py_ssize_t hashpos = find_empty_slot(mp->ma_keys, hash);
        ep0 = DK_ENTRIES(mp->ma_keys);
        ep = &ep0[mp->ma_keys->dk_nentries];
        dictkeys_set_index(mp->ma_keys, hashpos, mp->ma_keys->dk_nentries);
        Py_INCREF(key);
        Py_INCREF(value);
        /*...*/
    return value;
}

對於被intern機制處理了的PyStringObject對象，Python採用了特殊的引用計數機制。在將一個PyStringObject對象a的PyObject指針作為key和value添加到interned中時，PyDictObject對象會通過這兩個指針對a的引用計數進行兩次加1的操作。但是Python的設計者規定在interned中a的指針不能被視為對象a的有效引用，因為如果是有效引用的話，那麼a的引用計數在Python結束之前永遠都不可能為0，因為interned中至少有兩個指針引用了a，那麼刪除a就永遠不可能了，這顯然是沒有道理的。
摘自：《Python源碼剖析》 — 陳儒

注意：實際上，即使Python會對一個字符串進行intern機制的處理，也會先創建一個PyUnicodeObject對象，然後檢查在interned字典中是否有值和其相同，存在的話就將interned字典保存的value值返回，之前臨時創建的字符串對象會由於引用計數為零而回收。

是否可以直接對C原生對象做intern的動作呢？不需要創建臨時對象

事實上CPython確實提供了以char * 為參數的intern機制相關函數，但是，也是一樣的創建temp在設置intern.

PyUnicode_InternImmortal(PyObject **p)
{
    PyUnicode_InternInPlace(p);
    if (PyUnicode_CHECK_INTERNED(*p) != SSTATE_INTERNED_IMMORTAL) {
        _PyUnicode_STATE(*p).interned = SSTATE_INTERNED_IMMORTAL;
        Py_INCREF(*p);
    }
}

為什麼需要臨時對象？

因為PyDict_SetDefault() 操作的是PyDictObject對象，而該對象必須以PyObject*指針作為鍵

2.5 字符緩衝池（單字符）

python為小整數對象準備了小整數對象池，當然對於常用的字符，python對應的也建了字符串緩衝池，因為 python3 中通過 unicode_latin1[256] 將長度為 1 的 ascii 的字符也緩存了

/* Single character Unicode strings in the Latin-1 range are being
   shared as well. */
static PyObject *unicode_latin1[256] = {NULL};

unicode_decode_utf8(){
    /*省略*/
    /* ASCII is equivalent to the first 128 ordinals in Unicode. */
    if (size == 1 && (unsigned char)s[0] < 128) {
        if (consumed)
            *consumed = 1;
        return get_latin1_char((unsigned char)s[0]);
    }
    /*省略*/
}


static PyObject*
get_latin1_char(unsigned char ch)
{
    PyObject *unicode = unicode_latin1[ch];
    if (!unicode) {
        unicode = PyUnicode_New(1, ch);
        if (!unicode)
            return NULL;
        PyUnicode_1BYTE_DATA(unicode)[0] = ch;
        assert(_PyUnicode_CheckConsistency(unicode, 1));
        unicode_latin1[ch] = unicode;
    }
    Py_INCREF(unicode);
    return unicode;
}

In [46]: a = 'p'

In [47]: b = 'p'

In [48]: id(a)
Out[48]: 442757120384

In [49]: id(b)
Out[49]: 442757120384

當然單字符也包括空字符。

/* The empty Unicode object is shared to improve performance. */
static PyObject *unicode_empty = NULL;

In [8]: a = 'hello' + 'python'

In [9]: b = 'hellopython'

In [10]: a is b
Out[10]: True

In [11]: a = 'hello ' + 'python'

In [12]: b = 'hello python'

In [13]: id(a)
Out[13]: 118388503536

In [14]: id(b)
Out[14]: 118387544240

In [15]: 'hello ' + 'python' is 'hello python'
Out[15]: False

In [16]: 'hello_' + 'python' is 'hello_python'
Out[16]: True

2.6 小結：

• 小整數[-5， 257)共用對象，常駐內存

• 單個字母，長度為 1 的 ascii 的字符latin1會被interned， 包括空字符，共用對象，常駐內存

• 由字母、数字、下劃線([a-zA-Z0-9_])組成的字符串，不可修改，默認開啟intern機制，共用對象，引用計數為0時，銷毀

• 字符串（含有空格），不可修改，沒開啟intern機制，不共用對象，引用計數為0，銷毀

3. 標記清除+分代回收

為了防止出現循環引用的致命性問題，python採用的是引用計數機製為主，標記-清除和分代收集兩種機製為輔的策略。

我們設置 n1.next 指向 n2，同時設置 n2.prev 指回 n1，現在，我們的兩個節點使用循環引用的方式構成了一個`雙向鏈表`，同時請注意到 ABC 以及 DEF 的引用計數值已經增加到了2，現在，假定我們的程序不再使用這兩個節點了，我們將 n1 和 n2 都設置為None，Python會像往常一樣將每個節點的引用計數減少到1。

### 3.1 在python中的零代(Generation Zero)

Ruby使用一個鏈表(free_list)來持續追蹤未使用的、自由的對象，Python使用一種不同的鏈表來持續追蹤活躍的對象。而不將其稱之為“活躍列表”，Python的內部C代碼將其稱為零代(Generation Zero)。每次當你創建一個對象或其他什麼值的時候，Python會將其加入零代鏈表：

從上邊可以看到當我們創建ABC節點的時候，Python將其加入零代鏈表。請注意到這並不是一個真正的列表，並不能直接在你的代碼中訪問，事實上這個鏈表是一個完全內部的Python運行時。

疑惑1：對於容器對象(比如list、dict、class、instance等等)，是在什麼時候綁定GC，放入第0鏈表呢？

相似的，當我們創建DEF節點的時候，Python將其加入同樣的鏈表：

現在零代包含了兩個節點對象。(他還將包含Python創建的每個其他值，與一些Python自己使用的內部值。)

3.2 標記循環引用

當達到某個 閾值之後 解釋器會循環遍歷，循環遍歷零代列表上的每個對象，檢查列表中每個互相引用的對象，根據規則減掉其引用計數。在這個過程中，Python會一個接一個的統計內部引用的數量以防過早地釋放對象。以下例子便於理解：

從上面可以看到 ABC 和 DEF 節點包含的引用數為1.有三個其他的對象同時存在於零代鏈表中，藍色的箭頭指示了有一些對象正在被零代鏈表之外的其他對象所引用。

通過識別內部引用，Python能夠減少許多零代鏈表對象的引用計數。在上圖的第一行中你能夠看見ABC和DEF的引用計數已經變為零了，這意味着收集器可以釋放它們並回收內存空間了。剩下的活躍的對象則被移動到一個新的鏈表：一代鏈表。

疑惑2： 內部如何識別零代的循環引用計數，在什麼閾值下會觸發GC執行？

3.3 在源碼中摸索答案

Python通過PyGC_Head來跟蹤container對象，PyGC_Head信息位於PyObject_HEAD之前，定義在Include/objimpl.h中

typedef union _gc_head {
    struct {
        union _gc_head *gc_next;
        union _gc_head *gc_prev;
        Py_ssize_t gc_refs;
    } gc;
    double dummy;  /* force worst-case alignment */
} PyGC_Head;

表頭數據結構

//Include/internal/mem.h
struct gc_generation {
      PyGC_Head head;
      int threshold; /* collection threshold */  // 閾值
      int count; /* count of allocations or collections of younger
                    generations */    // 實時個數
  };

Python中用於分代垃圾收集的三個“代”由_gc_runtime_state.generations數組所表示着：

解答疑惑2，三個代的閾值如下數組

/* If we change this, we need to cbhange the default value in the
   signature of gc.collect. */
#define NUM_GENERATIONS 3

_PyGC_Initialize(struct _gc_runtime_state *state)
{
    state->enabled = 1; /* automatic collection enabled? */

#define _GEN_HEAD(n) (&state->generations[n].head)
    struct gc_generation generations[NUM_GENERATIONS] = {
        /* PyGC_Head,                                 threshold,      count */
        {{{_GEN_HEAD(0), _GEN_HEAD(0), 0}},           700,            0},
        {{{_GEN_HEAD(1), _GEN_HEAD(1), 0}},           10,             0},
        {{{_GEN_HEAD(2), _GEN_HEAD(2), 0}},           10,             0},
    };
    for (int i = 0; i < NUM_GENERATIONS; i++) {
        state->generations[i] = generations[i];
    };
    state->generation0 = GEN_HEAD(0);
    struct gc_generation permanent_generation = {
          {{&state->permanent_generation.head, &state->permanent_generation.head, 0}}, 0, 0
    };
    state->permanent_generation = permanent_generation;
}

**解答疑惑1：那container對象是什麼時候加入第0“代”的container對象鏈表呢？**

對於python內置對象的創建，container對象是通過PyObject_GC_New函數來創建的，而非container對象是通過PyObject_Malloc函數來創建的。

// Include/objimpl.h
#define PyObject_GC_New(type, typeobj) \
                ( (type *) _PyObject_GC_New(typeobj) )


// 調用了Modules/gcmodule.c中的_PyObject_GC_New函數：
PyObject *
_PyObject_GC_New(PyTypeObject *tp)
{
    PyObject *op = _PyObject_GC_Malloc(_PyObject_SIZE(tp));
    if (op != NULL)
        op = PyObject_INIT(op, tp);
    return op;
}

static PyObject *
_PyObject_GC_Alloc(int use_calloc, size_t basicsize)
{
    PyObject *op;
    PyGC_Head *g;
    size_t size;
    if (basicsize > PY_SSIZE_T_MAX - sizeof(PyGC_Head))
        return PyErr_NoMemory();
    size = sizeof(PyGC_Head) + basicsize;
    // [1]  申請PyGC_Head和對象本身的內存
    if (use_calloc)
        g = (PyGC_Head *)PyObject_Calloc(1, size);
    else
        g = (PyGC_Head *)PyObject_Malloc(size);
    if (g == NULL)
        return PyErr_NoMemory();
    // [2] 設置gc_refs的值
    g->gc.gc_refs = 0;
    _PyGCHead_SET_REFS(g, GC_UNTRACKED);
    // [3]
    generations[0].count++; /* number of allocated GC objects */
    if (generations[0].count > generations[0].threshold &&
        enabled &&
        generations[0].threshold &&
        !collecting &&
        !PyErr_Occurred()) {
        collecting = 1;
        collect_generations();
        collecting = 0;
    }
    // [4]  FROM_GC宏定義可以通過PyGC_Head地址轉換PyObject_HEAD地址，逆運算是AS_GC宏定義。
    op = FROM_GC(g);
    return op;
}

PyObject *
_PyObject_GC_Malloc(size_t basicsize)
{
    return _PyObject_GC_Alloc(0, basicsize);
}

[4] FROM_GC宏定義可以通過PyGC_Head地址轉換PyObject_HEAD地址，逆運算是AS_GC宏定義。

/* Get an object's GC head */
#define AS_GC(o) ((PyGC_Head *)(o)-1)

/* Get the object given the GC head */
#define FROM_GC(g) ((PyObject *)(((PyGC_Head *)g)+1))

當觸發閾值后，是如何進行GC回收的？

collect是垃圾回收的主入口函數。特別注意 finalizers 與 python 的__del__綁定了。

/* This is the main function.  Read this to understand how the
 * collection process works. */
static Py_ssize_t
collect(int generation, Py_ssize_t *n_collected, Py_ssize_t *n_uncollectable,
        int nofail)
{
    int i;
    Py_ssize_t m = 0; /* # objects collected */
    Py_ssize_t n = 0; /* # unreachable objects that couldn't be collected */
    PyGC_Head *young; /* the generation we are examining */
    PyGC_Head *old; /* next older generation */
    PyGC_Head unreachable; /* non-problematic unreachable trash */
    PyGC_Head finalizers;  /* objects with, & reachable from, __del__ */
    PyGC_Head *gc;
    _PyTime_t t1 = 0;   /* initialize to prevent a compiler warning */

    struct gc_generation_stats *stats = &_PyRuntime.gc.generation_stats[generation];
    
    ...

    // “標記-清除”前的準備
    
    // 垃圾標記

    // 垃圾清除
  
    ...

    /* Update stats */
    if (n_collected)
        *n_collected = m;
    if (n_uncollectable)
        *n_uncollectable = n;
    stats->collections++;
    stats->collected += m;
    stats->uncollectable += n;

    if (PyDTrace_GC_DONE_ENABLED())
        PyDTrace_GC_DONE(n+m);

    return n+m;
}

3.3.1 標記-清除前的準備

    // [1]
    /* update collection and allocation counters */
    if (generation+1 < NUM_GENERATIONS)
        _PyRuntime.gc.generations[generation+1].count += 1;
    for (i = 0; i <= generation; i++)
        _PyRuntime.gc.generations[i].count = 0;

    // [2]
    /* merge younger generations with one we are currently collecting */
    for (i = 0; i < generation; i++) {
        gc_list_merge(GEN_HEAD(i), GEN_HEAD(generation));
    }

    // [3]
    /* handy references */
    young = GEN_HEAD(generation);
    if (generation < NUM_GENERATIONS-1)
        old = GEN_HEAD(generation+1);
    else
        old = young;

    // [4]
    /* Using ob_refcnt and gc_refs, calculate which objects in the
     * container set are reachable from outside the set (i.e., have a
     * refcount greater than 0 when all the references within the
     * set are taken into account).
     */
    update_refs(young);
    subtract_refs(young);

[1] 先更新了將被回收的“代”以及老一“代”的count計數器。
這邊對老一“代”的count計數器增量1就可以看出來在第1“代”和第2“代”的count值其實表示的是該代垃圾回收的次數。
[2] 通過gc_list_merge函數將這些“代”合併成一個鏈表。

/* append list `from` onto list `to`; `from` becomes an empty list */
static void
gc_list_merge(PyGC_Head *from, PyGC_Head *to)
{
    PyGC_Head *tail;
    assert(from != to);
    if (!gc_list_is_empty(from)) {
        tail = to->gc.gc_prev;
        tail->gc.gc_next = from->gc.gc_next;
        tail->gc.gc_next->gc.gc_prev = tail;
        to->gc.gc_prev = from->gc.gc_prev;
        to->gc.gc_prev->gc.gc_next = to;
    }
    gc_list_init(from);
}

static void
gc_list_init(PyGC_Head *list)
{
    list->gc.gc_prev = list;
    list->gc.gc_next = list;
}

gc_list_merge函數將from鏈錶鏈接到to鏈表末尾並把from鏈表置為空鏈表。

[3] 經過合併操作之後，所有需要被進行垃圾回收的對象都鏈接到young“代”（滿足超過閾值的最老“代”），並記錄old“代”，後面需要將不可回收的對象移到old“代”。

鏈表的合併操作：

[4] 尋找root object集合

要對合併的鏈表進行垃圾標記，首先需要尋找root object集合。
所謂的root object即是一些全局引用和函數棧中的引用。這些引用所用的對象是不可被刪除的。

list1 = []
list2 = []
list1.append(list2)
list2.append(list1)
a = list1
del list1
del list2

上面的Python中循環引用的代碼，變量a所指向的對象就是root object。

三色標記模型

3.3.2 垃圾標記

// [1]
/* Leave everything reachable from outside young in young, and move
     * everything else (in young) to unreachable.
     * NOTE:  This used to move the reachable objects into a reachable
     * set instead.  But most things usually turn out to be reachable,
     * so it's more efficient to move the unreachable things.
     */
gc_list_init(&unreachable);
move_unreachable(young, &unreachable);

// [2]
/* Move reachable objects to next generation. */
if (young != old) {
    if (generation == NUM_GENERATIONS - 2) {
        _PyRuntime.gc.long_lived_pending += gc_list_size(young);
    }
    gc_list_merge(young, old);
}
else {
    /* We only untrack dicts in full collections, to avoid quadratic
           dict build-up. See issue #14775. */
    untrack_dicts(young);
    _PyRuntime.gc.long_lived_pending = 0;
    _PyRuntime.gc.long_lived_total = gc_list_size(young);
}

[1] 初始化不可達鏈表，調用move_unreachable函數將循環引用的對象移動到不可達鏈表中：

/* Move the unreachable objects from young to unreachable.  After this,
 * all objects in young have gc_refs = GC_REACHABLE, and all objects in
 * unreachable have gc_refs = GC_TENTATIVELY_UNREACHABLE.  All tracked
 * gc objects not in young or unreachable still have gc_refs = GC_REACHABLE.
 * All objects in young after this are directly or indirectly reachable
 * from outside the original young; and all objects in unreachable are
 * not.
 */
static void
move_unreachable(PyGC_Head *young, PyGC_Head *unreachable)
{
    PyGC_Head *gc = young->gc.gc_next;

    /* Invariants:  all objects "to the left" of us in young have gc_refs
     * = GC_REACHABLE, and are indeed reachable (directly or indirectly)
     * from outside the young list as it was at entry.  All other objects
     * from the original young "to the left" of us are in unreachable now,
     * and have gc_refs = GC_TENTATIVELY_UNREACHABLE.  All objects to the
     * left of us in 'young' now have been scanned, and no objects here
     * or to the right have been scanned yet.
     */

    while (gc != young) {
        PyGC_Head *next;

        if (_PyGCHead_REFS(gc)) {
            /* gc is definitely reachable from outside the
             * original 'young'.  Mark it as such, and traverse
             * its pointers to find any other objects that may
             * be directly reachable from it.  Note that the
             * call to tp_traverse may append objects to young,
             * so we have to wait until it returns to determine
             * the next object to visit.
             */
            PyObject *op = FROM_GC(gc);
            traverseproc traverse = Py_TYPE(op)->tp_traverse;
            assert(_PyGCHead_REFS(gc) > 0);
            _PyGCHead_SET_REFS(gc, GC_REACHABLE);
            (void) traverse(op,
                            (visitproc)visit_reachable,
                            (void *)young);
            next = gc->gc.gc_next;
            if (PyTuple_CheckExact(op)) {
                _PyTuple_MaybeUntrack(op);
            }
        }
        else {
            /* This *may* be unreachable.  To make progress,
             * assume it is.  gc isn't directly reachable from
             * any object we've already traversed, but may be
             * reachable from an object we haven't gotten to yet.
             * visit_reachable will eventually move gc back into
             * young if that's so, and we'll see it again.
             */
            next = gc->gc.gc_next;
            gc_list_move(gc, unreachable);
            _PyGCHead_SET_REFS(gc, GC_TENTATIVELY_UNREACHABLE);
        }
        gc = next;
    }
}

這邊遍歷young“代”的container對象鏈表，_PyGCHead_REFS(gc)判斷是不是root object或從某個root object能直接/間接引用的對象，由於root object集合中的對象是不能回收的，因此，被這些對象直接或間接引用的對象也是不能回收的。

_PyGCHead_REFS(gc)為0並不能斷定這個對象是可回收的，但是還是先移動到unreachable鏈表中，設置了GC_TENTATIVELY_UNREACHABLE標誌表示暫且認為是不可達的，如果是存在被root object直接或間接引用的對象，這樣的對象還會被移出unreachable鏈表中。

[2] 將可達的對象移到下一“代”。

3.3.3 垃圾清除

// [1]
    /* All objects in unreachable are trash, but objects reachable from
     * legacy finalizers (e.g. tp_del) can't safely be deleted.
     */
    gc_list_init(&finalizers);
    move_legacy_finalizers(&unreachable, &finalizers);
    /* finalizers contains the unreachable objects with a legacy finalizer;
     * unreachable objects reachable *from* those are also uncollectable,
     * and we move those into the finalizers list too.
     */
    move_legacy_finalizer_reachable(&finalizers);

    // [2]
    /* Collect statistics on collectable objects found and print
     * debugging information.
     */
    for (gc = unreachable.gc.gc_next; gc != &unreachable;
                    gc = gc->gc.gc_next) {
        m++;
    }

    // [3]
    /* Clear weakrefs and invoke callbacks as necessary. */
    m += handle_weakrefs(&unreachable, old);

    // [4]
    /* Call tp_finalize on objects which have one. */
    finalize_garbage(&unreachable);

    // [5]
    if (check_garbage(&unreachable)) {
        revive_garbage(&unreachable);
        gc_list_merge(&unreachable, old);
    }
    else {
        /* Call tp_clear on objects in the unreachable set.  This will cause
         * the reference cycles to be broken.  It may also cause some objects
         * in finalizers to be freed.
         */
        delete_garbage(&unreachable, old);
    }
    
    // [6]
    /* Collect statistics on uncollectable objects found and print
     * debugging information. */
    for (gc = finalizers.gc.gc_next;
         gc != &finalizers;
         gc = gc->gc.gc_next) {
        n++;
    }
    
    ...

    // [7]
    /* Append instances in the uncollectable set to a Python
     * reachable list of garbage.  The programmer has to deal with
     * this if they insist on creating this type of structure.
     */
    (void)handle_legacy_finalizers(&finalizers, old);
    
    /* Clear free list only during the collection of the highest
     * generation */
    if (generation == NUM_GENERATIONS-1) {
        clear_freelists();
    }

[1] 處理unreachable鏈表中有finalizer的對象。即python中 實現了__del__魔法方法的對象

/* Move the objects in unreachable with tp_del slots into `finalizers`.
 * Objects moved into `finalizers` have gc_refs set to GC_REACHABLE; the
 * objects remaining in unreachable are left at GC_TENTATIVELY_UNREACHABLE.
 */
static void
move_legacy_finalizers(PyGC_Head *unreachable, PyGC_Head *finalizers)
{
    PyGC_Head *gc;
    PyGC_Head *next;

    /* March over unreachable.  Move objects with finalizers into
     * `finalizers`.
     */
    for (gc = unreachable->gc.gc_next; gc != unreachable; gc = next) {
        PyObject *op = FROM_GC(gc);

        assert(IS_TENTATIVELY_UNREACHABLE(op));
        next = gc->gc.gc_next;

        if (has_legacy_finalizer(op)) {
            gc_list_move(gc, finalizers);
            _PyGCHead_SET_REFS(gc, GC_REACHABLE);
        }
    }
}

遍歷unreachable鏈表，將擁有finalizer的實例對象移到finalizers鏈表中，並標示為GC_REACHABLE。

/* Return true if object has a pre-PEP 442 finalization method. */
static int
has_legacy_finalizer(PyObject *op)
{
    return op->ob_type->tp_del != NULL;
}

擁有finalizer的實例對象指的就是實現了tp_del函數的對象。

/* Move objects that are reachable from finalizers, from the unreachable set
 * into finalizers set.
 */
static void
move_legacy_finalizer_reachable(PyGC_Head *finalizers)
{
    traverseproc traverse;
    PyGC_Head *gc = finalizers->gc.gc_next;
    for (; gc != finalizers; gc = gc->gc.gc_next) {
        /* Note that the finalizers list may grow during this. */
        traverse = Py_TYPE(FROM_GC(gc))->tp_traverse;
        (void) traverse(FROM_GC(gc),
                        (visitproc)visit_move,
                        (void *)finalizers);
    }
}

對finalizers鏈表中擁有finalizer的實例對象遍歷其引用對象，調用visit_move訪問者，這些被引用的對象也不應該被釋放。

/* A traversal callback for move_legacy_finalizer_reachable. */
static int
visit_move(PyObject *op, PyGC_Head *tolist)
{
    if (PyObject_IS_GC(op)) {
        if (IS_TENTATIVELY_UNREACHABLE(op)) {
            PyGC_Head *gc = AS_GC(op);
            gc_list_move(gc, tolist);
            _PyGCHead_SET_REFS(gc, GC_REACHABLE);
        }
    }
    return 0;
}

#define IS_TENTATIVELY_UNREACHABLE(o) ( \
    _PyGC_REFS(o) == GC_TENTATIVELY_UNREACHABLE)

visit_move函數將引用對象還在unreachable鏈表的對象移到finalizers鏈表中。

[2] 統計unreachable鏈表數量。
[3] 處理弱引用。
[4] [5] 開始清除垃圾對象，我們先只看delete_garbage函數：

/* Break reference cycles by clearing the containers involved.  This is
 * tricky business as the lists can be changing and we don't know which
 * objects may be freed.  It is possible I screwed something up here.
 */
static void
delete_garbage(PyGC_Head *collectable, PyGC_Head *old)
{
    inquiry clear;

    while (!gc_list_is_empty(collectable)) {
        PyGC_Head *gc = collectable->gc.gc_next;
        PyObject *op = FROM_GC(gc);

        if (_PyRuntime.gc.debug & DEBUG_SAVEALL) {
            PyList_Append(_PyRuntime.gc.garbage, op);
        }
        else {
            if ((clear = Py_TYPE(op)->tp_clear) != NULL) {
                Py_INCREF(op);
                clear(op);
                Py_DECREF(op);
            }
        }
        if (collectable->gc.gc_next == gc) {
            /* object is still alive, move it, it may die later */
            gc_list_move(gc, old);
            _PyGCHead_SET_REFS(gc, GC_REACHABLE);
        }
    }
}

遍歷unreachable鏈表中的container對象，調用其類型對象的tp_clear指針指向的函數，我們以list對象為例：

static int
_list_clear(PyListObject *a)
{
    Py_ssize_t i;
    PyObject **item = a->ob_item;
    if (item != NULL) {
        /* Because XDECREF can recursively invoke operations on
           this list, we make it empty first. */
        i = Py_SIZE(a);
        Py_SIZE(a) = 0;
        a->ob_item = NULL;
        a->allocated = 0;
        while (--i >= 0) {
            Py_XDECREF(item[i]);
        }
        PyMem_FREE(item);
    }
    /* Never fails; the return value can be ignored.
       Note that there is no guarantee that the list is actually empty
       at this point, because XDECREF may have populated it again! */
    return 0;
}

_list_clear函數對container對象的每個元素進行引用數減量操作並釋放container對象內存。

delete_garbage在對container對象進行clear操作之後，還會檢查是否成功，如果該container對象沒有從unreachable鏈表上摘除，表示container對象還不能銷毀，需要放回到老一“代”中，並標記GC_REACHABLE。

[6] 統計finalizers鏈表數量。
[7] 處理finalizers鏈表的對象。

/* Handle uncollectable garbage (cycles with tp_del slots, and stuff reachable
 * only from such cycles).
 * If DEBUG_SAVEALL, all objects in finalizers are appended to the module
 * garbage list (a Python list), else only the objects in finalizers with
 * __del__ methods are appended to garbage.  All objects in finalizers are
 * merged into the old list regardless.
 * Returns 0 if all OK, <0 on error (out of memory to grow the garbage list).
 * The finalizers list is made empty on a successful return.
 */
static int
handle_legacy_finalizers(PyGC_Head *finalizers, PyGC_Head *old)
{
    PyGC_Head *gc = finalizers->gc.gc_next;

    if (_PyRuntime.gc.garbage == NULL) {
        _PyRuntime.gc.garbage = PyList_New(0);
        if (_PyRuntime.gc.garbage == NULL)
            Py_FatalError("gc couldn't create gc.garbage list");
    }
    for (; gc != finalizers; gc = gc->gc.gc_next) {
        PyObject *op = FROM_GC(gc);

        if ((_PyRuntime.gc.debug & DEBUG_SAVEALL) || has_legacy_finalizer(op)) {
            if (PyList_Append(_PyRuntime.gc.garbage, op) < 0)
                return -1;
        }
    }

    gc_list_merge(finalizers, old);
    return 0;
}

遍歷finalizers鏈表，將擁有finalizer的實例對象放到一個名為garbage的PyListObject對象中，可以通過gc模塊查看。

>>> import gc
>>> gc.garbage

並把finalizers鏈表晉陞到老一“代”。

注意：__del__給gc帶來的影響, gc模塊唯一處理不了的是循環引用的類都有__del__方法，所以項目中要避免定義__del__方法 官方警告

3.4 小結

1. GC的流程:

   -> 發現超過閾值了
   -> 觸發垃圾回收
   -> 將所有可達對象鏈表放到一起
   -> 遍歷, 計算有效引用計數
   -> 分成 有效引用計數=0 和 有效引用計數 > 0 兩個集合
   -> 大於0的, 放入到更老一代
   -> =0的, 執行回收
   -> 回收遍歷容器內的各個元素, 減掉對應元素引用計數(破掉循環引用)
   -> 執行-1的邏輯, 若發現對象引用計數=0, 觸發內存回收
   -> 由python底層內存管理機制回收內存
   

2. 觸發GC的條件

   • 主動調用gc.collect(),

   • 當gc模塊的計數器達到閥值的時候

   • 程序退出的時候

4. GC閾值

分代回收 以空間換時間

重要思想：將系統中的所有內存塊根據其存活的時間劃分為不同的集合, 每個集合就成為一個”代”, 垃圾收集的頻率隨着”代”的存活時間的增大而減小(活得越長的對象, 就越不可能是垃圾, 就應該減少去收集的頻率)

弱代假說

分代垃圾回收算法的核心行為：垃圾回收器會更頻繁的處理新對象。一個新的對象即是你的程序剛剛創建的，而一個來的對象則是經過了幾個時間周期之後仍然存在的對象。Python會在當一個對象從零代移動到一代，或是從一代移動到二代的過程中提升(promote)這個對象。

為什麼要這麼做？這種算法的根源來自於弱代假說(weak generational hypothesis)。這個假說由兩個觀點構成：

首先是年親的對象通常死得也快，而老對象則很有可能存活更長的時間。

假定我們創建了一個Python創建：

n1 = Node("ABC")

根據假說，我的代碼很可能僅僅會使用ABC很短的時間。這個對象也許僅僅只是一個方法中的中間結果，並且隨着方法的返回這個對象就將變成垃圾了。大部分的新對象都是如此般地很快變成垃圾。然而，偶爾程序會創建一些很重要的，存活時間比較長的對象-例如web應用中的session變量或是配置項。

通過頻繁的處理零代鏈表中的新對象，Python的垃圾收集器將把時間花在更有意義的地方：它處理那些很快就可能變成垃圾的新對象。同時只在很少的時候，當滿足閾值的條件，收集器才回去處理那些老變量。

5. Python中的gc模塊使用

gc模塊默認是開啟自動回收垃圾的，gc.isenabled()=True

常用函數:

• gc.set_debug(flags) 設置gc的debug日誌，一般設置為gc.DEBUG_LEAK

"""
DEBUG_STATS - 在垃圾收集過程中打印所有統計信息
DEBUG_COLLECTABLE - 打印發現的可收集對象
DEBUG_UNCOLLECTABLE - 打印unreachable對象(除了uncollectable對象)
DEBUG_SAVEALL - 將對象保存到gc.garbage(一個列表)裏面，而不是釋放它
DEBUG_LEAK - 對內存泄漏的程序進行debug (everything but STATS).
    
"""

• gc.collect([generation]) 顯式進行垃圾回收，可以輸入參數，0代表只檢查第一代的對象，1代表檢查一，二代的對象，2代表檢查一，二，三代的對象，如果不傳參數，執行一個full collection，也就是等於傳2。 返回不可達（unreachable objects）對象的數目

• gc.get_threshold() 獲取的gc模塊中自動執行垃圾回收的頻率

• gc.get_stats()查看每一代的具體信息

• gc.set_threshold(threshold0[, threshold1[, threshold2]) 設置自動執行垃圾回收的頻率

• gc.get_count() 獲取當前自動執行垃圾回收的計數器，返回一個長度為3的列表

  例如(488,3,0)，其中488是指距離上一次一代垃圾檢查，Python分配內存的數目減去釋放內存的數目，注意是內存分配，而不是引用計數的增加。

  3是指距離上一次二代垃圾檢查，一代垃圾檢查的次數，同理，0是指距離上一次三代垃圾檢查，二代垃圾檢查的次數。

計數器和閾值關係解釋：

當計數器從(699,3,0)增加到(700,3,0)，gc模塊就會執行gc.collect(0),即檢查一代對象的垃圾，並重置計數器為(0,4,0)
當計數器從(699,9,0)增加到(700,9,0)，gc模塊就會執行gc.collect(1),即檢查一、二代對象的垃圾，並重置計數器為(0,0,1)
當計數器從(699,9,9)增加到(700,9,9)，gc模塊就會執行gc.collect(2),即檢查一、二、三代對象的垃圾，並重置計數器為(0,0,0)

6. 工作中如何避免循環引用？

To avoid circular references in your code, you can use weak references, that are implemented in the weakref module. Unlike the usual references, the weakref.ref doesn’t increase the reference count and returns None if an object was destroyed. rushter

import weakref


class Node():
    def __init__(self, value):
        self.value = value
        self._parent = None
        self.children = []

    def __repr__(self):
        return 'Node({!r:})'.format(self.value)

    @property
    def parent(self):
        return None if self._parent is None else self._parent()

    @parent.setter
    def parent(self, node):
        self._parent = weakref.ref(node)

    def add_child(self, child):
        self.children.append(child)
        child.parent = self


if __name__ == '__main__':

    a = Data()
    del a

    a = Node()
    del a

    a = Node()
    a.add_child(Node())
    del a

弱引用消除了引用循環的這個問題，本質來講，弱引用就是一個對象指針，它不會增加它的引用計數

弱引用的主要用途是實現保存大對象的高速緩存或映射，但又並希望大對象僅僅因為它出現在高速緩存或映射中而保持存活

為了訪問弱引用所引用的對象，你可以像函數一樣去調用它即可。如果那個對象還存在就會返回它，否則就返回一個None。 由於原始對象的引用計數沒有增加，那麼就可以去刪除它了

並非所有對象都可以被弱引用；可以被弱引用的對象包括類實例，用 Python（而不是用 C）編寫的函數，實例方法、集合、凍結集合，某些 文件對象，生成器，類型對象，套接字，數組，雙端隊列，正則表達式模式對象以及代碼對象等。

幾個內建類型如 list 和 dict 不直接支持弱引用，但可以通過子類化添加支持:

class Dict(dict):
    pass

obj = Dict(red=1, green=2, blue=3)   # this object is weak referenceable

其他內置類型例如 tuple 和 int 不支持弱引用，即使通過子類化也不支持

python Cookbook 書中推薦弱引用來處理循環引用

假設我們想創建一個類，用它的實例來代表臨時目錄。 當以下事件中的某一個發生時，這個目錄應當與其內容一起被刪除：

• 對象被作為垃圾回收，
• 對象的 remove() 方法被調用，或
• 程序退出。

原本用__del__()方法

class TempDir:
    def __init__(self):
        self.name = tempfile.mkdtemp()
       
   	def __remove(self):
        if self.name is not None:
            shutil.rmtree(self.name)
            self.name = None
    
    @property
    def removed(self):
        return self.name is None
   
	def __del__(self):
        self.__remove()

更健壯的替代方式可以是定義一個終結器，只引用它所需要的特定函數和對象，而不是獲取對整個對象狀態的訪問權:

class TempDir:
    def __init__(self):
        self.name = tempfile.mkdtemp()
        self._finalizer = weakref.finalize(self, shutil.rmtree, self.name)
       
   	def remove(self):
        self._finalizer()
    
    @property
    def removed(self):
        return not self._finalizer.alive

像這樣定義后，我們的終結器將只接受一個對其完成正確清理目錄任務所需細節的引用。 如果對象一直未被作為垃圾回收，終結器仍會在退出時被調用.weakref

參考文章和書籍:

1. visualizing garbage collection in ruby and python
2. 膜拜的大佬-Junnplus’blog
3. wklken前輩
4. The Garbage Collector
5. Garbage collection in Python: things you need to know
6. Python-CookBook-循環引用數據結構的內存管理
7. 《python源碼剖析》
8. Python-3.8.3/Modules/gcmodule.c

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※超省錢租車方案

※別再煩惱如何寫文案,掌握八大原則!

※回頭車貨運收費標準

※教你寫出一流的銷售文案?

※產品缺大量曝光嗎?你需要的是一流包裝設計!

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

一：背景

1. 講故事

好消息，.NET 5.0 終於在2020年6月10日發布了第五個預覽版，眼尖的同學一定看到了在這個版本中終於支持了 C# 9.0，此處有掌聲，太好了！！！

.Net5官方鏈接

可以看到目前的C#9還是預覽版，實現了一部分新語法供開發者提前嘗鮮，從github的roslyn倉庫上可以看到目前準備實現 17個新特性，現階段已經實現了8個，其中的 In Progress 表示正在開發中。

新特性預覽

2. 安裝必備

• 下載最新的net5 sdk吧： dotnet-sdk-5.0.100-preview.5.20279.10-win-x64.exe

• 下載最新的 visual studio 2019 preview 2

找好你自己的vs版本類型哦。。。

二：新特性研究

1. Target-typed new

這個取名一定要留給學易經的大師傅，沒見過世面的我不敢造次，取得不佳影響時運，所謂 運去金成鐵, 時來鐵似金 ，不過大概意思就是說直接new你定義的局部變量的類型，用issues中總結的話就是：

Summary: Allow Point p = new (x, y);
Shipped in preview in 16.7p1.

接下來就是全部代碼，看看使用前 和 使用后 的具體差別。

    class Program
    {
        static void Main(string[] args)
        {
            //老語法
            var person = new Person("mary", "123456");

            //新語法
            Person person2 = new("mary", "123456");

            Console.WriteLine($"person={person}person2={person2}");
        }
    }

    public class Person
    {
        private string username;
        private string password;

        public Person(string username, string password)
        {
            this.username = username;
            this.password = password;
        }

        public override string ToString()
        {
            return $"username={username},password={password} \n";
        }
    }

然後用ilspy去看看下面的il代碼，是不是省略了Person，讓自己心裏踏實一點。

總的來說這語法還行吧，能起到延長鍵盤使用壽命的功效。

2. Lambda discard parameters

從字面上看大概就是說可以在lambda上使用取消參數，聽起來怪怪的，那本意是什麼呢？有時候lambda上的匿名方法簽名的參數是不需要的，但在以前必須實打實的定義，這樣就會污染方法體，也就是可以在body中被訪問，如下圖：

但有時候因為客觀原因必須使用Func<int,int,int>這樣的委託，而且還不想讓方法簽名的參數污染方法體，我猜測在函數式編程中有這樣的場景吧，可能有點類似MVC中的EmptyResult效果。

好了，我想你大概知道啥意思了，接下來實操一把。。。

    Func<int, int, int> func = (_, _) =>
    {
        return 0;
    };

    var result = func(10, 20);

從圖中可以看到，我在方法體中是找不到所謂的 _ 變量的，這就神奇了，怎麼做到的呢？ 帶着這個好奇心看看它的IL代碼是個什麼樣子。

.method private hidebysig static 
	void Main (
		string[] args
	) cil managed 
{
	// Method begins at RVA 0x2048
	// Code size 45 (0x2d)
	.maxstack 3
	.entrypoint
	.locals init (
		[0] class [System.Runtime]System.Func`3<int32, int32, int32> func,
		[1] int32 result
	)

	IL_0000: nop
	IL_0001: ldsfld class [System.Runtime]System.Func`3<int32, int32, int32> ConsoleApp1.Program/'<>c'::'<>9__0_0'
	IL_0006: dup
	IL_0007: brtrue.s IL_0020

	IL_0009: pop
	IL_000a: ldsfld class ConsoleApp1.Program/'<>c' ConsoleApp1.Program/'<>c'::'<>9'
	IL_000f: ldftn instance int32 ConsoleApp1.Program/'<>c'::'<Main>b__0_0'(int32, int32)
	IL_0015: newobj instance void class [System.Runtime]System.Func`3<int32, int32, int32>::.ctor(object, native int)
	IL_001a: dup
	IL_001b: stsfld class [System.Runtime]System.Func`3<int32, int32, int32> ConsoleApp1.Program/'<>c'::'<>9__0_0'

	IL_0020: stloc.0
	IL_0021: ldloc.0
	IL_0022: ldc.i4.s 10
	IL_0024: ldc.i4.s 20
	IL_0026: callvirt instance !2 class [System.Runtime]System.Func`3<int32, int32, int32>::Invoke(!0, !1)
	IL_002b: stloc.1
	IL_002c: ret
} // end of method Program::Main

從上面的IL代碼來看 匿名方法 變成了<>c類的<Main>b__0_0方法,完整簽名： ConsoleApp1.Program/'<>c'::'<Main>b__0_0'(int32, int32)，然後再找一下 <Main>b__0_0 方法的定義。

.class nested private auto ansi sealed serializable beforefieldinit '<>c'
	extends [System.Runtime]System.Object
	.method assembly hidebysig 
		instance int32 '<Main>b__0_0' (
			int32 _,
			int32 _
		) cil managed 
	{
		// Method begins at RVA 0x2100
		// Code size 7 (0x7)
		.maxstack 1
		.locals init (
			[0] int32
		)

		IL_0000: nop
		IL_0001: ldc.i4.0
		IL_0002: stloc.0
		IL_0003: br.s IL_0005

		IL_0005: ldloc.0
		IL_0006: ret
	} // end of method '<>c'::'<Main>b__0_0'

這說明什麼呢？ 說明兩個參數是真實存在的，但編譯器搗了鬼，做了語法上的限制，不讓你訪問所謂的 _。

等等。。。有一個問題，IL中的方法簽名怎麼是這樣的： <Main>b__0_0 (int32 _,int32 _) , 大家應該知道方法簽名中不可以出現重複的參數名，比如下面這樣定義肯定是報錯的。

這說明什麼？ 說明這個語法糖不僅需要編譯器支持，更需要底層的JIT支持，那怎麼證明呢？我們用windbg去底層挖一挖。。。為了方便調試,修改如下：

        static void Main(string[] args)
        {
            Func<int, int, int> func = (_, _) =>
            {
                Console.WriteLine("進入方法體了！！！");
                Console.ReadLine();
                return 0;
            };

            var result = func(10, 20);
        }

0:000> !clrstack -p
OS Thread Id: 0x52e8 (0)
0000007035F7E5C0 00007ffaff362655 ConsoleApp1.Program+c.b__0_0(Int32, Int32) [C:\5\ConsoleApp1\ConsoleApp1\Program.cs @ 13]
    PARAMETERS:
        this (0x0000007035F7E600) = 0x000001968000cb48
        _ (0x0000007035F7E608) = 0x000000000000000a
        _ (0x0000007035F7E610) = 0x0000000000000014

從圖中可以看到，雖然都是 _ ,但在線程棧上是完完全全的兩個棧地址。 0x0000007035F7E608 和 0x0000007035F7E610。

三：總結

總的來說，C#是越來越向函數式編程靠攏，越來越像Scala，就像Jquery的口號一樣： Write less，do more。

好了，先就說這兩個吧，大家先安裝好工具，明天繼續解剖~~~

如您有更多問題與我互動，掃描下方進來吧~

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※網頁設計公司推薦不同的風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

數據存儲和有效期

在 redis 工作流程中，過期的數據並不需要馬上就要執行刪除操作。因為這些刪不刪除只是一種狀態表示，可以異步的去處理，在不忙的時候去把這些不緊急的刪除操作做了，從而保證 redis 的高效

數據的存儲

在redis中數據的存儲不僅僅需要保存數據本身還要保存數據的生命周期，也就是過期時間。在redis 中 數據的存儲結構如下圖：

獲取有效期

Redis是一種內存級數據庫，所有數據均存放在內存中，內存中的數據可以通過TTL指令獲取其狀態

刪除策略

在內存佔用與CPU佔用之間尋找一種平衡，顧此失彼都會造成整體redis性能的下降，甚至引發服務器宕機或內存泄漏。

定時刪除

創建一個定時器，當key設置過期時間，且過期時間到達時，由定時器任務立即執行對鍵的刪除操作

優點

節約內存，到時就刪除，快速釋放掉不必要的內存佔用

缺點

CPU壓力很大，無論CPU此時負載多高，均佔用CPU，會影響redis服務器響應時間和指令吞吐量

總結

用處理器性能換取存儲空間

惰性刪除

數據到達過期時間，不做處理。等下次訪問該數據，如果未過期，返回數據。發現已經過期，刪除，返回不存在。這樣每次讀寫數據都需要檢測數據是否已經到達過期時間。也就是惰性刪除總是在數據的讀寫時發生的。

expireIfNeeded函數

對所有的讀寫命令進行檢查，檢查操作的對象是否過期。過期就刪除返回過期，不過期就什麼也不做～。

執行數據寫入過程中，首先通過expireIfNeeded函數對寫入的key進行過期判斷。

/*
 * 為執行寫入操作而取出鍵 key 在數據庫 db 中的值。
 *
 * 和 lookupKeyRead 不同，這個函數不會更新服務器的命中/不命中信息。
 *
 * 找到時返回值對象，沒找到返回 NULL 。
 */
robj *lookupKeyWrite(redisDb *db, robj *key) {

    // 刪除過期鍵
    expireIfNeeded(db,key);

    // 查找並返回 key 的值對象
    return lookupKey(db,key);
}

執行數據讀取過程中，首先通過expireIfNeeded函數對寫入的key進行過期判斷。

/*
 * 為執行讀取操作而取出鍵 key 在數據庫 db 中的值。
 *
 * 並根據是否成功找到值，更新服務器的命中/不命中信息。
 *
 * 找到時返回值對象，沒找到返回 NULL 。
 */
robj *lookupKeyRead(redisDb *db, robj *key) {
    robj *val;

    // 檢查 key 釋放已經過期
    expireIfNeeded(db,key);

    // 從數據庫中取出鍵的值
    val = lookupKey(db,key);

    // 更新命中/不命中信息
    if (val == NULL)
        server.stat_keyspace_misses++;
    else
        server.stat_keyspace_hits++;

    // 返回值
    return val;
}

執行過期動作expireIfNeeded其實內部做了三件事情，分別是：

• 查看key判斷是否過期
• 向slave節點傳播執行過期key的動作併發送事件通知
• 刪除過期key

/*
 * 檢查 key 是否已經過期，如果是的話，將它從數據庫中刪除。
 *
 * 返回 0 表示鍵沒有過期時間，或者鍵未過期。
 *
 * 返回 1 表示鍵已經因為過期而被刪除了。
 */
int expireIfNeeded(redisDb *db, robj *key) {

    // 取出鍵的過期時間
    mstime_t when = getExpire(db,key);
    mstime_t now;

    // 沒有過期時間
    if (when < 0) return 0; /* No expire for this key */

    /* Don't expire anything while loading. It will be done later. */
    // 如果服務器正在進行載入，那麼不進行任何過期檢查
    if (server.loading) return 0;

    // 當服務器運行在 replication 模式時
    // 附屬節點並不主動刪除 key
    // 它只返回一個邏輯上正確的返回值
    // 真正的刪除操作要等待主節點發來刪除命令時才執行
    // 從而保證數據的同步
    if (server.masterhost != NULL) return now > when;

    // 運行到這裏，表示鍵帶有過期時間，並且服務器為主節點

    /* Return when this key has not expired */
    // 如果未過期，返回 0
    if (now <= when) return 0;

    /* Delete the key */
    server.stat_expiredkeys++;

    // 向 AOF 文件和附屬節點傳播過期信息
    propagateExpire(db,key);

    // 發送事件通知
    notifyKeyspaceEvent(REDIS_NOTIFY_EXPIRED,
        "expired",key,db->id);

    // 將過期鍵從數據庫中刪除
    return dbDelete(db,key);
}

判斷key是否過期的數據結構是db->expires，也就是通過expires的數據結構判斷數據是否過期。
內部獲取過期時間並返回。

/*
 * 返回字典中包含鍵 key 的節點
 *
 * 找到返回節點，找不到返回 NULL
 *
 * T = O(1)
 */
dictEntry *dictFind(dict *d, const void *key)
{
    dictEntry *he;
    unsigned int h, idx, table;

    // 字典（的哈希表）為空
    if (d->ht[0].size == 0) return NULL; /* We don't have a table at all */

    // 如果條件允許的話，進行單步 rehash
    if (dictIsRehashing(d)) _dictRehashStep(d);

    // 計算鍵的哈希值
    h = dictHashKey(d, key);
    // 在字典的哈希表中查找這個鍵
    // T = O(1)
    for (table = 0; table <= 1; table++) {

        // 計算索引值
        idx = h & d->ht[table].sizemask;

        // 遍歷給定索引上的鏈表的所有節點，查找 key
        he = d->ht[table].table[idx];
        // T = O(1)
        while(he) {

            if (dictCompareKeys(d, key, he->key))
                return he;

            he = he->next;
        }

        // 如果程序遍歷完 0 號哈希表，仍然沒找到指定的鍵的節點
        // 那麼程序會檢查字典是否在進行 rehash ，
        // 然後才決定是直接返回 NULL ，還是繼續查找 1 號哈希表
        if (!dictIsRehashing(d)) return NULL;
    }

    // 進行到這裏時，說明兩個哈希表都沒找到
    return NULL;
}

優點

節約CPU性能，發現必須刪除的時候才刪除。

缺點

內存壓力很大，出現長期佔用內存的數據。

總結

用存儲空間換取處理器性能

定期刪除

周期性輪詢redis庫中時效性數據，採用隨機抽取的策略，利用過期數據佔比的方式刪除頻度。

優點

CPU性能佔用設置有峰值，檢測頻度可自定義設置

內存壓力不是很大，長期佔用內存的冷數據會被持續清理

缺點

需要周期性抽查存儲空間

定期刪除詳解

redis的定期刪除是通過定時任務實現的，也就是定時任務會循環調用serverCron方法。然後定時檢查過期數據的方法是databasesCron。定期刪除的一大特點就是考慮了定時刪除過期數據會佔用cpu時間，所以每次執行databasesCron的時候會限制cpu的佔用不超過25%。真正執行刪除的是 activeExpireCycle方法。

時間事件

對於持續運行的服務器來說， 服務器需要定期對自身的資源和狀態進行必要的檢查和整理， 從而讓服務器維持在一個健康穩定的狀態， 這類操作被統稱為常規操作（cron job）

在 Redis 中， 常規操作由 redis.c/serverCron() 實現， 它主要執行以下操作

1 更新服務器的各類統計信息，比如時間、內存佔用、數據庫佔用情況等。

2 清理數據庫中的過期鍵值對。

3 對不合理的數據庫進行大小調整。

4 關閉和清理連接失效的客戶端。

5 嘗試進行 AOF 或 RDB 持久化操作。

6 如果服務器是主節點的話，對附屬節點進行定期同步。

7 如果處於集群模式的話，對集群進行定期同步和連接測試。

因為 serverCron() 需要在 Redis 服務器運行期間一直定期運行， 所以它是一個循環時間事件： serverCron() 會一直定期執行，直到服務器關閉為止。

在 Redis 2.6 版本中， 程序規定 serverCron() 每秒運行 10 次， 平均每 100 毫秒運行一次。 從 Redis 2.8 開始， 用戶可以通過修改 hz選項來調整 serverCron() 的每秒執行次數， 具體信息請參考 redis.conf 文件中關於 hz 選項的說明

查看hz

way1 : config get hz  # "hz" "10"
way2 : info server  # server.hz 10

serverCron()

serverCron()會定期的執行，在serverCron()執行中會調用databasesCron() 方法(serverCron()還做了其他很多事情，但是現在不討論，只談刪除策略)

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    // 略去多無關代碼

    /* We need to do a few operations on clients asynchronously. */
    // 檢查客戶端，關閉超時客戶端，並釋放客戶端多餘的緩衝區
    clientsCron();

    /* Handle background operations on Redis databases. */
    // 對數據庫執行各種操作
    databasesCron();   /* ！我們關注的方法！ */

databasesCron()

在 databasesCron() 中 調用了 activeExpireCycle()方法，來對過期的數據進行處理。(在這裏還會做一些其他操作～ 調整數據庫大小，主動和漸進式rehash)

// 對數據庫執行刪除過期鍵，調整大小，以及主動和漸進式 rehash
void databasesCron(void) {

    // 判斷是否是主服務器 如果是 執行主動過期鍵清除
    if (server.active_expire_enabled && server.masterhost == NULL)
        // 清除模式為 CYCLE_SLOW ，這個模式會盡量多清除過期鍵
        activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);

    // 在沒有 BGSAVE 或者 BGREWRITEAOF 執行時，對哈希表進行 rehash
    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) {
        static unsigned int resize_db = 0;
        static unsigned int rehash_db = 0;
        unsigned int dbs_per_call = REDIS_DBCRON_DBS_PER_CALL;
        unsigned int j;

        /* Don't test more DBs than we have. */
        // 設定要測試的數據庫數量
        if (dbs_per_call > server.dbnum) dbs_per_call = server.dbnum;

        /* Resize */
        // 調整字典的大小
        for (j = 0; j < dbs_per_call; j++) {
            tryResizeHashTables(resize_db % server.dbnum);
            resize_db++;
        }

        /* Rehash */
        // 對字典進行漸進式 rehash
        if (server.activerehashing) {
            for (j = 0; j < dbs_per_call; j++) {
                int work_done = incrementallyRehash(rehash_db % server.dbnum);
                rehash_db++;
                if (work_done) {
                    /* If the function did some work, stop here, we'll do
                     * more at the next cron loop. */
                    break;
                }
            }
        }
    }
}

activeExpireCycle()

大致流程如下

1 遍歷指定個數的db（默認的 16 ）進行刪除操作

2 針對每個db隨機獲取過期數據每次遍歷不超過指定數量（如20），發現過期數據並進行刪除。

3 如果有多於25%的keys過期，重複步驟 2

除了主動淘汰的頻率外，Redis對每次淘汰任務執行的最大時長也有一個限定，這樣保證了每次主動淘汰不會過多阻塞應用請求，以下是這個限定計算公式：

#define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25 /* CPU max % for keys collection */ ``... ``timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;

也就是每次執行時間的25%用於過期數據刪除。

void activeExpireCycle(int type) {
    // 靜態變量，用來累積函數連續執行時的數據
    static unsigned int current_db = 0; /* Last DB tested. */
    static int timelimit_exit = 0;      /* Time limit hit in previous call? */
    static long long last_fast_cycle = 0; /* When last fast cycle ran. */

    unsigned int j, iteration = 0;
    // 默認每次處理的數據庫數量
    unsigned int dbs_per_call = REDIS_DBCRON_DBS_PER_CALL;
    // 函數開始的時間
    long long start = ustime(), timelimit;

    // 快速模式
    if (type == ACTIVE_EXPIRE_CYCLE_FAST) {
        // 如果上次函數沒有觸發 timelimit_exit ，那麼不執行處理
        if (!timelimit_exit) return;
        // 如果距離上次執行未夠一定時間，那麼不執行處理
        if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return;
        // 運行到這裏，說明執行快速處理，記錄當前時間
        last_fast_cycle = start;
    }

    /* 
     * 一般情況下，函數只處理 REDIS_DBCRON_DBS_PER_CALL 個數據庫，
     * 除非：
     *
     * 1) 當前數據庫的數量小於 REDIS_DBCRON_DBS_PER_CALL
     * 2) 如果上次處理遇到了時間上限，那麼這次需要對所有數據庫進行掃描，
     *     這可以避免過多的過期鍵佔用空間
     */
    if (dbs_per_call > server.dbnum || timelimit_exit)
        dbs_per_call = server.dbnum;

    // 函數處理的微秒時間上限
    // ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 默認為 25 ，也即是 25 % 的 CPU 時間
    timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;
    timelimit_exit = 0;
    if (timelimit <= 0) timelimit = 1;

    // 如果是運行在快速模式之下
    // 那麼最多只能運行 FAST_DURATION 微秒 
    // 默認值為 1000 （微秒）
    if (type == ACTIVE_EXPIRE_CYCLE_FAST)
        timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */

    // 遍曆數據庫
    for (j = 0; j < dbs_per_call; j++) {
        int expired;
        // 指向要處理的數據庫
        redisDb *db = server.db+(current_db % server.dbnum);

        // 為 DB 計數器加一，如果進入 do 循環之後因為超時而跳出
        // 那麼下次會直接從下個 DB 開始處理
        current_db++;

        do {
            unsigned long num, slots;
            long long now, ttl_sum;
            int ttl_samples;

            /* If there is nothing to expire try next DB ASAP. */
            // 獲取數據庫中帶過期時間的鍵的數量
            // 如果該數量為 0 ，直接跳過這個數據庫
            if ((num = dictSize(db->expires)) == 0) {
                db->avg_ttl = 0;
                break;
            }
            // 獲取數據庫中鍵值對的數量
            slots = dictSlots(db->expires);
            // 當前時間
            now = mstime();

            // 這個數據庫的使用率低於 1% ，掃描起來太費力了（大部分都會 MISS）
            // 跳過，等待字典收縮程序運行
            if (num && slots > DICT_HT_INITIAL_SIZE &&
                (num*100/slots < 1)) break;

            /* 
             * 樣本計數器
             */
            // 已處理過期鍵計數器
            expired = 0;
            // 鍵的總 TTL 計數器
            ttl_sum = 0;
            // 總共處理的鍵計數器
            ttl_samples = 0;

            // 每次最多只能檢查 LOOKUPS_PER_LOOP 個鍵
            if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP)
                num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP;

            // 開始遍曆數據庫
            while (num--) {
                dictEntry *de;
                long long ttl;

                // 從 expires 中隨機取出一個帶過期時間的鍵
                if ((de = dictGetRandomKey(db->expires)) == NULL) break;
                // 計算 TTL
                ttl = dictGetSignedIntegerVal(de)-now;
                // 如果鍵已經過期，那麼刪除它，並將 expired 計數器增一
                if (activeExpireCycleTryExpire(db,de,now)) expired++;
                if (ttl < 0) ttl = 0;
                // 累積鍵的 TTL
                ttl_sum += ttl;
                // 累積處理鍵的個數
                ttl_samples++;
            }

            /* Update the average TTL stats for this database. */
            // 為這個數據庫更新平均 TTL 統計數據
            if (ttl_samples) {
                // 計算當前平均值
                long long avg_ttl = ttl_sum/ttl_samples;
                
                // 如果這是第一次設置數據庫平均 TTL ，那麼進行初始化
                if (db->avg_ttl == 0) db->avg_ttl = avg_ttl;
                /* Smooth the value averaging with the previous one. */
                // 取數據庫的上次平均 TTL 和今次平均 TTL 的平均值
                db->avg_ttl = (db->avg_ttl+avg_ttl)/2;
            }

            // 我們不能用太長時間處理過期鍵，
            // 所以這個函數執行一定時間之後就要返回

            // 更新遍歷次數
            iteration++;

            // 每遍歷 16 次執行一次
            if ((iteration & 0xf) == 0 && /* check once every 16 iterations. */
                (ustime()-start) > timelimit)
            {
                // 如果遍歷次數正好是 16 的倍數
                // 並且遍歷的時間超過了 timelimit
                // 那麼斷開 timelimit_exit
                timelimit_exit = 1;
            }

            // 已經超時了，返回
            if (timelimit_exit) return;

            // 如果已刪除的過期鍵占當前總數據庫帶過期時間的鍵數量的 25 %
            // 那麼不再遍歷
        } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4);
    }
}

hz調大將會提高Redis主動淘汰的頻率，如果你的Redis存儲中包含很多冷數據佔用內存過大的話，可以考慮將這個值調大，但Redis作者建議這個值不要超過100。我們實際線上將這個值調大到100，觀察到CPU會增加2%左右，但對冷數據的內存釋放速度確實有明顯的提高（通過觀察keyspace個數和used_memory大小）。

可以看出timelimit和server.hz是一個倒數的關係，也就是說hz配置越大，timelimit就越小。換句話說是每秒鐘期望的主動淘汰頻率越高，則每次淘汰最長佔用時間就越短。這裏每秒鐘的最長淘汰佔用時間是固定的250ms（1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/100），而淘汰頻率和每次淘汰的最長時間是通過hz參數控制的。

因此當redis中的過期key比率沒有超過25%之前，提高hz可以明顯提高掃描key的最小個數。假設hz為10，則一秒內最少掃描200個key（一秒調用10次*每次最少隨機取出20個key），如果hz改為100，則一秒內最少掃描2000個key；另一方面，如果過期key比率超過25%，則掃描key的個數無上限，但是cpu時間每秒鐘最多佔用250ms。

當REDIS運行在主從模式時，只有主結點才會執行上述這兩種過期刪除策略，然後把刪除操作”del key”同步到從結點。

if (server.active_expire_enabled && server.masterhost == NULL)  // 判斷是否是主節點 從節點不需要執行activeExpireCycle()函數。
        // 清除模式為 CYCLE_SLOW ，這個模式會盡量多清除過期鍵
        activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);

隨機個數

redis.config.ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP 決定每次循環從數據庫 expire中隨機挑選值的個數

逐出算法

如果不限制 reids 對內存使用的限制，它將會使用全部的內存。可以通過 config.memory 來指定redis 對內存的使用量 。

下面是redis 配置文件中的說明

 543 # Set a memory usage limit to the specified amount of bytes.
 544 # When the memory limit is reached Redis will try to remove keys
 545 # according to the eviction policy selected (see maxmemory-policy).
 546 #
 547 # If Redis can't remove keys according to the policy, or if the policy is
 548 # set to 'noeviction', Redis will start to reply with errors to commands
 549 # that would use more memory, like SET, LPUSH, and so on, and will continue
 550 # to reply to read-only commands like GET.
 551 #
 552 # This option is usually useful when using Redis as an LRU or LFU cache, or to
 553 # set a hard memory limit for an instance (using the 'noeviction' policy).
 554 #
 555 # WARNING: If you have replicas attached to an instance with maxmemory on,
 556 # the size of the output buffers needed to feed the replicas are subtracted
 557 # from the used memory count, so that network problems / resyncs will
 558 # not trigger a loop where keys are evicted, and in turn the output
 559 # buffer of replicas is full with DELs of keys evicted triggering the deletion
 560 # of more keys, and so forth until the database is completely emptied.
 561 #
 562 # In short... if you have replicas attached it is suggested that you set a lower
 563 # limit for maxmemory so that there is some free RAM on the system for replica
 564 # output buffers (but this is not needed if the policy is 'noeviction').
 
將內存使用限制設置為指定的字節。當已達到內存限制Redis將根據所選的逐出策略（請參閱maxmemory策略）嘗試刪除數據。

如果Redis無法根據逐出策略移除密鑰，或者策略設置為“noeviction”，Redis將開始對使用更多內存的命令（如set、LPUSH等）進行錯誤回復，並將繼續回復只讀命令，如GET。

當將Redis用作LRU或LFU緩存或設置實例的硬內存限制（使用“noeviction”策略）時，此選項通常很有用。

警告：如果將副本附加到啟用maxmemory的實例，則將從已用內存計數中減去饋送副本所需的輸出緩衝區的大小，這樣，網絡問題/重新同步將不會觸發收回密鑰的循環，而副本的輸出緩衝區將充滿收回的密鑰增量，從而觸發刪除更多鍵，依此類推，直到數據庫完全清空。

簡而言之。。。如果附加了副本，建議您設置maxmemory的下限，以便系統上有一些空閑RAM用於副本輸出緩衝區（但如果策略為“noeviction”，則不需要此限制）。

驅逐策略的配置

Maxmemery-policy volatile-lru

當前已用內存超過 maxmemory 限定時，觸發主動清理策略

易失數據清理

volatile-lru：只對設置了過期時間的key進行LRU（默認值）

volatile-random：隨機刪除即將過期key

volatile-ttl ： 刪除即將過期的

volatile-lfu：挑選最近使用次數最少的數據淘汰

全部數據清理

allkeys-lru ： 刪除lru算法的key

allkeys-lfu：挑選最近使用次數最少的數據淘汰

allkeys-random：隨機刪除

禁止驅逐

(Redis 4.0 默認策略)

noeviction ： 永不過期，返回錯誤當mem_used內存已經超過maxmemory的設定，對於所有的讀寫請求都會觸發redis.c/freeMemoryIfNeeded(void)函數以清理超出的內存。注意這個清理過程是阻塞的，直到清理出足夠的內存空間。所以如果在達到maxmemory並且調用方還在不斷寫入的情況下，可能會反覆觸發主動清理策略，導致請求會有一定的延遲。

清理時會根據用戶配置的maxmemory-policy來做適當的清理（一般是LRU或TTL），這裏的LRU或TTL策略並不是針對redis的所有key，而是以配置文件中的maxmemory-samples個key作為樣本池進行抽樣清理。

maxmemory-samples在redis-3.0.0中的默認配置為5，如果增加，會提高LRU或TTL的精準度，redis作者測試的結果是當這個配置為10時已經非常接近全量LRU的精準度了，並且增加maxmemory-samples會導致在主動清理時消耗更多的CPU時間，建議：

1 盡量不要觸發maxmemory，最好在mem_used內存佔用達到maxmemory的一定比例后，需要考慮調大hz以加快淘汰，或者進行集群擴容。

2 如果能夠控制住內存，則可以不用修改maxmemory-samples配置；如果Redis本身就作為LRU cache服務（這種服務一般長時間處於maxmemory狀態，由Redis自動做LRU淘汰），可以適當調大maxmemory-samples。

這裏提一句，實際上redis根本就不會準確的將整個數據庫中最久未被使用的鍵刪除，而是每次從數據庫中隨機取5個鍵並刪除這5個鍵里最久未被使用的鍵。上面提到的所有的隨機的操作實際上都是這樣的，這個5可以用過redis的配置文件中的maxmemeory-samples參數配置。

數據逐出策略配置依據

使用INFO命令輸出監控信息，查詢緩存int和miss的次數，根據業務需求調優Redis配置。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※別再煩惱如何寫文案,掌握八大原則!

※教你寫出一流的銷售文案?

※超省錢租車方案

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※產品缺大量曝光嗎?你需要的是一流包裝設計!