有一本小冊子Static Single Assignment Book寫的很好，內容又較少，試着翻譯一下，意譯較多（說人話），不是嚴肅的翻譯，感興趣的可以看看。頻率可能是周更。。anyway，stay tuned～

在日常編程中，名字是一個很有用的東西。這本書想傳遞的關鍵內容是對於每個不同的東西給它一個獨一無二的名字可以消除很多不確定性以及不精確性。

舉個例子，如果你無意中聽到一段對話中有’Homer’這個詞，沒有上下文的情況下你不知道他說的是Homer Simpson（辛普森）還是古希臘詩人荷馬還是你認識的某個叫Homer的人。但是只要你聽到對話提及Springfield (辛普森一家）而不是Smyrna（希臘詩歌），你就能知道他們說的是辛普森一家這個電視劇。不過話又說回來，如果每個人都有一個獨一無二的名字，那麼就不可能混淆電視劇角色和古希臘文學人物，這個問題都不會成立。

這本書主要討論靜態單賦值（Static Single Assignment Form，SSA）形式，它是一種變量的命名約定。術語static說明SSA與屬性和代碼分析相關，術語single說明SSA強制變量名具有唯一性。術語assignment表示變量的定義。舉個例子，在下面的代碼中：

x = y + 1;

變量x被賦予表達式(y+1)的值。這是一個定義，或者對於x來說是賦值語句。編譯器工程師會說上面的賦值語句將值(y+1)儲存到左值x中。

1.1 SSA定義

關於SSA最簡單，限制最少的定義如下：

“如果每個變量在程序中有且只有一個賦值語句，那麼該程序是SSA形式”

但是實際上SSA還有很多變體，有更多的限制。這些變體可能使變量定義和使用與圖論的一些特性有關，或者封裝一些特定的控制流/數據流信息。每個SSA變體都有特設的性質。基本的SSA變體將會在第二章討論，本書的第三部分還會討論更多這部分的內容。

所有SSA變體，包括上面最簡單的定義都有一個最基本屬性就是引用透明性（referential transparency），所謂引用透明性是指程序中的每個變量只有一個定義，變量的值和它所在程序的位置無關。我們可能根據分支的條件完善對於某個變量的認識。舉個例子，不用看代碼我們就知道下面if語句后緊跟着的then/else條件塊中x的值

if (x == 0)

因為x的值在這個if語句中是沒有改變的。函數式編程語言寫的程序是引用透明的，引用透明性對於形式化方法和數學推理很有用，因為表達式的值只依賴它的子表達式而不依賴求值的順序或者表達式的副作用，或者其它表達式。對於一個引用透明的程序，考慮下面的代碼片段：

x = 1;
y = x + 1;
x = 2;
z = x + 1;

一個naive（而且不正確）的分析器可能認為y和z的值相等，因為他們的定義是一樣的，都是(x+1)，然而x的值取決於當前代碼位置是在第二個賦值的前面還是後面，即變量的值取決於上下文。當編譯器將這個代碼段轉換為SSA形式時，它會具有引用透明性。轉換的過程會為一個變量的多次定義使用不同的名字（譯註：x1和x2）。使用SSA形式后，只有當x1和x2相等時y和z才相等。

x1 = 1;
y = x1 + 1;
x2 = 2;
z = x2 + 1;

1.2 SSA的非形式化語義

在前一節中，我們看到了如何通過簡單的重命名將代碼轉換為SSA形式。賦值語句左邊被定義的變量叫做target，在SSA中，每個target都有唯一的名字。反過來，賦值語句右邊可以多次使用target，在這裏它們叫做source。貫穿本書，SSA的target名字定義都是變量名再加一個下標這種形式。一般來說這是不重要的實現細節，雖然它對於編譯器debug來說很有用。

φ函數是SSA最重要的一個概念，它很特別，又叫做偽賦值函數（pseudo-assignment function）。有些人也叫它notational fiction。ɸ函數的用途是合併來自不同路徑的值，一般出現在控制流的合併點。

考慮下面的代碼示例和它對應的控制流圖（Control Flow Graph，CFG）表示：

在if不同分支中，y有不同的定義。y的不同定義最終在print那個地方交匯。當編譯器將該代碼轉換為SSA形式時，y的不同定義被命名為y1和y2。print既可以使用y1也可以使用y2，這取決於if的條件。在這種情況下，需要用φ函數引入新的變量y3，它的參數是y1和y2。因此SSA版本的上述程序如下：

就放置位置來說，φ函數一般是放到控制流交匯點，即CFG中有多個前驅基本塊的那個基本塊頭部。如果有n條路徑可以進入基本塊b，那麼在基本塊b頭部的φ函數有n個參數。φ函數會動態的選擇正確的參數。φ函數根據n個參數，創建新的變量名，這個名字是唯一的，因為它要保證SSA的基本性質。因此，在上面的例子中，如果控制流從基本塊A流向下面的基本塊，那麼y3使用φ函數選擇y1作為它的值，反之φ函數使用y2作為它的值。注意CFG圖φ函數的參數y1和y2前面還加了基本塊的標籤，這種形式是比較多餘的，在本書的後面部分，這個基本塊標籤能不加就不加，除非沒了它會引起歧義。

這裏還要強調一下，如果基本塊頭部有多個φ函數，這些φ函數是并行的，即，它們是同時執行，不需要順序執行。這一點是很重要的，因為在經過一些優化，比如複寫傳播（copy propagation）后φ函數的target可能是其它φ函數的source。在SSA解構階段，φ函數會被消除（譯註：就是編譯器不需要SSA形式，想將它轉換為其它IR，這就叫SSA解構），在解構階段使用常規的複製操作序列化，這點會在17.6小結描述。這個小細節對於寄存器分配后的代碼來說是相當重要的。

嚴格來說，φ函數不能被軟件直接執行，因為進入φ函數的控制流沒有被顯式的編碼進φ函數的參數。這是可以接受的，因為φ函數通常只用於程序的靜態分析。然而，有很多擴展使得φ函數可以執行，如 φif 或者γ函數（參見第12章），它有一個額外的參數，告訴φ函數選擇那個值。關於這個會在第12章，第16章和第18章討論。

接下來我們再展示一個例子，它說明了一個循環控制流解構的SSA形式。下面是非SSA形式的程序和SSA形式的控制流圖：

SSA代碼在循環頭部新增了兩個φ函數。它們合併循環前的值定義和循環中的值的定義。

要注意不要混淆SSA和自動并行化優化中的（動態）靜態賦值這兩個概念。SSA不會阻止在程序執行的時候對一個變量的多次定義，比如，上面的SSA代碼中，變量y3和x3在循環體內，每次循環都會重定義它們。

SSA構造的詳細描述會在第3章給出，現在只需要明白下面的內容：

1. 如果程序的交匯點的某個變量有多個定義，那麼會在交匯點插入φ函數
2. 整數下標用於重命名原來程序中的變量x和y

1.3 與傳統數據流分析的比較

在未來的第11章我們會提到，SSA主要的一個優點是它對數據流分析（data-flow analysis）很友好。數據流分析在程序編譯的時候收集信息，為未來的代碼優化做準備。在程序運行時，這些信息會在變量間流動。靜態分析通過在控制流圖中傳播這些信息，得以捕獲關於數據流的一些事實（fact）。這種方式在傳統的數據流分析中很常見。

通常，如果程序是一種功能性的（functional）或者稀疏（sparse）的表示，如SSA形式，那麼數據流信息能程序中更高效的傳播。當程序被轉換為SSA形式時，變量在定義點被重命名。對於一個確鑿的數據流問題，比如常量傳播，它表現為一個程序點的集合，在這些程序點數據流事實可能改變。因此可以直接關聯數據流事實和變量名字，而不是在每個程序點為所有變量維護各自的數據流事實的集合，下圖展示了一個非零值分析（non-zero value analysis）

對於程序中的每個變量，分析的目標是靜態確定哪些變量在運行時包含0值（即null）。在這裏0就表示變量為null，0打一把叉表示不為null，T表示可能為null。上圖(a)表示傳統的數據流分析，我們會在六個基本塊的入口點和出處都計算一次變量x和y信息。而在上圖(b)的基於SSA的數據流分析中，我們只需要在變量定義處計算一下，然後就能獲得六個數據流事實。

對於其它的數據流問題，屬性也可能在變量定義之外發生改變，這些問題只要插入一些φ函數就能放入稀疏數據流分析的框架中，第11章會有一個例子討論這個。總的來說，目前這個例子說明了SSA能給分析算法帶來的關鍵好處是：

1. 數據流信息直接從定義語句處傳播到使用它的地方，即通過def-use鏈，這個鏈條由SSA命名方式隱示給出。相反，傳統的數據流分析需要將信息傳遍整個程序，即便在很多地方這些信息都沒改變，或者不相關。
2. 基於SSA的數據流分析更簡潔。在示例中，比起傳統方式，基於SSA的分析只有很少的數據流事實。

這本書的第二部分給出了一個完整的基於SSA數據流分析的描述。

1.4 此情此景此SSA

歷史背景。在整個20世紀80年代，優化編譯器技術越來越成熟，各種中間表示被提出，它們包含了數據依賴，使得數據流分析在這些中間表示上很容易進行。在這些中間表示背後的設計理念是顯式/隱式包含變量定義和使用的關係，即def-use鏈條，使得數據流信息能有效的傳播。程序依賴圖（program dependence graph）和程序依賴網（program dependence web）均屬此類IR。第12章還會討論這些風格的IR的更多細節。

靜態單賦值是由IBM Research開發的一種IR，並在20世紀80年代末的幾篇研究論文中公開發表。SSA由於其符合直覺的性質和直觀的構造算法得到了廣泛的應用。SSA給出了一個標準化的變量def-use鏈，簡化了很多數據流分析技術。

當前狀況。當前主流的商業編譯器和開源編譯器，包括GCC，LLVM，HotSpot Java虛擬機，V8 JavaScript引擎都將SSA作為程序分析中的關鍵表示。由於在SSA執行優化速度快而且高效，那些即時編譯器（JIT）會在一些高級地、與平台無關的表示（如Java字節碼，CLI字節碼，LLVM bitcode）上廣泛使用SSA。
SSA最初是為了簡化高級程序表示的變形而開發而創建的，因為其良好的特性，能夠簡化算法和減少計算複雜性。今天，SSA形式甚至被用於最後的代碼生成階段（見第四部分），即後端。好幾個工業編譯器和學術編譯器，既有靜態，也有just-in-time，都在它們的後端使用SSA，如LLVM，HotSpot，LAO，libFirm，Mono。很多使用SSA的編譯器在編譯快要結束時，即寄存器分配前才解構SSA。最近的研究甚至能在寄存器分配期間也使用SSA，SSA形式會保持到非常非常後面的機器代碼生成過程才會被解構。

SSA與高級語言。到目前為止，我們展示了在低級代碼上使用SSA形式做分析的優勢。有趣的是，在高級代碼上如果強制遵循某些準則也可能具有SSA的性質。根據SISAL語言的定義，程序自動具備引用透明性，因為變量不允許多次賦值。其它語言也能有SSA的性質，比如Java的變量加個final或者C#的變量加個const/readonly。

強制寫出具有SSA性質的高級語言程序主要好處是這些程序能具備不變形，這簡化了併發編程。豬肚的數據能在多個線程中自由的共享，沒有任何數據依賴問題。數據依賴對於多核處理器來說是一個大問題。

在函數式編程語言中，引用透明是語言的基本特性。因此函數式編程隱式具有SSA性質。第6章會介紹SSA和函數式編程。

1.5 餘下本章

本章引入了SSA的符號表示，本書的剩下部分就SSA的各個方面詳細討論。本書的終極目標是：

1. 清晰的描述SSA能為程序分析帶來哪些好處
2. 消除那些阻止人們使用SSA的謬誤

本節還剩下一些內容，它們與下一章的一些主題相關。

1.5.1 SSA的好處

SSA對於變量命名有嚴格要求，每個變量的名字都是獨一無二的。賦值語言和控制流交匯點會引入新的變量名。這些簡化了表達變量def-use關係的數據結構實現和變量存活範圍。本書第二部分關注基於SSA的數據流恩熙，使用SSA主要有三個好處：

編譯時受益。如果程序是SSA形式，很多編譯器優化可以高效的進行，因為引用透明性意味着數據流信息直接與變量關聯，而不是每個程序點的變量。關於這一點我們已經在1.3的非零值分析中演示過了。

編譯器開發受益。SSA使得程序分析和轉換能更容易表達。這意味着編譯器工程師能更高產，可以寫更多的pass，並且能debug更多的pass（譯註：smile）。舉個例子，基於SSA的GCC4.x的死代碼優化比非GCC3.x的非SSA死代碼優化實現總代碼少了40%。

程序運行時受益。理論上，能基於SSA實現的分析和優化也能基於其它非SSA形式。前一點提到過，基於SSA的實現代碼更少，因此很多基於SSA的編譯器優化也能更高效進行，關於這一點的示例是一類控制流不敏感分析（control-flow insensitive analysis），具體參見論文Using static single assignment form to improve flowinsensitive pointer analysis

1.5.2 SSA謬論

一些人認為SSA很複雜很繁瑣，不能高效表達程序。這本書的目的就是讓讀者免去這些擔憂。下面的表單展示了關於SSA常見的謬論，以及破除謬論的章節。

謬論	破除謬論
SSA讓變量數爆炸	第二章會回顧SSA的主要變體，一些變體引入的變量數比原始SSA形式少很多
SSA的性質難以維持	第三章和第五章討論了一些修復SSA性質的簡單技術（因為一些優化可能重寫中間表示，導致SSA性質被破壞）
SSA的性質難以維持	第三章和第十七章展示了高效且效果顯著的SSA解構算法的複製操作

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

【其他文章推薦】

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

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

※想知道最厲害的網頁設計公司"嚨底家"!

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

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

前言

在 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

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

【其他文章推薦】

※超省錢租車方案

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

※回頭車貨運收費標準

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

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

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

前言

寫這篇博客主要目的是記錄一下自己的學習過程，只能是簡單入門級別的，因為水平有限就寫到哪算哪吧，寫的不對之處歡迎指正。
代碼放在：https://github.com/xiajingren/NetCoreMicroserviceDemo

什麼是微服務？

關於微服務的概念解釋網上有很多…
個人理解，微服務是一種系統架構模式，它和語言無關，和框架無關，和工具無關，和服務器環境無關…
微服務思想是將傳統的單體系統按照業務拆分成多個職責單一、且可獨立運行的接口服務。至於服務如何拆分，沒有明確的定義。
幾乎任何後端語言都能做微服務開發。
微服務也並不是完美無缺的，微服務架構會帶來更多的問題，增加系統的複雜度，引入更多的技術棧…

創建項目

一個客戶端，一個產品服務，一個訂單服務。3個項目都是asp.net core web應用程序。創建項目的時候記得啟用一下Docker支持，或者後面添加也行。

為產品、訂單服務添加一些基礎代碼，就簡單的返回一下 服務名稱，當前時間，服務的ip、端口。

在Docker中運行服務

為了方便，我使用Docker來運行服務，不用Docker也行，關於docker的安裝及基本使用就不介紹了。

• build鏡像：

在項目根目錄打開PowerShell窗口執行：docker build -t productapi -f ./Product.API/Dockerfile .

Successfully代表build成功了。

• 運行容器:

執行：docker run -d -p 9050:80 --name productservice productapi

執行：docker ps查看運行的容器：

沒問題，使用瀏覽器訪問一下接口：

也沒問題，其中的ip端口是Docker容器內部的ip端口，所以端口是80，這個無所謂。

• 產品服務部署好了，下面部署一下訂單服務，也是同樣的流程，就把指令簡單貼一下吧：

build鏡像：docker build -t orderapi -f ./Order.API/Dockerfile .
運行容器：docker run -d -p 9060:80 --name orderservice orderapi
瀏覽器訪問一下：

OK，訂單服務也部署完成了。

客戶端調用

客戶端我這裏只做了一個web客戶端，實際可能是各種業務系統、什麼PC端、手機端、小程序。。。這個明白就好，為了簡單就不搞那麼多了。

• 因為客戶端需要http請求服務端接口，所以需要一個http請求客戶端，我個人比較習慣RestSharp，安利一波：https://github.com/restsharp/RestSharp

• 添加基礎代碼：

IServiceHelper.cs：

    public interface IServiceHelper
    {
        /// <summary>
        /// 獲取產品數據
        /// </summary>
        /// <returns></returns>
        Task<string> GetProduct();

        /// <summary>
        /// 獲取訂單數據
        /// </summary>
        /// <returns></returns>
        Task<string> GetOrder();
    }

ServiceHelper.cs：

    public class ServiceHelper : IServiceHelper
    {
        public async Task<string> GetOrder()
        {
            string serviceUrl = "http://localhost:9060";//訂單服務的地址，可以放在配置文件或者數據庫等等...

            var Client = new RestClient(serviceUrl);
            var request = new RestRequest("/orders", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }

        public async Task<string> GetProduct()
        {
            string serviceUrl = "http://localhost:9050";//產品服務的地址，可以放在配置文件或者數據庫等等...

            var Client = new RestClient(serviceUrl);
            var request = new RestRequest("/products", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }
    }

Startup.cs：

    public class Startup
    {
        public Startup(IConfiguration configuration)
        {
            Configuration = configuration;
        }

        public IConfiguration Configuration { get; }

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            services.AddControllersWithViews();
            
            //注入IServiceHelper
            services.AddSingleton<IServiceHelper, ServiceHelper>();
        }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }
            else
            {
                app.UseExceptionHandler("/Home/Error");
            }
            app.UseStaticFiles();

            app.UseRouting();

            app.UseAuthorization();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapControllerRoute(
                    name: "default",
                    pattern: "{controller=Home}/{action=Index}/{id?}");
            });
        }
    }

