Python 中的方法解析順序（Method Resolution Order, MRO）定義了多繼承存在時 Python 解釋器查找函數解析的正確方式。當 Python 版本從 2.2 發展到 2.3 再到現在的 Python 3，MRO算法也隨之發生了相應的變化。這種變化在很多時候影響了我們使用不同版本 Python 編程的過程。

大部分內容轉載自C3 線性化算法與 MRO 理解Python中的多繼承

什麼是 MRO

MRO 全稱方法解析順序（Method Resolution Order）。它定義了 Python 中多繼承存在的情況下，解釋器查找函數解析的具體順序。什麼是函數解析順序？我們首先用一個簡單的例子來說明。請仔細看下面代碼：

class A():
    def who_am_i(self):
        print("I am A")
        
class B(A):
    pass
        
class C(A):
    def who_am_i(self):
        print("I am C")

class D(B,C):
    pass
    
d = D()

如果我問在 Python 2 中使用 D 的實例調用 d.who_am_i()，究竟執行的是 A 中的 who_am_i() 還是 C 中的 who_am_i()，我想百分之九十以上的人都會不假思索地回答：肯定是 C 中的 who_am_i()，因為 C 是 D 的直接父類。然而，如果你把代碼用 Python 2 運行一下就可以看到 d.who_am_i() 打印的是 I am A。

是不是覺得很混亂很奇怪？感到奇怪就對了！！！

這個例子充分展示了 MRO 的作用：決定基類中的函數到底應該以什麼樣的順序調用父類中的函數。可以明確地說，Python 發展到現在，MRO 算法已經不是一個憑藉著執行結果就能猜出來的算法了。如果沒有深入到 MRO 算法的細節，稍微複雜一點的繼承關係和方法調用都能徹底繞暈你。

New-style Class vs. Old-style Class

在介紹不同版本的 MRO 算法之前，我們有必要簡單地回顧一下 Python 中類定義方式的發展歷史。儘管在 Python 3 中已經廢除了老式的類定義方式和 MRO 算法，但對於仍然廣泛使用的 Python 2 來說，不同的類定義方式與 MRO 算法之間具有緊密的聯繫。了解這一點將幫助我們從 Python 2 向 Python 3 遷移時不會出現莫名其妙的錯誤。

在 Python 2.1 及以前，我們定義一個類的時候往往是這個樣子（我們把這種類稱為 old-style class)：

class A:
    def __init__(self):
        pass

Python 2.2 引入了新的模型對象（new-style class），其建議新的類型通過如下方式定義：

class A(object):
    def __init__(self):
        pass

注意后一種定義方式显示註明類 A 繼承自 object。Python 2.3 及後續版本為了保持向下兼容，同時提供以上兩種類定義用以區分 old-style class 和 new-style class。Python 3 則完全廢棄了 old-style class 的概念，不論你通過以上哪種方式書寫代碼，Python 3 都將明確認為類 A 繼承自 object。這裏我們只是引入 old-style 和 new-style 的概念，如果你對他們的區別感興趣，可以自行看 stackoverflow 上有關該問題的解釋。

理解 old-style class 的 MRO

我們使用前文中的類繼承關係來介紹 Python 2 中針對 old-style class 的 MRO 算法。如果你在前面執行過那段代碼，你可以看到調用 d.who_am_i() 打印的應該是 I am A。為什麼 Python 2 的解釋器在確定 D 中的函數調用時要先搜索 A 而不是先搜索 D 的直接父類 C 呢？

這是由於 Python 2 對於 old-style class 使用了非常簡單的基於深度優先遍歷的 MRO 算法（關於深度優先遍歷，我想大家肯定都不陌生）。當一個類繼承自多個類時，Python 2 按照從左到右的順序深度遍歷類的繼承圖，從而確定類中函數的調用順序。這個過程具體如下：

1. 檢查當前的類裏面是否有該函數，如果有則直接調用。
2. 檢查當前類的第一個父類裏面是否有該函數，如果沒有則檢查父類的第一個父類是否有該函數，以此遞歸深度遍歷。
3. 如果沒有則回溯一層，檢查下一個父類裏面是否有該函數並按照 2 中的方式遞歸。

上面的過程與標準的深度優先遍歷只有一點細微的差別：步驟 2 總是按照繼承列表中類的先後順序來選擇分支的遍歷順序。具體來說，類 D 的繼承列表中類順序為 B, C，因此，類 D 按照先遍歷 B 分支再遍歷 C 分支的順序來確定 MRO。

我們繼續用第一個例子中的函數繼承圖來說明這個過程：

按照上述深度遞歸的方式，函數 d.who_am_i() 調用的搜索順序是 D, B, A, C, A。由於一個類不能兩次出現，因此在搜索路徑中去除掉重複出現的 A，得到最終的方法解析順序是 D, B, A, C。這樣一來你就明白了為什麼 d.who_am_i() 打印的是 I am A 了。

在 Python 2 中，我們可以通過如下方式來查看 old-style class 的 MRO：

>>> import inspect
>>> inspect.getmro(D)

理解 new-style class 的 MRO

從上面的結果可以看到，使用深度優先遍歷的查找算法並不合理。因此，Python 3 以及 Python 2 針對 new-style class 採用了新的 MRO 算法。如果你使用 Python 3 重新運行一遍上述腳本，你就可以看到函數 d.who_am_i() 的打印結果是 I am C。

>>> d.who_am_i()
I am C
>>> D.__mro__
(<class 'test.D'>, <class 'test.B'>, <class 'test.C'>, <class 'test.A'>, <class 'object'>)

新算法與基於深度遍歷的算法類似，但是不同在於新算法會對深度優先遍歷得到的搜索路徑進行額外的檢查。其從左到右掃描得到的搜索路徑，對於每一個節點解釋器都會判斷該節點是不是好的節點。如果不是好的節點，那麼將其從當前的搜索路徑中移除。

那麼問題在於，什麼是一個好的節點？我們說 N 是一個好的節點當且僅當搜索路徑中 N 之後的節點都不繼承自 N。我們還以上述的類繼承圖為例，按照深度優先遍歷得到類 D 中函數的搜索路徑 D, B, A, C, A。之後 Python 解釋器從左向右檢查時發現第三個節點 A 不是一個好的節點，因為 A 之後的節點 C 繼承自 A。因此其將 A 從搜索路徑中移除，然後得到最後的調用順序 D, B, C, A。

採用上述算法，D 中的函數調用將優先查找其直接父類 B 和 C 中的相應函數。

C3線性化算法

上一小結我們從直觀上概述了針對 new-style class 的 MRO 算法過程。事實上這個算法有一個明確的名字 C3 linearization。下面我們給出其形式化的計算過程。

上面的過程看起來好像很複雜，我們用一個例子來具體執行一下，你就會覺得其實還是挺簡單的。假設我們有如下的一個類繼承關係：

參考來源：Understanding Python MRO – Class search path

class X():
    def who_am_i(self):
        print("I am a X")
        
class Y():
    def who_am_i(self):
        print("I am a Y")
        
class A(X, Y):
    def who_am_i(self):
        print("I am a A")
        
class B(Y, X):
     def who_am_i(self):
         print("I am a B")
         
class F(A, B):
    def who_am_i(self):
        print("I am a F")

Traceback (most recent call last):
  File "test.py", line 17, in <module>
    class F(A, B):
TypeError: Cannot create a consistent method resolution
order (MRO) for bases X, Y

為什麼採用C3算法