HomeController.cs：

    public class HomeController : Controller
    {
        private readonly ILogger<HomeController> _logger;
        private readonly IServiceHelper _serviceHelper;

        public HomeController(ILogger<HomeController> logger, IServiceHelper serviceHelper)
        {
            _logger = logger;
            _serviceHelper = serviceHelper;
        }

        public async Task<IActionResult> Index()
        {
            ViewBag.OrderData = await _serviceHelper.GetOrder();
            ViewBag.ProductData = await _serviceHelper.GetProduct();

            return View();
        }

        public IActionResult Privacy()
        {
            return View();
        }

        [ResponseCache(Duration = 0, Location = ResponseCacheLocation.None, NoStore = true)]
        public IActionResult Error()
        {
            return View(new ErrorViewModel { RequestId = Activity.Current?.Id ?? HttpContext.TraceIdentifier });
        }
    }

Index.cshtml：

@{
    ViewData["Title"] = "Home Page";
}

<div class="text-center">
    <h1 class="display-4">Welcome</h1>
    <p>
        @ViewBag.OrderData
    </p>
    <p>
        @ViewBag.ProductData
    </p>
</div>

代碼比較簡單，這裏就不用docker了，直接控制台啟動，使用瀏覽器訪問：

• 一切正常。進行到這裏，各個服務也獨立運行了，客戶端也能正常調用了，貌似算是完成一個簡易的微服務了。但是，微服務架構最重要的原則就是——“高可用”。以上的做法明顯不能滿足高可用性，因為任何一個服務掛掉，所有依賴這個服務的業務系統都會受影響。