上圖中都是使用BFS算法來尋找繼承鏈，但是都會有問題，左邊的繼承模式會違背單調性的原則，右邊的棱形繼承鏈，如果是C重寫了繼承於A的方法，B沒有，但是根據MRO繼承鏈，最終調用的都是A類的方法，C中實現的方法永遠不會被調用，這些都是再python2的問題，python引入C3算法后就解決了這些問題。

C3算法最早被提出是用於Lisp的，應用在Python中是為了解決原來基於深度優先搜索算法不滿足本地優先級，和單調性的問題。
本地優先級：指聲明時父類的順序，比如C(A,B)，如果訪問C類對象屬性時，應該根據聲明順序，優先查找A類，然後再查找B類。
單調性：如果在C的解析順序中，A排在B的前面，那麼在C的所有子類里，也必須滿足這個順序。

在Python官網的The Python 2.3 Method Resolution Order中作者舉了例子，說明這一情況

F=type('Food', (), {remember2buy:'spam'})
E=type('Eggs', (F,), {remember2buy:'eggs'})
G=type('GoodFood', (F,E), {})

根據本地優先級在調用G類對象屬性時應該優先查找F類，而在Python2.3之前的算法給出的順序是G E F O，而在心得C3算法中通過阻止類層次不清晰的聲明來解決這一問題，以上聲明在C3算法中就是非法的。

小結

C3算法的核心 ：

1. 遍歷執行merge操作的序列，如果一個序列的第一個元素，在其他序列中也是第一個元素，或不在其他序列出現，則從所有執行merge操作序列中刪除這個元素，合併到當前的mro中。

2. merge操作后的序列，繼續執行merge操作，直到merge操作的序列為空。

3. 如果merge操作的序列無法為空，則說明不合法。

參考資料

理解Python中的多繼承-C3 線性化算法

Python的多重繼承問題-MRO和C3算法

Deep Thoughts by Raymond Hettinger

C3 linearization

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

【其他文章推薦】

※超省錢租車方案

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

※回頭車貨運收費標準

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

※FB行銷專家,教你從零開始的技巧

 1 def loadTestsFromTestCase(self, testCaseClass) #看名稱分析：從TestCase找測試集--那麼就是把我們的def用例加載到testSuit裏面
 2 def loadTestsFromModule(self, module, *args, pattern=None, **kws) #看名稱分析：從模塊裏面找測試集，那麼 模塊>類>test_方法>添加到testSuit裏面
 3 def loadTestsFromName(self, name, module=None)  #看名稱分析： 接收到name直接添加到testSuit裏面
 4 def loadTestsFromNames(self, names, module=None) #看名稱分析：接受到的是一個包含測試test_方法的列表
 5 def getTestCaseNames(self, testCaseClass) # 看名稱分析：  取出一個包含test_方法的列表
 6 def discover(self, start_dir, pattern='test*.py', top_level_dir=None) ## 看名稱分析：發現--找test_方法
 7 def _get_directory_containing_module(self, module_name) #獲取目錄包含的模塊
 8 def _get_name_from_path(self, path)  #從路徑從找名稱
 9 def _get_module_from_name(self, name)  #從名稱找模塊
10 def _match_path(self, path, full_path, pattern)   #正則匹配路徑--參數包含pattern 那估計是匹配我們測試腳本格式的
11 def _find_tests(self, start_dir, pattern, namespace=False)  #找測試集合
12 def _find_test_path(self, full_path, pattern, namespace=False)  #找測試集合的路徑

View Code

 1 那就是1234
 2 一個discover,getTest,_match_path
 3 二個find
 4 三個_get
 5 四個loadTests
 6 
 7 discover  邏輯
 8 >
 9 _find_tests【兩個處理邏輯  一個是本次傳的目錄和上次傳的一樣或不一樣，】
10 【一樣：直接從我們傳的目錄下面繼續去找testcaose---】
11 【不一樣：會從我們傳的目錄下面去執行os.path.listdir找到所有的子文件列表paths(文件))，然後遍歷得到單獨的path做 start_dir+path拼接】
12 >
13 ①—get_name_from_path【傳入start_dir,判斷當前傳入的目錄是否為上次傳入的頂級目錄返回".",不一樣可能有點繞-並返回一個值這個值有四種情況 . test  ...test dir.tests--正常應該是返回test文件名
14 ②_find_test_path(self, full_path, pattern, namespace=False)
15 【執行這個從路徑中找test，那麼很明顯 一樣：傳目錄路徑   不一樣傳文件路徑  】
16 _find_test_path

View Code

 1 class TestLoader(object):
 2     """
 3     This class is responsible for loading tests according to various criteria
 4     and returning them wrapped in a TestSuite
 5     """
 6     testMethodPrefix = 'test'
 7     sortTestMethodsUsing = staticmethod(util.three_way_cmp)
 8     suiteClass = suite.TestSuite
 9     _top_level_dir = None
10 
11     def __init__(self):
12         super(TestLoader, self).__init__()
13         self.errors = []
14         # Tracks packages which we have called into via load_tests, to
15         # avoid infinite re-entrancy.
16         self._loading_packages = set()  #這裏創建了一個空的self._loading_packages={}無序且不重複的元素集合