停止一下訂單服務：docker stop orderservice

訂單服務停止，導致客戶端業務系統無法獲取訂單數據。
要解決這個問題，很容易想到：集群。

簡單的服務集群

既然單個服務實例有掛掉的風險，那麼部署多個服務實例就好了嘛，只要大家不同時全掛就行。

• 使用docker運行多個服務實例：

docker run -d -p 9061:80 --name orderservice1 orderapi
docker run -d -p 9062:80 --name orderservice2 orderapi
docker run -d -p 9051:80 --name productservice1 productapi
docker run -d -p 9052:80 --name productservice2 productapi

現在訂單服務和產品服務都增加到3個服務實例。

• 那麼稍微改造一下客戶端代碼吧：
  ServiceHelper.cs：

public class ServiceHelper : IServiceHelper
    {
        public async Task<string> GetOrder()
        {
            string[] serviceUrls = { "http://localhost:9060", "http://localhost:9061", "http://localhost:9062" };//訂單服務的地址，可以放在配置文件或者數據庫等等...

            //每次隨機訪問一個服務實例
            var Client = new RestClient(serviceUrls[new Random().Next(0, 3)]);
            var request = new RestRequest("/orders", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }

        public async Task<string> GetProduct()
        {
            string[] serviceUrls = { "http://localhost:9050", "http://localhost:9051", "http://localhost:9052" };//產品服務的地址，可以放在配置文件或者數據庫等等...

            //每次隨機訪問一個服務實例
            var Client = new RestClient(serviceUrls[new Random().Next(0, 3)]);
            var request = new RestRequest("/products", Method.GET);

            var response = await Client.ExecuteAsync(request);
            return response.Content;
        }
    }

當然拿到這些服務地址可以自己做複雜的負載均衡策略，比如輪詢，隨機，權重等等 都行，甚至在中間弄個nginx也可以。這些不是重點，所以就簡單做一個隨機吧，每次請求來了隨便訪問一個服務實例。

• 瀏覽器測試一下：

  可以看到請求被隨機分配了。但是這種做法依然不安全，如果隨機訪問到的實例剛好掛掉，那麼業務系統依然會出問題。
  簡單處理思路是：
  1.如果某個地址請求失敗了，那麼換一個地址接着執行。
  2.如果某個地址的請求連續多次失敗了，那麼就移除這個地址，下次就不會訪問到它了。
  。。。。。。
  業務系統實現以上邏輯，基本上風險就很低了，也算是大大增加了系統可用性了。

• 然後思考另一個問題：

實際應用中，上層的業務系統可能非常多，為了保證可用性，每個業務系統都去考慮服務實例掛沒掛掉嗎？
而且實際應用中服務實例的數量或者地址大多是不固定的，例如雙十一來了，流量大了，增加了一堆服務實例，這時候每個業務系統再去配置文件里配置一下這些地址嗎？雙十一過了又去把配置刪掉嗎？顯然是不現實的，服務必須要做到可靈活伸縮。

• 這時候就引入一個名詞：服務註冊與發現

未完待續…

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

【其他文章推薦】

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

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

※想知道最厲害的網頁設計公司"嚨底家"!

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

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

數據存儲和有效期

在 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配置。

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

【其他文章推薦】

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

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

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

※超省錢租車方案

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

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

系列文章

1. 基於 abp vNext 和 .NET Core 開發博客項目 – 使用 abp cli 搭建項目
2. 基於 abp vNext 和 .NET Core 開發博客項目 – 給項目瘦身，讓它跑起來
3. 基於 abp vNext 和 .NET Core 開發博客項目 – 完善與美化，Swagger登場
4. 基於 abp vNext 和 .NET Core 開發博客項目 – 數據訪問和代碼優先
5. 基於 abp vNext 和 .NET Core 開發博客項目 – 自定義倉儲之增刪改查
6. 基於 abp vNext 和 .NET Core 開發博客項目 – 統一規範API，包裝返回模型
7. 基於 abp vNext 和 .NET Core 開發博客項目 – 再說Swagger，分組、描述、小綠鎖
8. 基於 abp vNext 和 .NET Core 開發博客項目 – 接入GitHub，用JWT保護你的API
9. 基於 abp vNext 和 .NET Core 開發博客項目 – 異常處理和日誌記錄
10. 基於 abp vNext 和 .NET Core 開發博客項目 – 使用Redis緩存數據
11. 基於 abp vNext 和 .NET Core 開發博客項目 – 集成Hangfire實現定時任務處理
12. 基於 abp vNext 和 .NET Core 開發博客項目 – 用AutoMapper搞定對象映射
13. 基於 abp vNext 和 .NET Core 開發博客項目 – 定時任務最佳實戰（一）
14. 基於 abp vNext 和 .NET Core 開發博客項目 – 定時任務最佳實戰（二）
15. 基於 abp vNext 和 .NET Core 開發博客項目 – 定時任務最佳實戰（三）
16. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇（一）
17. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇（二）
18. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇（三）
19. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇（四）
20. 基於 abp vNext 和 .NET Core 開發博客項目 – 博客接口實戰篇（五）
21. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（一）
22. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（二）
23. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（三）
24. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（四）
25. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（五）
26. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（六）
27. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（七）
28. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（八）
29. 基於 abp vNext 和 .NET Core 開發博客項目 – Blazor 實戰系列（九）
30. 基於 abp vNext 和 .NET Core 開發博客項目 – 終結篇之發布項目

上一篇完成了博客的分頁查詢文章列表頁面的數據綁定和分頁功能，本篇將繼續完成剩下的幾個頁面。

在開始主題之前重新解決上一篇的最後一個問題，當點擊了頭部組件的/posts鏈接時直接強制刷新了頁面，經過查看文檔和實踐有了更好的解決方案。

先將頭部組件Header.razor中的NavLink恢復成<NavLink class="menu-item" href="posts">Posts</NavLink>，不需要點擊事件了。

然後在Posts.razor中添加生命周期函數OnParametersSetAsync()，在初始化完成后執行。

/// <summary>
/// 初始化完成后執行
/// </summary>
/// <returns></returns>
protected override async Task OnParametersSetAsync()
{
    if (!page.HasValue)
    {
        page = 1;
        await RenderPage(page);
    }
}

判斷當前page參數是否有值，有值的話說明請求肯定是來自於翻頁，當page沒有值的時候就說明是頭部的菜單點進來的。那麼此時給page賦值為1，調用API加載數據即可。

分類列表

Categories.razor是分類列表頁面，上篇文章已經實現了從API獲取數據的方法，所以這裏就很簡單了，指定接受類型，然後在生命周期初始化OnInitializedAsync()中去獲取數據。

@code{
    /// <summary>
    /// categories
    /// </summary>
    private ServiceResult<IEnumerable<QueryCategoryDto>> categories;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // 獲取數據
        categories = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryCategoryDto>>>($"/blog/categories");
    }
}

當獲取到數據的時候進行綁定，沒有數據的時候還是显示加載中的組件<Loading />讓他轉圈圈。

@if (categories == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap categories">
            <h2 class="post-title">-&nbsp;Categories&nbsp;-</h2>
            <div class="categories-card">
                @if (categories.Success && categories.Result.Any())
                {
                    @foreach (var item in categories.Result)
                    {
                        <div class="card-item">
                            <div class="categories">
                                <a href="/category/@item.DisplayName/">
                                    <h3>
                                        <i class="iconfont iconcode" style="padding-right:3px"></i>
                                        @item.CategoryName
                                    </h3>
                                    <small>(@item.Count)</small>
                                </a>
                            </div>
                        </div>
                    }
                }
                else
                {
                    <ErrorTip />
                }
            </div>
        </div>
    </div>
}

直接循環返回的數據列表categories.Result，綁定數據就好，當獲取失敗或者沒有返回數據的時候显示錯誤提示組件<ErrorTip />

標籤列表

Categories.razor是標籤列表頁面，和分類列表HTML結構差不多一樣的，除了返回類型和接口地址不一樣，將上面代碼複製過來改改即可。

@code{
    /// <summary>
    /// tags
    /// </summary>
    private ServiceResult<IEnumerable<QueryTagDto>> tags;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // 獲取數據
        tags = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryTagDto>>>($"/blog/tags");
    }
}