View Code 1.discover方法：unittest.defaultTestLoader a.定義了三個布爾值屬性 is_not_importable==True則是不能導入，is_namespace ，set_implicit_top b.對頂層目錄做了處理–當服務首次啟動 執行unittest.defaultTestLoader.discover(“文件目錄A”,pattern,top_level_dir=None):self._top_level_dir = top_level_dir = start_dir 這三個相等。 再次執行unittest.defaultTestLoader.discover(“文件目錄B”,pattern,top_level_dir=None): top_level_dir=self._top_level_dir【也就是他會默認上次start_dir 為頂層目錄】– 無論是首次還是複次–上面的操作完成之後 self._top_level_dir = top_level_dir 仍然繼續執行了一句這個—也就是說 self._top_level_dir == top_level_dir 始終一樣 c.針對頂層目錄不是一個目錄文件做了一系列的處理如果你傳的目錄是一個可導入的模塊-他在這個異常處理中.會重新自導入這個模塊。並開始追尋他的絕對路徑,判斷其模塊的可用性，然後執行_find_tests()尋找用例 d.如果是一個目錄就直接開始執行_find_tests()尋找用例 e.所以他這裏分兩種情況可以找到用例 第一種:傳的目錄 第二種:傳入的可導入模塊-這種情況self._top_level_dir 最終也是一個絕對路徑

  1 def discover(self, start_dir, pattern='test*.py', top_level_dir=None):  #一般我們top_level_dir傳的都None
  2     set_implicit_top = False   #是否存在頂層目錄
  3     if top_level_dir is None and self._top_level_dir is not None:
  4         # make top_level_dir optional if called from load_tests in a package
  5         top_level_dir = self._top_level_dir  #複次走這裏
  6     elif top_level_dir is None:  #初次走這裏
  7         set_implicit_top = True
  8         top_level_dir = start_dir
  9     #上面這一串花里胡哨的東西就是處理頂層目錄-如果是第一次啟動服務-
 10     #就走elif-top_level_dir==我們下面傳的值--之後--self._top_level_dir就不為空了，
 11     #但是top_level_dir  頂部是處理==None所以會走if=True
 12     top_level_dir = os.path.abspath(top_level_dir)#轉絕對路徑
 13     if not top_level_dir in sys.path:
 14         #這裡是防止重複將top_level_dir加入執行目錄--BUT如果我第一次傳的start_dir=a,第二次傳的start_dir=b
 15         #分析一下--第一次就是把a加入到了執行目錄---下面self._top_level_dir=a 二次(複次)傳b的時候,會出現top_level_dir=a---並沒有判斷b是否在執行目錄
 16         #這裏加一波問號？？？？？？？？？？？？？？？？？？？？？
 17         #但是一般情況 我們目錄就只有一個--所以這裏--先放着。。。先看後面再來看這裏
 18         # all test modules must be importable from the top level directory
 19         # should we *unconditionally* put the start directory in first
 20         # in sys.path to minimise likelihood of conflicts between installed
 21         # modules and development versions?
 22         sys.path.insert(0, top_level_dir)
 23     self._top_level_dir = top_level_dir
 24     #如果top_level_dir我們傳的目錄不在可執行目錄--則臨時添加進去
 25     is_not_importable = False  #是否  不能導入
 26     is_namespace = False    #is_namespace那麼這個字段的意思就是是否可以找到傳入的路徑
 27     tests = []
 28     if os.path.isdir(os.path.abspath(start_dir)):  #判斷我們傳的是否為一個目錄--實際上這裏直接用top_level_dir不香嗎--
 29         start_dir = os.path.abspath(start_dir)
 30         # 之前把top_level_dir = start_dir
 31         # 然後top_level_dir = os.path.abspath(top_level_dir)
 32         #現在start_dir = os.path.abspath(start_dir)
 33         #為什麼不   直接用top_level_dir？  小朋友你是否有許多問號
 34         # 問題出在上面--複次的時候--並沒有走 top_level_dir = start_dir 而是走的 top_level_dir = self._top_level_dir ，
 35         #所以如果我們上次傳的路徑如果和這次不一樣--那麼top_level_dir是不等於start_dir--而start_dir才是我們傳的--
 36         if start_dir != top_level_dir:  #所以這裏相當於判斷前後傳的路徑是否一樣--一般來說我們的start_dir都是等於top_level_dir的
 37             is_not_importable = not os.path.isfile(os.path.join(start_dir, '__init__.py'))#如果是一個文件返回false
 38             #如果不一樣--則判斷我們當前傳入的start_dir/__init__.py是不是一個正確的文件路徑..os.path.isfile() 返回布爾值
 39     else: #如果我們傳入的不是一個目錄,就開始一堆花里胡哨的報錯東西了。。暫時不用管
 40         # support for discovery from dotted module names
 41         try:
 42             __import__(start_dir)
 43             #這裏就很有意思了---__impor__("PyFiles.Besettest")那就是導入PyFiles
 44             #那麼也就是說這個97.33的概率會報錯--也就是說你如果目錄錯了--下面的else基本不會走。。。除非你很神奇的填的路徑右側是一個可導入的模塊
 45         except ImportError:
 46             is_not_importable = True   #如果導入不鳥--就is_not_importable設置為true  在這裏我清楚了這個字段的含義--不能導入=true
 47         else:#那麼這裏假設導入成功之後
 48             the_module = sys.modules[start_dir]  #這裡是如果我們導入成功--就走這裏-取出start_dir導入的賦值給the_module
 49             top_part = start_dir.split('.')[0]    #這裡是將我們導入的模塊名稱取出來
 50             try:
 51                 start_dir = os.path.abspath( #打印導入模塊所在目錄的絕對路徑
 52                    os.path.dirname((the_module.__file__)))
 53             except AttributeError:     #這裡是如果導入模塊成功了---但是尼瑪打印導入模塊的絕對路徑又報錯--不想看了+2
 54                 # look for namespace packages
 55                 try:  #然後有開始進行模塊導入檢查---日了狗了。。。。。
 56                     #  fuck----想直接關機了+1，這裏估計是想找到為什麼不能導入的原因。。大神的思路就是完美，如果是我就拋出一個目錄不對就完事
 57                     #這一塊的學習 文檔 Python標準模塊--import
 58                     spec = the_module.__spec__
 59                     #將導入成功的模塊的規格說明賦值給spec--
 60                     #打印出來就是ModuleSpec(name='besettest.interface', loader=<_frozen_importlib_external.SourceFileLoader object at 0x0000000003D6E780>, origin='E:\\PyFiles\\Besettest\\besettest\\interface\\__init__.py', submodule_search_locations=['E:\\PyFiles\\Besettest\\besettest\\interface'])
 61                     #這麼一串東西--也沒用過。。。。大概就是模塊名稱、路徑、導入的模塊對象吧
 62                     #origin 加載模塊的位置--
 63                     #loader_state模塊特定數據的容器
 64                 except AttributeError:  #如果模塊的規格說明取不出來。。。。。。。
 65                     spec = None   #我查閱了一下。。。的確存在部分模塊規格說明為None的--所以還得繼續往下看
 66 
 67                 if spec and spec.loader is None:   #如果存在規格說明 且 數據容器為None。
 68                     if spec.submodule_search_locations is not None: 
 69                         #這是個什麼玩意呢--模塊 搜索 位置s(列表)。。。
 70                         is_namespace = True    #如果spec.submodule_search_locations不為none -----
 71                         # 2.5級英文翻譯 就是模塊的路徑 如果模塊路徑不為空is_namespace可以找到---is_namespace那麼這個字段的意思就是存在命名空間。。就是可以找到這個模塊
 72                         for path in the_module.__path__:   #這裏我研究懷疑是故意提升逼格。。the_module.__path__==spec.submodule_search_locations
 73                             if (not set_implicit_top and     #首次set_implicit_top==True  複次set_implicit_top==Fase
 74                                 not path.startswith(top_level_dir)):
 75                                 continue
 76                                 #這裏讓我稍微有點疑惑。。為什麼要判斷是首次還是複次--我猜是判斷the_module.__path__列表裡面有幾個某塊的路徑
 77                                 #如果是首次直接下一步--如果是複次會有多個路徑。但是如果是複次top_level_dir這個路徑又是上次的。。。日
 78                                 #他的作用是找到導入模塊的路徑-知道這個就行。。。
 79                             self._top_level_dir = \
 80                                 (path.split(the_module.__name__
 81                                      .replace(".", os.path.sep))[0])   #取出導入模塊的上級目錄絕對路徑。。。
 82                             #the_module.__name__.replace(".", os.path.sep) 這一串我看來是沒有必要的。。因為  the_module.__name__既然取到了模塊名稱那他肯定是一個字符串
 83                             tests.extend(self._find_tests(path,     #然後調用_find_tests 尋找測試。加入tests列表--這個有點熟悉的味道---
 84                                                           pattern,  #我覺得基本不會走這裏去找---腳本路徑一般都會填對  填錯了，都不知道執行到哪裡去了。。。
 85                                                           namespace=True))
 86                 elif the_module.__name__ in sys.builtin_module_names:
 87                     #判斷 sys.builtin_module_names返回一個列表，包含所有已經編譯到Python解釋器里的模塊的名字 和sys.models是一個字典
 88                     #就是沒法導入報錯
 89                     # builtin module
 90                     raise TypeError('Can not use builtin modules '
 91                                     'as dotted module names') from None
 92                 else:  #沒發現這個模塊
 93                     raise TypeError(
 94                         'don\'t know how to discover from {!r}'
 95                         .format(the_module)) from None
 96 
 97             if set_implicit_top:     #如果是首次。。。。
 98                 if not is_namespace:  #is_namespace默認的是false-且可以找到模塊相關規格
 99                     self._top_level_dir = \