@if (tags == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap tags">
            <h2 class="post-title">-&nbsp;Tags&nbsp;-</h2>
            <div class="tag-cloud-tags">
                @if (tags.Success && tags.Result.Any())
                {
                    @foreach (var item in tags.Result)
                    {
                        <a href="/tag/@item.DisplayName/">@item.TagName<small>(@item.Count)</small></a>
                    }
                }
                else
                {
                    <ErrorTip />
                }
            </div>
        </div>
    </div>
}

友鏈列表

FriendLinks.razor是友情鏈接列表頁面，實現方式和上面兩個套路一模一樣。

@code {
    /// <summary>
    /// friendlinks
    /// </summary>
    private ServiceResult<IEnumerable<FriendLinkDto>> friendlinks;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // 獲取數據
        friendlinks = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<FriendLinkDto>>>($"/blog/friendlinks");
    }
}

@if (friendlinks == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap categories">
            <h2 class="post-title">-&nbsp;FriendLinks&nbsp;-</h2>
            <div class="categories-card">
                @if (friendlinks.Success && friendlinks.Result.Any())
                {
                    @foreach (var item in friendlinks.Result)
                    {
                        <div class="card-item">
                            <div class="categories">
                                <a target="_blank" href="@item.LinkUrl">
                                    <h3>@item.Title</h3>
                                </a>
                            </div>
                        </div>
                    }
                }
                else
                {
                    <ErrorTip />
                }
            </div>
        </div>
    </div>
}

文章列表(分類)

Posts.Category.razor是根據分類查詢文章列表頁面，他接受一個參數name，我們要根據name去API查詢數據然後綁定頁面即可。

這裏的參數name實際上就是從標籤列表傳遞過來的DisplayName的值，它是一個比較友好的名稱，我們還要通過這個值去查詢真正的分類名稱進行展示，所以這裏需要調用兩個API，這點在設計API的時候沒有考慮好，我們其實可以將這兩個API合併變成一個，後續再進行優化吧，這裏就請求兩次。

添加兩個接收參數：分類名稱和返回的文章列表數據。

/// <summary>
/// 分類名稱
/// </summary>
private string categoryName;

/// <summary>
/// 文章列表數據
/// </summary>
private ServiceResult<IEnumerable<QueryPostDto>> posts;

然後在OnInitializedAsync()初始化方法中調用API獲取數據，賦值給變量。

/// <summary>
/// 初始化
/// </summary>
protected override async Task OnInitializedAsync()
{
    // TODO:獲取數據,可以在API中合併這兩個請求。
    var category = await Http.GetFromJsonAsync<ServiceResult<string>>($"/blog/category?name={name}");
    posts = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryPostDto>>>($"/blog/posts/category?name={name}");

    if (category.Success)
    {
        categoryName = category.Result;
    }
}

有了數據，直接在頁面上進行循環綁定。

@if (posts == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap tags">
            @if (categoryName != null)
            {
                <h2 class="post-title">-&nbsp;Category&nbsp;·&nbsp;@categoryName&nbsp;-</h2>
            }
        </div>
        <div class="post-wrap archive">
            @if (posts.Success && posts.Result.Any())
            {
                @foreach (var item in posts.Result)
                {
                    <h3>@item.Year</h3>
                    @foreach (var post in item.Posts)
                    {
                        <article class="archive-item">
                            <NavLink href="@("/post"+post.Url)">@post.Title</NavLink>
                            <span class="archive-item-date">@post.CreationTime</span>
                        </article>
                    }
                }
            }
            else
            {
                <ErrorTip />
            }
        </div>
    </div>
}