100                        self._get_directory_containing_module(top_part)  #interface.testFiles   interface假設這個是導入的 -self._top_level_dir 是一個目錄的絕對路徑
101                     #top_part導入的模塊名稱-----
102                     sys.path.remove(top_level_dir)    #只知道是從系統路徑移除--但是不知道為什麼移除。。。。
103                 else:
104                     sys.path.remove(top_level_dir)   #
105 
106     if is_not_importable:   #如果我們傳的文件不能導入---就直接拋出異常
107         raise ImportError('Start directory is not importable: %r' % start_dir)
108 
109     if not is_namespace:  #is_namespace默認的是false--這裏就是可以找到模塊。。。。
110         tests = list(self._find_tests(start_dir, pattern))
111     return self.suiteClass(tests)

View Code 2._find_tests()–尋找testCase並生成測試套件tests=[]

 1 def _find_tests(self, start_dir, pattern, namespace=False):   #注意這裏如果我們傳的不是腳本目錄而是一個可導入的模塊namespace是等於True的
 2     """Used by discovery. Yields test suites it loads."""
 3     # Handle the __init__ in this package
 4     name = self._get_name_from_path(start_dir)   #返回一個name   name存在三種返回情況  "."-當本次和上次傳入的start_dir一致     不一致  "文件名"    "...文件名"
 5      #get_name_from_path的邏輯在這裏就很清晰了 
 6           
 7     # name is '.' when start_dir == top_level_dir (and top_level_dir is by
 8     # definition not a package).
 9     if name != '.' and name not in self._loading_packages:  
10     #當name最少有一個且也不再self._loading_packages.【self._loading_packages初始化的時候建的空集合】 走下面這個
11         # name is in self._loading_packages while we have called into
12         # loadTestsFromModule with name.
13         tests, should_recurse = self._find_test_path(      #然後這裏start_dir是我們傳的模塊--他就去找。。這裏就恢復到了傳測試目錄的邏輯了
14             start_dir, pattern, namespace)
15         if tests is not None:
16             yield tests
17         if not should_recurse:
18             # Either an error occurred, or load_tests was used by the
19             # package.
20             return
21     # Handle the contents.
22     paths = sorted(os.listdir(start_dir))  #那就從這裏開始--當我們穿的目錄和上次一樣-他會找到目錄下所有的文件然後排序--我們的用例執行順序就是從這裏開始搞了。。
23     for path in paths:       #遍歷我們傳的目錄下的所有文件
24         full_path = os.path.join(start_dir, path)     將我們傳入的目錄和目錄下的py文件拼接的完整路徑
25         tests, should_recurse = self._find_test_path(    #把我們文件路徑和我們的文件格式傳入_find_test_path這個方法--
26             full_path, pattern, namespace)
27         如果當前傳的是一個目錄-會返回should_recurse=True--這個英文直譯是應該_遞歸--下面yield from 就是執行遞歸的操作
28         if tests is not None:
29             yield tests
30         if should_recurse:  #這句是判斷他是不是一個目錄
31             # we found a package that didn't use load_tests.
32             name = self._get_name_from_path(full_path)
33             self._loading_packages.add(name)
34             try:
35                 yield from self._find_tests(full_path, pattern, namespace)
36             finally:
37                 self._loading_packages.discard(name)

View Code yield與yield from

 1 def  a(n):
 2     testList=b(n)
 3     return testList
 4 
 5 def  b(n,m=1):
 6     print("執行第%s次"%m)
 7     for a in range(n):
 8         if not divmod(a,2)[1] and a!=0:
 9             print(a)
10             yield a   #是用yield之後返回的是一個生成器
11             if divmod(a,3)[1]:
12                 m =m+1
13                 yield from b(a,m)  #重新執行b方法
14 
15 print(list(a(7)))
16 
17 執行第1次
18 2
19 執行第2次
20 4
21 執行第3次
22 2
23 執行第4次
24 6
25 [2, 4, 2, 6]

View Code _get_name_from_path 他主要做了：從我們傳的路徑裏面找腳本文件 名。。。如果找到了腳本文件則返迴文件名稱—如果沒找到也就是我們返回一個點 或者 至少一個點（三種情況 . test_case …..test_case 返回name可能存在的三種值） 在_test_find調用這個方法path=start_dir(我們傳的目錄)–這個返回一個點 或者 至少一個點 在 _fin_test_path調用這個方法是傳的我們傳的目錄下的文件路口–返回的name就是文件名

 1 name = self._get_name_from_path(start_dir)    #因為discover我們是支持我們傳目錄或者模塊尋找testcase的，所以這個方法
 2 
 3 def _get_name_from_path(self, path):
 4      #主要正確邏輯三個 比如我們的腳本目錄結構是  E://a/b/     b目錄下面有script.py    和  /c/script.py
 5      #第一次是我們自己傳的目錄--之前在discover他做了一個處理 就是第一次運行時會把我們傳的目錄賦值給頂層目錄---
 6      #第一個邏輯判斷我們傳的是不是  -和頂層一樣---一樣的話===_find_tests方法就直接從目錄下面找腳本-當如如果有目錄也會繼續走--他是在_find_tests_path判斷的--最終也是回到找腳本模塊上
 7      #如果不一樣--那就是找了 找到這個目錄了---那麼就從頂層開始找這個目錄的相對路徑--其實就是找最後那個目錄(必須是一個packge。上面說的目錄都是包)、。。然後返回一個name
 8      #如果還有子目錄  d---那就會返回  c.d
 9     if path == self._top_level_dir:  #首次運行pattern,top_level_dir=None):self._top_level_dir = top_level_dir = start_dir -第二次運行如果目錄沒有變，這裏也是直接返回的
10         return '.'
11         
12         
13     path = _jython_aware_splitext(os.path.normpath(path))  #如果我們當前傳的和上次傳的目錄不一致。。這裏得path我們當前傳的路徑
14     
15     _relpath = os.path.relpath(path, self._top_level_dir)    #從self._top_level_dir開始找path的相對路徑
16     #這裡是從我們傳的path開始找到self._top_level_dir上次傳的相對路徑 
17     #例如： path=path1="E:\\PyFiles\\Besettest\\besettest\\interface\\testFiles"   self._top_level_dir="E:\\PyFiles\\Besettest\\besettest\\interface\\result"
18     #那麼_relpath="..\testFiles"    -暫時還不清楚為什麼要找這個？？？？？？？？？？？？？？？？？？？？？？？？？？
19     assert not os.path.isabs(_relpath), "Path must be within the project"
20     #↑↑斷言 不是絕對路徑-也就是說_relpath是否為相對路徑↑↑↑特么的 這裏肯定是一個相對路徑啊。。。上面都有relpath了。。。丟
21     #↓↓↓↓斷言以..開頭就失敗。。。↓↓↓--這兩處超出理解範圍了。。。。。。
22     assert not _relpath.startswith('..'), "Path must be within the project"
23     
24     name = _relpath.replace(os.path.sep, '.')  #然後這裏又把分隔符替換成.  返回 a.b  當然或許會有異常情況返回.....這是我意淫的
25     return name

View Code self._find_test_path #找測試的路徑 兩個主要邏輯： 傳到full_path 是一個文件 還是一個目錄

 1 def _find_test_path(self, full_path, pattern, namespace=False):  
 2  #_find_tests()調用這個方法 傳了一個我們傳的目錄下的a文件路徑、和需要找的文件pattern-namespace【傳的可能是true 也可能是false】，如果我的目錄是對的-namespace傳的就是false
 3     """Used by discovery.
 4 
 5     Loads tests from a single file, or a directories' __init__.py when
 6     passed the directory.
 7 
 8     Returns a tuple (None_or_tests_from_file, should_recurse).
 9     """
10     basename = os.path.basename(full_path)   #basename==文件名.py後續帶py的統稱文件--不帶後綴的統稱文件名。。。
11     if os.path.isfile(full_path):   #如果我們傳的full_path是一個文件---我們在discover傳的是一個腳本目錄-之前在_test_find是做了一個拼接得到的完整路徑full_path
12         if not VALID_MODULE_NAME.match(basename): #判斷他是不是一個py文件----
13             
14             # valid Python identifiers only
15             return None, False   #如果不是直接返回
16         if not self._match_path(basename, full_path, pattern):  #這裏雖然傳了三個值--但是實際上只有basename，pattern有用--
17          #_match_path調用fnmatch(文件, 我們傳的文件格式或文件)這個需要————from fnmatch import fnmatch他的主要作用是做此模塊的主要作用是文件名稱的匹配
18          #當此次傳入的文件名與我們的文件格式匹配一致self._match_path返回true
19             return None, False    #如果不一樣 就直接回到 _find_test繼續找
20         # if the test file matches, load it
21         name = self._get_name_from_path(full_path)   #然後這裏把文件路徑又傳到 self._get_name_from_path去返迴文件名-這個時候因為我們傳的是腳本目錄-full_path目錄下的文件路徑，所以返回的name 就是文件名
22        
23         #self._top_level_dir是當前文件目錄路徑，path是當前文件路徑--從目錄找文件--直接就是文件名--他返回的name就是文件名   
24         try:
25             module = self._get_module_from_name(name)    #_get_module_from_name  這個方法就是動態導入模塊名--然後返回一個所有導入的模塊的對象 moudel.__file__路徑、moudel.__name__名稱
26         except case.SkipTest as e:   #如果導入不成功 case.SkipTest 實際上case是繼承--Exception--所以把這個理解為Exception就可以了--
27             return _make_skipped_test(name, e, self.suiteClass), False
28         except:
29             error_case, error_message = \
30                 _make_failed_import_test(name, self.suiteClass)   
31             self.errors.append(error_message)
32             return error_case, False
33         else: #module 獲取到值之後走這裏。。
34             mod_file = os.path.abspath(
35                 getattr(module, '__file__', full_path))    #然後這裏取出我們導入模塊的 絕對路徑---如果反射找不到就返回該文件的路徑-其實差別不大-處理一下更嚴謹
36             realpath = _jython_aware_splitext(
37                 os.path.realpath(mod_file))   #os.path.realpath(mod_file)然後又返回真實路徑---然後又去掉路徑的.py，。，，，，，，，，丟
38             fullpath_noext = _jython_aware_splitext(
39                 os.path.realpath(full_path))   #然後full_path 找真實路徑去掉.py
40             if realpath.lower() != fullpath_noext.lower():  #如果動態導入的模塊的目錄路徑 不等於 傳進來(也就是pattern)的目錄路徑--實際上傳進來的路徑肯定是個絕對路徑--因為前面已經轉了好幾次絕對路徑了
41                 module_dir = os.path.dirname(realpath)   #不等於就找動態導入模塊所在的目錄----實際上上面處理的realpath已經是一個目錄了。。但是他防止realpath還是一個.py文件。所以又操作了一次
42                 mod_name = _jython_aware_splitext(    #full_path是文件的路徑.py的，然後這裏又先是basename取出文件(就是把路徑去掉，只留下xxx.py) 然後外面那個方法 把.py去掉--留下文件名
43                     os.path.basename(full_path))
44                 expected_dir = os.path.dirname(full_path)  
45                 #然後找到需要執行腳本所在的目錄。。。。。也就是說正常情況  假設expected_dir="e://a/b"  那麼 mod_file =realpathfullpath_noext="e://a/b/scripy" 
46                 #scripy是一個py文件---上面這個if是說的 正常情況。。。我想不到導入模塊和導入模塊的路徑不相等的情況--不過這個不重要-源碼這樣肯定是有道理的
47                   msg = ("%r module incorrectly imported from %r. Expected "
48                        "%r. Is this module globally installed?")
49                 raise ImportError(
50                     msg % (mod_name, module_dir, expected_dir))
51             return self.loadTestsFromModule(module, pattern=pattern), False   
52             #然後走 從模塊從加載測試s 這個方法---也就是說discover實際上是調用loadTestsFromMould這個方法的。。測試套件也是在這一步處理的
53     elif os.path.isdir(full_path):  #dicover裏面傳腳本目錄是走這裏。。
54         if (not namespace and    #namespace-默認是false   not namespace就是true  
55             not os.path.isfile(os.path.join(full_path, '__init__.py'))): #不是一個包。。-也就是說我們傳的目錄應該是一個包，下面包含__init__.py
56             return None, False
57         
58         load_tests = None  #這個load_tests是啥意思呢？？？？？？？？？？後面繼續看----看了一遍--並且用unittest.main()試了一下-模塊下面是沒有這個屬性的。。只是unittest的初始化文件有這個方法--他也是通過discover找的。。
59         tests = None
60         name = self._get_name_from_path(full_path)   #這裏就是走子目錄的邏輯了
61         #get_name_from_path的邏輯在這裏就很清晰了 
62             #A.如果通過_find_test 調用self._get_name_from_path  是為了判斷兩次start_dir是否一致一致返回.  不一致返回從上次的start_dir1找到本次start_dir12的相對路徑--
63             #這裏又分兩種情況-A1正常情況-start_dir1是start_dir12的上級目錄。。。那麼返回的那麼就是-A.B這樣的了。。因為第一次的A是已經os.path.insert到環境變量了..所以A.B是可以直接用
64                                 #A2不正常情況 就是之前說的 最少返回一個點的...A這種返回---然後問題來了--他為什麼要這麼處理呢--原因是？？？？？？？？？
65                                 #我猜是與腳本同級存在另一個腳本目錄。。。後面驗證這一點。-----這裡在上面補充了--是因為子目錄中還存在腳本所有這麼走邏輯-完美的
66         try:
67             package = self._get_module_from_name(name)   #上面是導入的一個module--這裡是導入一個包-- 返回--
68         except case.SkipTest as e:
69             return _make_skipped_test(name, e, self.suiteClass), False
70         except:
71             error_case, error_message = \
72                 _make_failed_import_test(name, self.suiteClass)
73             self.errors.append(error_message)
74             return error_case, False
75         else:
76             load_tests = getattr(package, 'load_tests', None)  #然後判斷這個包裏面有沒有'load_tests'這個屬性---這裏我代碼一直看下來，我們是不知道這個lood_tests是什麼的，字面意思 加載測試集合
77             # Mark this package as being in load_tests (possibly ;))
78             self._loading_packages.add(name)  #然後把模塊名稱添加到set集合
79             try:
80                 tests = self.loadTestsFromModule(package, pattern=pattern)   
81                 #這裏傳了一個package模塊對象,和文件匹配規則合作或者文件。。--但是這裏導入一個包之後實際上是找不到testCase的-因為包下面的屬性肯定不是一個類-不會走loadTestsFromTestCase
82                 #所以這裏返回的tests是一個空列表
83                 if load_tests is not None: #貌似這個是棄用的，向後兼容-暫時沒看明白這個load_tests代表的意思
84                     # loadTestsFromModule(package) has loaded tests for us.
85                     return tests, False
86                 return tests, True  #  如果能走到這裏------就返回True_就是給_find_test判斷走遞歸的--_find_tests裏面就得到 should_recurse=True
87             finally:
88                 self._loading_packages.discard(name)  #然後這個刪掉set集合裏面之前導入的那個包
89         else:
90         return None, False