文章列表(標籤)

Posts.Tag.razor是根據標籤查詢文章列表，這個和分類查詢文章列表實現方式一樣，直接上代碼。

@code {
    /// <summary>
    /// 標籤名稱參數
    /// </summary>
    [Parameter]
    public string name { get; set; }

    /// <summary>
    /// 標籤名稱
    /// </summary>
    private string tagName;

    /// <summary>
    /// 文章列表數據
    /// </summary>
    private ServiceResult<IEnumerable<QueryPostDto>> posts;

    /// <summary>
    /// 初始化
    /// </summary>
    protected override async Task OnInitializedAsync()
    {
        // TODO:獲取數據,可以在API中合併這兩個請求。
        var tag = await Http.GetFromJsonAsync<ServiceResult<string>>($"/blog/tag?name={name}");
        posts = await Http.GetFromJsonAsync<ServiceResult<IEnumerable<QueryPostDto>>>($"/blog/posts/tag?name={name}");

        if (tag.Success)
        {
            tagName = tag.Result;
        }
    }
}

@if (posts == null)
{
    <Loading />
}
else
{
    <div class="container">
        <div class="post-wrap tags">
            @if (tagName != null)
            {
                <h2 class="post-title">-&nbsp;Tag&nbsp;·&nbsp;@tagName&nbsp;-</h2>
            }
        </div>
        <div class="post-wrap archive">
            @if (posts.Success && posts.Result.Any())
            {
                @foreach (var item in posts.Result)
                {
                    <h3>@item.Year</h3>
                    @foreach (var post in item.Posts)
                    {
                        <article class="archive-item">
                            <NavLink href="@("/post"+post.Url)">@post.Title</NavLink>
                            <span class="archive-item-date">@post.CreationTime</span>
                        </article>
                    }
                }
            }
            else
            {
                <ErrorTip />
            }
        </div>
    </div>
}

以上完成了以上幾個頁面的數據綁定，頁面之間的跳轉已經關聯起來了，然後還剩下文章詳情頁，大家可以先自己動手完成它，今天就到這裏，未完待續…

開源地址：https://github.com/Meowv/Blog/tree/blog_tutorial

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

【其他文章推薦】

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

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

※超省錢租車方案

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

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

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

目錄

• 簡介
• 虛擬地址空間
• 詳解MappedByteBuffer
  • MapMode
• MappedByteBuffer的最大值
• MappedByteBuffer的使用
• MappedByteBuffer要注意的事項
• 總結

簡介

大大大，我要大！小師妹要讀取的文件越來越大，該怎麼幫幫她，讓程序在性能和速度上面得到平衡呢？快來跟F師兄一起看看吧。

虛擬地址空間

小師妹：F師兄，你有沒有發現，最近硬盤的價格真的是好便宜好便宜，1T的硬盤大概要500塊，平均1M五毛錢。現在下個電影都1G起步，這是不是意味着我們買入了大數據時代？

沒錯，小師妹，硬件技術的進步也帶來了軟件技術的進步，兩者相輔相成，缺一不可。

小師妹：F師兄，如果要是去讀取G級的文件，有沒有什麼快捷簡單的方法？

更多精彩內容且看：

• 區塊鏈從入門到放棄系列教程-涵蓋密碼學,超級賬本,以太坊,Libra,比特幣等持續更新
• Spring Boot 2.X系列教程:七天從無到有掌握Spring Boot-持續更新
• Spring 5.X系列教程:滿足你對Spring5的一切想象-持續更新
• java程序員從小工到專家成神之路（2020版）-持續更新中,附詳細文章教程

還記得上次我們講的虛擬地址空間嗎？

再把上次講的圖搬過來：

通常來說我們的應用程序調用系統的接口從磁盤空間獲取Buffer數據，我們把自己的應用程序稱之為用戶空間，把系統的底層稱之為系統空間。

傳統的IO操作，是操作系統講磁盤中的文件讀入到系統空間裏面，然後再拷貝到用戶空間中，供用戶使用。

這中間多了一個Buffer拷貝的過程，如果這個量夠大的話，其實還是挺浪費時間的。

於是有人在想了，拷貝太麻煩太耗時了，我們單獨劃出一塊內存區域，讓系統空間和用戶空間同時映射到同一塊地址不就省略了拷貝的步驟嗎？

這個被劃出來的單獨的內存區域叫做虛擬地址空間，而不同空間到虛擬地址的映射就叫做Buffer Map。 Java中是有一個專門的MappedByteBuffer來代表這種操作。

小師妹：F師兄，那這個虛擬地址空間和內存有什麼區別呢？有了內存還要啥虛擬地址空間？

虛擬地址空間有兩個好處。

第一個好處就是虛擬地址空間對於應用程序本身而言是獨立的，從而保證了程序的互相隔離和程序中地址的確定性。比如說一個程序如果運行在虛擬地址空間中，那麼它的空間地址是固定的，不管他運行多少次。如果直接使用內存地址，那麼可能這次運行的時候內存地址可用，下次運行的時候內存地址不可用，就會導致潛在的程序出錯。

第二個好處就是虛擬空間地址可以比真實的內存地址大，這個大其實是對內存的使用做了優化，比如說會把很少使用的內存寫如磁盤，從而釋放出更多的內存來做更有意義的事情，而之前存儲到磁盤的數據，當真正需要的時候，再從磁盤中加載到內存中。