View Code

 1 def loadTestsFromModule(self, module, *args, pattern=None, **kws):
 2     """Return a suite of all test cases contained in the given module"""
 3     # This method used to take an undocumented and unofficial
 4     # use_load_tests argument.  For backward compatibility, we still
 5     # accept the argument (which can also be the first position) but we
 6     # ignore it and issue a deprecation warning if it's present.
 7     if len(args) > 0 or 'use_load_tests' in kws:    #args這個默認是一個空元組  長度默認為0  kws是一個空字典
 8         warnings.warn('use_load_tests is deprecated and ignored',
 9                       DeprecationWarning)
10         kws.pop('use_load_tests', None)
11     if len(args) > 1:
12         # Complain about the number of arguments, but don't forget the
13         # required `module` argument.
14         complaint = len(args) + 1
15         raise TypeError('loadTestsFromModule() takes 1 positional argument but {} were given'.format(complaint))
16     if len(kws) != 0:
17         # Since the keyword arguments are unsorted (see PEP 468), just
18         # pick the alphabetically sorted first argument to complain about,
19         # if multiple were given.  At least the error message will be
20         # predictable.
21         complaint = sorted(kws)[0]  #取出第一個Key-
22         raise TypeError("loadTestsFromModule() got an unexpected keyword argument '{}'".format(complaint))
23     tests = []
24     for name in dir(module):    #這裏得modul實際上使我們傳入的模塊對象---dir(object)  返回模塊下的所有屬性列表-
25         obj = getattr(module, name)   #然後反射返回name對象。。返回的是一個class對象
26         if isinstance(obj, type) and issubclass(obj, case.TestCase):    #這裏判斷obj是否是一個類--並且這個類是case.TestCase的子類，也就是說 是否寫在我們繼承unitest.testCase那個類的下面
27             tests.append(self.loadTestsFromTestCase(obj))  #可以看到最後走loadTestFromTestCase obj這裡是傳入的一個類名
28 
29     load_tests = getattr(module, 'load_tests', None)
30     tests = self.suiteClass(tests)
31     if load_tests is not None:
32         try:
33             return load_tests(self, tests, pattern)
34         except Exception as e:
35             error_case, error_message = _make_failed_load_tests(
36                 module.__name__, e, self.suiteClass)
37             self.errors.append(error_message)
38             return error_case
39     return tests  #返回集合

View Code loadTestsFromTestCase：這裏就是添加testcase到suit集合裏面的主要邏輯

 1 def loadTestsFromTestCase(self, testCaseClass):  #testCaseClass是我們傳的一個用例類
 2     """Return a suite of all test cases contained in testCaseClass"""
 3     if issubclass(testCaseClass, suite.TestSuite):  #這個類是不是suite.TestSuite的子類--如果是的就拋出異常==
 4         raise TypeError("Test cases should not be derived from "
 5                         "TestSuite. Maybe you meant to derive from "
 6                         "TestCase?")
 7     testCaseNames = self.getTestCaseNames(testCaseClass)  #getTestCaseNames  從類下面找到用例名稱--找到名稱返回的是一個列表
 8     if not testCaseNames and hasattr(testCaseClass, 'runTest'):  #這裏判斷testCaseNames是否為空-並且 是否存在"runTest"這個元素
 9         testCaseNames = ['runTest']
10     loaded_suite = self.suiteClass(map(testCaseClass, testCaseNames)) #是我的一個用例類--然後將testCaseNames類下面的測試方法--帶進去遍歷。。。高級用法---第一次見--這個方法很關鍵
11     return loaded_suite

View Code getTestCaseNames

 1 def getTestCaseNames(self, testCaseClass):
 2     """Return a sorted sequence of method names found within testCaseClass
 3     """
 4     def isTestMethod(attrname, testCaseClass=testCaseClass,   #定義一個內部方法
 5                      prefix=self.testMethodPrefix):     #self.testMethodPrefix="test"   這個在TestLoader下第一行就已經默認了，他是我們用例開頭的固定格式
 6         return attrname.startswith(prefix) and \  #這裏判斷了是否已test開頭以及 方法對象是否可用----getattr返回方法對象--callable()是檢測對象是否可用    返回一個布爾值
 7             callable(getattr(testCaseClass, attrname))
 8     testFnNames = list(filter(isTestMethod, dir(testCaseClass)))
 9      #dir(testCaseClass)返回該對象下面所有的屬性--包括變量test_1--所以上面需要檢測屬性時test開頭且是一個可以調用的對象--
10      #filter函數---前面是一個function -後面是一個可迭代對象-會遍歷可迭代對象-並傳入function-functions返回true則添加到列表--這樣就找到了所有的
11     if self.sortTestMethodsUsing:
12         testFnNames.sort(key=functools.cmp_to_key(self.sortTestMethodsUsing))     
13         #sortTestMethodsUsing = staticmethod(util.three_way_cmp) 轉為為靜態方法-內存地址指向self.sortTestMethodsUsing
14         #然後通過functools這個模塊排序==
15     return testFnNames  #然後返回用例方法名稱列表

View Code 所有的邏輯 就是在TestLoader找測試用例的時候–通過_find_tests這個方法從目錄開始找文件(子目錄)-模塊-類-方法名 然後將某個模塊下的類通過他下面的方法map返回多個對象，也就是說一個testClass下面存在五個test_method，他就會返回五個實例對象-並生成一個suite集合–然後加入到一個列表 如果一個模塊下有多個testClasee 同樣-實際上是一樣的–實際上他是先通過loadTestsFromModule這個方法找到所有的類對象之後在遍歷–然後才走上面那一步的，，，多個testClasss就存在多個suite集合– 也就是說 一個modul下面的 suite集合會添加到一個列表–[suite=[A-TestCase實例化對象1,A-TestCase實例化對象2]，suite=[B-TestCase實例化對象1，B-TestCase實例化對象2]]—然後在將這個列表當做參數傳入TestSuite實例化一個新對象[suite=-[suite=[A-TestCase實例化對象1,A-TestCase實例化對象2]，suite=[B-TestCase實例化對象1，B-TestCase實例化對象2]]]—-這樣就是一個模塊下用的結構 但是還沒有完–這裏只是一個modul下的—還有多個modul–到了大家估計也知道剩下的會幹什麼了— 沒錯–當我得到modul的全部suite集合之後—這個集合最終會返回給_find_tests方法–通過生成器返回給discover–也就是將這個suite集合又加入到了一個新的列表–然後discover又將這個list 帶入形成了一個—–最終的實例對象，最終返回的格式如下——- [suite= [suite1=-[suite=[A-TestCase實例化對象1,A-TestCase實例化對象2]，suite=[B-TestCase實例化對象1，B-TestCase實例化對象2]]], [suite2=-[suite=[A-TestCase實例化對象1,A-TestCase實例化對象2]，suite=[B-TestCase實例化對象1，B-TestCase實例化對象2]]] ]   –看過源碼的都知道—我們run的時候—就這個實例對象是可以接受參數的–而這個參數就是result—因為TestSuite繼承的BaseTestsSuite 有一個__call__這個
魔術方法：如果在類中實現了 __call__ 方法，那麼實例對象也將成為一個可調用對象，具體百度。這裏不做過多解釋—–所以最終的suite是可以接受參數的test(result)–接受參數之後直接走——call下面的邏輯了 本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

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

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

※超省錢租車方案

※FB行銷專家,教你從零開始的技巧

萬級TPS億級流水-中台賬戶系統架構設計

標籤：高併發 萬級TPS 億級流水 賬戶系統

• 背景
• 業務模型
• 應用層設計
• 數據層設計
• 日切對賬

背景

我們需要給所有前台業務提供統一的賬戶系統，用來支撐所有前台產品線的用戶資產管理，統一提供支持大併發萬級TPS、億級流水、數據強一致、風控安全、日切對賬、財務核算、審計等能力，在萬級TPS下保證絕對的數據準確性和數據溯源能力。

注：資金類系統只有合格和不合格，哪怕數據出現只有0.01分的差錯也是不合格的，局部數據不準也就意味着全局數據都不可信。

本文只分享系統的核心模型部分的設計，其他常規類的（如壓測驗收、系統保護策略-限流、降級、熔斷等）設計就不做多介紹，如果對其他方面有興趣歡迎進一步交流。

業務模型

基本賬戶管理： 根據交易的不同主體，可以分為個人賬戶、機構賬戶。
賬戶餘額在使用上沒有任何限制，很純粹的賬戶存儲、轉賬管理，可以滿足90%業務場景。

子賬戶功能： 一個用戶可以開通多個子賬戶，根據餘額屬性不同可以分為基本賬戶、過期賬戶，根據幣種不同可以分為人民幣賬戶、虛擬幣賬戶，根據業務形態不同可以自定義。
（不同賬戶的特定功能是通過賬戶上的賬戶屬性來區分實現。）

過期賬戶管理： 該賬戶中的餘額是會隨着進賬流水到期自動過期。
如：在某平台充值1000元送300元，其中300元是有過期時間的，但是1000元是沒有時間限制的。這裏的1000元存在你的基本賬戶中，300元存在你的過期賬戶中。

注：過期賬戶的每一筆入賬流水都會有一個到期時間。系統根據交易流水的到期時間，自動核銷用戶過期賬戶中的餘額，記為平台的確認收入。

賬戶組合使用：支持多賬戶組合使用，根據配置的優先扣減順序進行扣減餘額。比如：在 基本賬戶 和 過期賬戶 （充值賬戶）中扣錢一般的順序是優先扣減過期賬戶的餘額。

應用層設計

根據上述業務模型，賬戶系統是一個典型的 數據密集型系統 ，業務層的邏輯不複雜。整個系統的設計關鍵點在於如何平衡大併發TPS和數據一致性。

熱點賬戶：前台直播類業務存在熱點賬戶問題，每到各種活動賽事的時候會存在 90%DAU 給少數幾個頭部主播打賞的場景。
DB就會有熱點行問題，由於 行鎖 關係併發一大肯定大量超時、RT突增 、 DB活躍線程 增加等一系列問題，最終DB會被拖掛。

賬戶類系統有一個特點，原賬戶的扣減可以實時處理，目標賬戶可以異步處理，我們可以將轉賬動作拆解為兩個階段進行異步化。（可以參考銀行轉賬業務。）

比如：A給B轉賬100元，原賬戶A的100元餘額扣減可以同步處理，B賬戶的100增加可以異步處理。這樣哪怕10w人給主播打賞，這10w人的賬戶都是分散的，而主播的餘額增加則是異步處理的。

賬戶轉賬扣減A賬戶餘額，記錄A賬戶出賬流水，記錄B賬戶入賬流水，這三個動作可以在一個DBTransaction中處理，可以保證源賬戶進出帳一致性。目標賬戶B的入賬可以異步處理，為了保證萬無一失且滿足一定的實時性，需要兩步結合，程序里通過MQ走異步入賬，同時增加DB的兜底JOB定時掃描 入賬流水記錄 為未到賬的流水進行入賬。

我們通過異步化緩解熱點行處理，但是如果 收款方 強烈要求收款必須在一定的時間內完成，我們還是需要進一步處理，後面會講到。

過期賬戶： 通常過期賬戶用來管理贈送類賬戶，這類賬戶有一定的時效性，用戶在使用上也是優先扣減此類賬戶餘額。
這類使用需求其實覆蓋面不大，真正用戶賬戶餘額不使用等着被系統過期的很少，畢竟這是一個很傻的行為。

過期賬戶的兩種核銷情況：第一種是用戶使用過期賬戶時的核銷。第二種是某個過期流水到了過期時間，系統自動核銷記為平台的確認收入。

過期賬戶核銷邏輯：用戶充值1000元到基本賬戶，平台贈送300元到贈送賬戶。此時，基本賬戶記錄進賬流水+1000元，贈送賬戶記錄進賬流水+300元並且該筆流水的過期時間為 2020-12-29 23:59:59 （過期時間由前台業務方設置） 。

系統自動核銷：如果用戶不在此時間之前用完就會被系統自動划進平台的收入，贈送賬戶餘額扣減-300元。

用戶使用核銷：如果用戶在過期時間前陸續在使用贈送賬戶，比如使用100元，那麼我們需要核銷原本進賬的300元的那筆流水，減少-150元。
也就是說，該筆過期流水已經核銷掉150元，帶過期核銷150元，到期后只要核銷150元即可，而不是300元。

過期賬戶每次使用均產生待核銷負向流水，系統自動核銷前必須保證沒有任何負向流水記錄才可以去扣減贈送賬戶餘額。

考慮到極端情況下，剛好過期JOB在進行自動過期核銷，用戶又在此時使用過期賬戶，這點需要注意下。可以簡單通過加DB-X鎖解決，這個場景其實非常稀少。

數據層設計

在應用層設計的時候，我們通過異步化方式來繞開熱點問題。
同樣我們在設計數據層的時候也要考慮單次操作DB的性能，比如控制事務的大小，事務跨網絡的次數等問題。當然還包括金額存儲的精度問題，精度問題處理不好也會影響性能。

浮點數問題： 如果我們用浮點數近似值來存儲金額，那麼就一定會有偏差，隨着金額越大時間越長偏差就會越大。比較好的方式是通過整型來存儲，通過放大金額比例來達到不同的業務場景下對金額比率的要求。

正常的1.12元，存儲比率是1=100元，那麼表裡的存儲值就是112，不同的貨幣比例都可以自由縮放，永遠都可以保持最準確的精度。

分庫分表+讀寫分離： 根據業務特點和未來增量規劃，將DB分為16個邏輯庫，前期使用2個物理庫承載。16個邏輯庫，按照每次2倍擴容，最大擴容上限是16個物理庫。單實例的配置 8c 32g 2t 8000conn 9000iops 。

按照單次TPS-rt 1ms計算，TPS 1w 需求，每台承載5k TPS，單庫的活躍線程大概在8-10個（考慮網絡延遲）。
最後到達瓶頸的都是iops，因為只要rt足夠短，最終壓力都會在IO上。

分庫按照uid分為16個庫，賬戶表不分表默認16張。每張表按照 1kw*16=1.6 億個賬戶。

單表能存儲多少要綜合考慮，比如查詢類型，單次查詢的RT，冷熱數據佔比（ innodb_buffer_pool 利用率）、是否充分發揮了索引，索引是否達到3星級別，索引片中沒有經常變更的字段等。

賬戶流水表按照日期分表365張，流水數據會隨着時間推移逐漸變成冷數據，定期歸檔冷數據。（這裏約定了，流水查詢只能按照uid+日期查詢。如果運營類的需求，要橫跨分片key獲取，走OLAP方案 clickhouse、hive等）