這樣物理內存實際上可以看做虛擬空間地址的緩存。

詳解MappedByteBuffer

小師妹：MappedByteBuffer聽起來好神奇，怎麼使用它呢？

我們先來看看MappedByteBuffer的定義：

public abstract class MappedByteBuffer
    extends ByteBuffer

它實際上是一個抽象類，具體的實現有兩個：

class DirectByteBuffer extends MappedByteBuffer implements DirectBuffer

class DirectByteBufferR extends DirectByteBuffer
implements DirectBuffer

分別是DirectByteBuffer和DirectByteBufferR。

小師妹：F師兄，這兩個ByteBuffer有什麼區別呢？這個R是什麼意思？

R代表的是ReadOnly的意思，可能是因為本身是個類的名字就夠長了，所以搞了個縮寫。但是也不寫個註解，讓人看起來十分費解….

我們可以從RandomAccessFile的FilChannel中調用map方法獲得它的實例。

我們看下map方法的定義：

 public abstract MappedByteBuffer map(MapMode mode, long position, long size)
        throws IOException;

MapMode代表的是映射的模式，position表示是map開始的地址，size表示是ByteBuffer的大小。

MapMode

小師妹：F師兄，文件有隻讀，讀寫兩種模式，是不是MapMode也包含這兩類？

對的，其實NIO中的MapMode除了這兩個之外，還有一些其他很有趣的用法。

• FileChannel.MapMode.READ_ONLY 表示只讀模式
• FileChannel.MapMode.READ_WRITE 表示讀寫模式
• FileChannel.MapMode.PRIVATE 表示copy-on-write模式，這個模式和READ_ONLY有點相似，它的操作是先對原數據進行拷貝，然後可以在拷貝之後的Buffer中進行讀寫。但是這個寫入並不會影響原數據。可以看做是數據的本地拷貝，所以叫做Private。

基本的MapMode就這三種了，其實除了基礎的MapMode，還有兩種擴展的MapMode：

• ExtendedMapMode.READ_ONLY_SYNC 同步的讀
• ExtendedMapMode.READ_WRITE_SYNC 同步的讀寫

MappedByteBuffer的最大值

小師妹：F師兄，既然可以映射到虛擬內存空間，那麼這個MappedByteBuffer是不是可以無限大？

當然不是了，首先虛擬地址空間的大小是有限制的，如果是32位的CPU，那麼一個指針佔用的地址就是4個字節，那麼能夠表示的最大值是0xFFFFFFFF，也就是4G。

另外我們看下map方法中size的類型是long，在java中long能夠表示的最大值是0x7fffffff，也就是2147483647字節，換算一下大概是2G。也就是說MappedByteBuffer的最大值是2G，一次最多只能map 2G的數據。

MappedByteBuffer的使用

小師妹，F師兄我們來舉兩個使用MappedByteBuffer讀寫的例子吧。

善！

先看一下怎麼使用MappedByteBuffer來讀數據：

public void readWithMap() throws IOException {
        try (RandomAccessFile file = new RandomAccessFile(new File("src/main/resources/big.www.flydean.com"), "r"))
        {
            //get Channel
            FileChannel fileChannel = file.getChannel();
            //get mappedByteBuffer from fileChannel
            MappedByteBuffer buffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileChannel.size());
            // check buffer
            log.info("is Loaded in physical memory: {}",buffer.isLoaded());  //只是一個提醒而不是guarantee
            log.info("capacity {}",buffer.capacity());
            //read the buffer
            for (int i = 0; i < buffer.limit(); i++)
            {
                log.info("get {}", buffer.get());
            }
        }
    }

然後再看一個使用MappedByteBuffer來寫數據的例子：

public void writeWithMap() throws IOException {
        try (RandomAccessFile file = new RandomAccessFile(new File("src/main/resources/big.www.flydean.com"), "rw"))
        {
            //get Channel
            FileChannel fileChannel = file.getChannel();
            //get mappedByteBuffer from fileChannel
            MappedByteBuffer buffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 4096 * 8 );
            // check buffer
            log.info("is Loaded in physical memory: {}",buffer.isLoaded());  //只是一個提醒而不是guarantee
            log.info("capacity {}",buffer.capacity());
            //write the content
            buffer.put("www.flydean.com".getBytes());
        }
    }

MappedByteBuffer要注意的事項

小師妹：F師兄，MappedByteBuffer因為使用了內存映射，所以讀寫的速度都會有所提升。那麼我們在使用中應該注意哪些問題呢？

MappedByteBuffer是沒有close方法的，即使它的FileChannel被close了，MappedByteBuffer仍然處於打開狀態，只有JVM進行垃圾回收的時候才會被關閉。而這個時間是不確定的。

總結

本文再次介紹了虛擬地址空間和MappedByteBuffer的使用。

本文的例子https://github.com/ddean2009/learn-java-io-nio

本文作者：flydean程序那些事

本文鏈接：http://www.flydean.com/io-nio-mappedbytebuffer/

本文來源：flydean的博客

歡迎關注我的公眾號:程序那些事，更多精彩等着您！

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

【其他文章推薦】

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

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

※回頭車貨運收費標準

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

※超省錢租車方案

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