分庫分表採用阿里雲分佈式數據庫產品DRDS，1個主庫集群+2個讀庫集群（讀庫做了讀負載均衡，可以按需擴容）。

讀負載均衡器：https://github.com/Plen-wang/read-loadbalance

既然用了DRDS分佈式數據庫產品，那麼在查詢上需要充分考慮分片鍵的限制，如果存儲和查詢出現分片鍵衝突問題就需要我們手動計算分片路由，直接訪問物理節點。

訪問物理節點需要藉助DRDS專用SQL註釋子句來完成。

先通過 show node 查看物理DB ID、show topology from logic_table_name 查看物理表ID，然後在SQL帶上特定的註釋子句。

SELECT /*+TDDL:scan('logic_table_name', real_table=("real_table_name"),node='real_db_node_id')*/ 
count(1) FROM logic_table_name ;

賬戶更新： 對賬戶更新都有一個前提就是賬戶已經開通，但是我們為了最大化賬戶系統在使用上的便利性，讓前台業務方不需要做初始化動作，由賬戶系統惰性初始化，比如發現賬戶不存在就自動初始化賬戶數據。

但是我們怎麼知道賬戶不存在，不可能每次都去查詢一次或者根據執行返回錯誤判斷。而且 update 語句是區分不了錯誤的 賬戶不存在 還是 餘額不足 或者其他原因。

那麼如何巧妙的解決這個問題，只要一次DB往返。

我們可以使用 Mysql INSERT INTO ... ON DUPLICATE KEY UPDATE ... 子句，但是該子句有一個限制就是不支持 where 子句。

-- cut_version 樂觀鎖、account_property 賬戶屬性
insert into tb_account(uid,balance,cut_version,account_property) values("%s",%d,%d,%d) ON DUPLICATE KEY UPDATE balance = balance + %d,cut_version = cut_version+1

其實不完全推薦使用這個方法，因為這個方法也有弊端就是將來 where 子句無法使用，還有一個辦法就是合併 賬戶查詢 和 插入 為一條 sql 提交。

DB操作本身rt可能很短，但是如果跨網絡那麼事務的延遲會帶來DB的串行化增加，降低併發度，整體應用 rt就會增加。所以一個原則就是盡量不要跨網絡開事務，合併sql做一次事務提交，最短的事務周期，減少跨網絡的事務操作，如果我們將單次事務網絡交互減少2-3次，性能的提高可能會增加2-3倍，同樣由於網絡的不穩定抖動丟包對 999rt 線的影響也會減少2-3倍。

平衡好當前系統是業務密集型還是數據密集型。
判斷當前系統是否有很強的業務層邏輯，是否要運用DDD、RUP等強模型的工程方法。畢竟強模型和高性能在落地的時候有些方面是衝突的，需要進一步藉助 CRQS、GRASP等工程方法來解決。

單行熱點問題： 單行的TPS都是串行的，事務rt越短TPS就越高，按照1ms計算，差不多TPS就是1000。一般只有機構賬戶類型才會有這個需求。

我們可以將單行變成多行，增加行的并行度，加大賬戶操作的併發度。（這個方案要評估好寫入和查詢兩端需求）

id	uid	balance	slot
1	10101010	1000	1
2	10101010	2000	2
3	10101010	3000	3
4	10101010	400	4
5	10101010	300	5
6	10101010	200	6
7	10101010	200	7
8	10101010	200	8
9	10101010	200	9
10	10101010	200	10

insert into tb_account (uid,balance,slot)
values(10101010, 1000, round(rand()*9)+1) 
on  duplicate key update balance=balance+values(balance)

這裏的 10slot*單個slot 1000TPS，理論上可以跑到1w，如果機構賬戶數據量很大，可以擴展slot個數。

賬戶的總餘額通過sum()匯總，如果業務場景中有餘額的頻繁sum()操作，可以通過增加餘額中間表，定期 insert into tb_account_total select sum(balance) total_balance from tb_account group by uid 。

通常機構賬戶的結算是有周期的（T+7、T+30等），而且基本是沒有併發，所以在賬戶餘額扣減方面就可以輕鬆處理。
有兩種實現方案：

第一種，賬戶餘額允許單個slot為負數，但是總的sum()是正數。通過子查詢來對餘額進行檢查。

insert into tb_account (uid, balance, slot)
select uid,-1000 as balance,round(rand() *9+ 1)
from(
    select uid, sum(balance) as ss
    from tb_account
    where uid= 10101010
    group by uid having ss>= 1000 for update) as tmp
on duplicate key update balance= balance+ values(balance)

第二種，如果條件允許可以藉助用戶自定義變量來在DB上完成餘額累計掃描，將可以扣減的slot的主鍵id返回給程序，但是只需要一次DB交互就可以獲取出可以扣減的賬戶solt，然後分別開始對slot賬戶進行扣減。

set @f:=0;
select * from tb_account where id in(select id from (select id, @f:=@f+balance from tb_account where @f<1000 order by id) as t);

第二種方案在默認的mysql數據庫上都是支持的，但是有些數據庫雲產品不支持，阿里雲rds是不支持的。

日切對賬

賬戶系統有一個基本的需求，就是每天餘額鏡像，簡單講就是餘額在每天的快照，用來做T+1對賬。
不管財務還是每季度的審計都會需要，最重要的是我們自己也需要對賬戶數據做摸底對賬。

由於每天產生上億的流水，這需要在大數據平台中完成。

日切對賬：昨天賬戶餘額 – 前天賬戶餘額 = 昨天的流水 – 前天的流水。

比如，昨天的賬戶餘額是5000w，前台的賬戶餘額是4500w，差值就是500w。同樣道理，昨天的賬戶流水是5000w，前天的賬戶流水是4500w，那麼差值是500w，這就是沒問題的。

賬戶不僅有增加也有減少，可能昨天賬戶餘額比前天賬戶餘額差值是-500w，但是流水也要是-500w才行。

由於每天會產生億級的流水，用傳統的全量抽取不現實，這類數據抽取的速度都會有延遲，而且對賬最重要的是時間點必須非常精準，才能保證餘額和流水是對得上的。

要不然會出現HDFS的分區是2020-06-10號，但是該分區里有2020-06-11的數據，就是因為拉取的時候會延遲到第二天。這個問題也可以通過增加拉取sql的條件限制來解決這個問題，但是無法做到0點瞬間鏡像全部賬戶。

解決方案： 全量餘額+binlog增量更新
1.賬戶表，先做一次全量同步。
2.DB的所有變更通過binlog（默認row複製）進到數倉。（因為 binlog 是基於發生時間的，所以無所謂我們是不是在0點去計算鏡像）
3.T+1跑JOB的時候，獲取前一天的賬戶餘額，然後通過 binlog 來覆蓋前天與昨天的交集部分。

由於數倉的 binlog 數據都是增量的，所以要想取到正確的全量數據需要用到一定的技巧。

select app_id,sub_type,sum(amount) records_amount from (
      select *,row_number()over(partition by id order by updated_at) as rn
      from hive_db_table
      where dt='${YESTERDAY}'
  ) t where t.rn=1
       group by t.sub_type,t.app_id

使用 hive 開窗函數 row_number()over() 對同樣的id進行分組，然後獲取最新的一條數據就是賬戶在T的最後的值。

作者：王清培（趣頭條 Tech Leader）

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

【其他文章推薦】

※超省錢租車方案

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

※回頭車貨運收費標準

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

※FB行銷專家,教你從零開始的技巧