Python 和 Django 性能測試和優化指南

Donald Knuth 說“過早的優化是萬惡之源”。 但是有一段時間，通常在高負載的成熟項目中，不可避免地需要優化。 在本文中，我想談談優化 Web 項目代碼的五種常用方法。 我將使用 Django，但其他框架和語言的原理應該類似。 在本文中，我將使用這些方法將查詢的響應時間從 77 秒減少到 3.7 秒。

示例代碼改編自我參與過的一個真實項目，並演示了性能優化技術。 如果您想自己跟隨並查看結果，您可以在 GitHub 上獲取初始狀態的代碼，並在跟隨文章的同時進行相應的更改。 我將使用 Python 2，因為 Python 3 還沒有一些第三方包。

介紹我們的應用程序

我們的網絡項目只是跟踪每個國家的房地產報價。 因此，只有兩種模型：

 # houses/models.py from utils.hash import Hasher class HashableModel(models.Model): """Provide a hash property for models.""" class Meta: abstract = True @property def hash(self): return Hasher.from_model(self) class Country(HashableModel): """Represent a country in which the house is positioned.""" name = models.CharField(max_length=30) def __unicode__(self): return self.name class House(HashableModel): """Represent a house with its characteristics.""" # Relations country = models.ForeignKey(Country, related_name='houses') # Attributes address = models.CharField(max_length=255) sq_meters = models.PositiveIntegerField() kitchen_sq_meters = models.PositiveSmallIntegerField() nr_bedrooms = models.PositiveSmallIntegerField() nr_bathrooms = models.PositiveSmallIntegerField() nr_floors = models.PositiveSmallIntegerField(default=1) year_built = models.PositiveIntegerField(null=True, blank=True) house_color_outside = models.CharField(max_length=20) distance_to_nearest_kindergarten = models.PositiveIntegerField(null=True, blank=True) distance_to_nearest_school = models.PositiveIntegerField(null=True, blank=True) distance_to_nearest_hospital = models.PositiveIntegerField(null=True, blank=True) has_cellar = models.BooleanField(default=False) has_pool = models.BooleanField(default=False) has_garage = models.BooleanField(default=False) price = models.PositiveIntegerField() def __unicode__(self): return '{} {}'.format(self.country, self.address)

抽象的HashableModel提供從它繼承的任何模型的hash屬性，該屬性包含實例的主鍵和模型的內容類型。 這會通過用哈希替換敏感數據（例如實例 ID）來隱藏它們。 當您的項目有多個模型並且您需要一個集中的地方來取消哈希並決定如何處理不同類的不同模型實例時，它也可能很有用。 請注意，對於我們的小項目，散列並不是真正需要的，因為我們可以不用它，但它有助於演示一些優化技術，所以我會保留它。

這是Hasher類：

 # utils/hash.py import basehash class Hasher(object): @classmethod def from_model(cls, obj, klass=None): if obj.pk is None: return None return cls.make_hash(obj.pk, klass if klass is not None else obj) @classmethod def make_hash(cls, object_pk, klass): base36 = basehash.base36() content_type = ContentType.objects.get_for_model(klass, for_concrete_model=False) return base36.hash('%(contenttype_pk)03d%(object_pk)06d' % { 'contenttype_pk': content_type.pk, 'object_pk': object_pk }) @classmethod def parse_hash(cls, obj_hash): base36 = basehash.base36() unhashed = '%09d' % base36.unhash(obj_hash) contenttype_pk = int(unhashed[:-6]) object_pk = int(unhashed[-6:]) return contenttype_pk, object_pk @classmethod def to_object_pk(cls, obj_hash): return cls.parse_hash(obj_hash)[1]

由於我們想通過 API 端點提供這些數據，我們安裝 Django REST Framework 並定義以下序列化程序和視圖：

 # houses/serializers.py class HouseSerializer(serializers.ModelSerializer): """Serialize a `houses.House` instance.""" id = serializers.ReadOnlyField(source="hash") country = serializers.ReadOnlyField(source="country.hash") class Meta: model = House fields = ( 'id', 'address', 'country', 'sq_meters', 'price' )

 # houses/views.py class HouseListAPIView(ListAPIView): model = House serializer_class = HouseSerializer country = None def get_queryset(self): country = get_object_or_404(Country, pk=self.country) queryset = self.model.objects.filter(country=country) return queryset def list(self, request, *args, **kwargs): # Skipping validation code for brevity country = self.request.GET.get("country") self.country = Hasher.to_object_pk(country) queryset = self.get_queryset() serializer = self.serializer_class(queryset, many=True) return Response(serializer.data)

現在，我們用一些數據填充我們的數據庫（使用factory-boy生成的總共 100,000 個房屋實例：一個國家/地區為 50,000，另一個國家/地區為 40,000，第三國家/地區為 10,000），並準備測試我們的應用程序的性能。

性能優化就是衡量

我們可以在項目中測量幾件事：

• 執行時間處理時間
• 代碼行數
• 函數調用次數
• 分配的內存
• 等等。

但並非所有這些都與衡量我們項目的執行情況有關。 一般來說，有兩個主要指標是最重要的：執行多長時間以及需要多少內存。

在 Web 項目中，響應時間（服務器接收某個用戶操作生成的請求、處理它並返回結果所需的時間）通常是最重要的指標，因為它不會讓用戶在等待時感到無聊以獲得響應並切換到瀏覽器中的另一個選項卡。

在編程中，分析項目性能稱為分析。 為了分析我們 API 端點的性能，我們將使用 Silk 包。 在安裝它並進行/api/v1/houses/?country=5T22RI調用（對應於具有 50,000 個房屋條目的國家/地區的哈希值）之後，我們得到以下信息：

總體響應時間為 77 秒，其中 16 秒用於數據庫中的查詢，總共進行了 50,000 次查詢。 如此龐大的數字，還有很大的改進空間，所以讓我們開始吧。

1.優化數據庫查詢

性能優化最常見的技巧之一是確保優化您的數據庫查詢。 本案也不例外。 此外，我們可以對查詢做幾件事來優化響應時間。

1.1 一次性提供所有數據

仔細看看這 50,000 個查詢是什麼，您會發現這些都是對houses_country表的冗餘查詢：

這個問題的根源在於，在 Django 中，查詢集是惰性的。 這意味著在您實際需要獲取數據之前，不會評估查詢集並且不會訪問數據庫。 同時，它只獲取您告訴它的數據，如果需要任何其他數據，則會發出後續請求。

這正是我們案例中發生的情況。 通過House.objects.filter(country=country)獲取查詢集時，Django 獲取給定國家/地區所有房屋的列表。 但是，在序列化house實例時， HouseSerializer需要房屋的country /地區實例來計算序列化程序的country /地區字段。 由於查詢集中不存在國家/地區數據，django 會發出一個額外的請求來獲取該數據。 它對查詢集中的每一所房子都這樣做——總共有 50,000 次。

不過，解決方案非常簡單。 為了提取序列化所需的所有數據，您可以在查詢集上使用select_related()方法。 因此，我們的get_queryset將如下所示：

 def get_queryset(self): country = get_object_or_404(Country, pk=self.country) queryset = self.model.objects.filter(country=country).select_related('country') return queryset

讓我們看看這對性能有何影響：

整體響應時間下降到 36 秒，在數據庫中花費的時間約為 100 毫秒，只有 4 個查詢！ 這是個好消息，但我們可以做得更多。

1.2 只提供相關數據

默認情況下，Django 從數據庫中提取所有字段。 但是，當您有包含許多列和行的巨大表時，告訴 Django 要提取哪些特定字段是有意義的，這樣它就不會花時間獲取根本不會使用的信息。 在我們的例子中，我們只需要五個字段進行序列化，但我們有 17 個字段。 準確指定從數據庫中提取哪些字段是有意義的，這樣我們可以進一步縮短響應時間。

Django 有defer()和only()查詢集方法可以做到這一點。 第一個指定不加載哪些字段，第二個指定只加載哪些字段。

 def get_queryset(self): country = get_object_or_404(Country, pk=self.country) queryset = self.model.objects.filter(country=country)\ .select_related('country')\ .only('id', 'address', 'country', 'sq_meters', 'price') return queryset

這將花在查詢上的時間減少了一半，這很好，但 50 毫秒並沒有那麼多。 整體時間也略有下降，但還有更多的空間可以減少。

2. 優化你的代碼

您不能無限優化數據庫查詢，而我們的上一個結果正好表明了這一點。 即使我們假設將查詢花費的時間減少到 0，我們仍然會面臨等待半分鐘才能獲得響應的現實。 是時候切換到另一個優化級別了：業務邏輯。

2.1 簡化你的代碼

有時，第三方包會為簡單的任務帶來大量開銷。 一個這樣的例子是我們返回序列化房屋實例的任務。

Django REST Framework 很棒，有很多開箱即用的有用功能。 但是，我們現在的主要目標是減少響應時間，因此它是一個很好的優化候選者，尤其是序列化對象非常簡單。

讓我們為此目的編寫一個自定義序列化程序。 為了簡單起見，我們將使用一個靜態方法來完成這項工作。 實際上，您可能希望擁有相同的類和方法簽名，以便能夠互換使用序列化程序：

 # houses/serializers.py class HousePlainSerializer(object): """ Serializes a House queryset consisting of dicts with the following keys: 'id', 'address', 'country', 'sq_meters', 'price'. """ @staticmethod def serialize_data(queryset): """ Return a list of hashed objects from the given queryset. """ return [ { 'id': Hasher.from_pk_and_class(entry['id'], House), 'address': entry['address'], 'country': Hasher.from_pk_and_class(entry['country'], Country), 'sq_meters': entry['sq_meters'], 'price': entry['price'] } for entry in queryset ] # houses/views.py class HouseListAPIView(ListAPIView): model = House serializer_class = HouseSerializer plain_serializer_class = HousePlainSerializer # <-- added custom serializer country = None def get_queryset(self): country = get_object_or_404(Country, pk=self.country) queryset = self.model.objects.filter(country=country) return queryset def list(self, request, *args, **kwargs): # Skipping validation code for brevity country = self.request.GET.get("country") self.country = Hasher.to_object_pk(country) queryset = self.get_queryset() data = self.plain_serializer_class.serialize_data(queryset) # <-- serialize return Response(data)

現在看起來好多了。 由於我們沒有使用 DRF 序列化程序代碼，響應時間幾乎減半。

另一個可衡量的結果——在請求/響應週期中進行的總函數調用次數——從 15,859,427 次調用（來自上面第 1.2 節中的請求）下降到 9,257,469 次。 這意味著大約 1/3 的函數調用是由 Django REST 框架進行的。

2.2 更新/替換第三方包

上面描述的優化技術是最常見的，你可以在沒有徹底分析和思考的情況下完成這些技術。 但是，17 秒仍然感覺很長； 為了減少這個數字，我們需要更深入地研究我們的代碼並分析引擎蓋下發生的事情。 換句話說，我們需要分析我們的代碼。

您可以使用內置的 Python 分析器自己進行分析，也可以使用一些第三方包（使用內置的 Python 分析器）。 由於我們已經使用silk ，它可以分析代碼並生成二進製配置文件，我們可以進一步可視化。 有幾個可視化包可以將二進製配置文件轉換為一些有洞察力的可視化。 我將使用snakeviz包。

這是上面最後一個請求的二進製配置文件的可視化，與視圖的調度方法掛鉤：

從上到下是調用堆棧，顯示文件名、方法/函數名稱及其行號以及在該方法中花費的相應累積時間。 現在更容易看出大部分時間都用於計算哈希（紫色的__init__.py和primes.py矩形）。

目前，這是我們代碼的主要性能瓶頸，但同時它並不是我們的代碼——它是一個第三方包。

在這種情況下，我們可以做的事情有限：

• 檢查軟件包的新版本（希望有更好的性能）。
• 找到另一個在我們需要的任務上表現更好的包。
• 編寫我們自己的實現，它將擊敗我們當前使用的包的性能。

對我來說幸運的是，有一個更新版本的basehash包負責散列。 該代碼使用v.2.1.0，但有一個v.3.0.4。 當您能夠更新到更新版本的包時，這種情況更有可能在您處理現有項目時發生。

在查看 v.3 的發行說明時，有一個特定的句子聽起來很有希望：

對素數算法進行了大規模檢修。 包括（原文如此）對 gmpy2 的支持，如果它在系統上可用（原文如此），則增加更多。

讓我們找出答案！

 pip install -U basehash gmpy2

我們將響應時間從 17 秒縮短到 8 秒以下。 很好的結果，但我們還應該注意一件事。

2.3 重構自己的代碼

到目前為止，我們已經改進了我們的查詢，用我們自己非常具體的功能替換了第三方復雜和通用的代碼，並更新了第三方包，但我們保留了現有代碼不變。 但有時對現有代碼的小規模重構可以帶來令人印象深刻的結果。 但是為此，我們需要再次分析分析結果。

仔細觀察，您會發現散列仍然是一個問題（毫不奇怪，這是我們對數據所做的唯一事情），儘管我們確實在這個方向上有所改進。 但是，表示__init__.py消耗 2.14 秒的綠色矩形以及緊隨其後的灰色__init__.py:54(hash)讓我感到困擾。 這意味著一些初始化需要很長時間。

我們來看看basehash包的源碼。

 # basehash/__init__.py # Initialization of `base36` class initializes the parent, `base` class. class base36(base): def __init__(self, length=HASH_LENGTH, generator=GENERATOR): super(base36, self).__init__(BASE36, length, generator) class base(object): def __init__(self, alphabet, length=HASH_LENGTH, generator=GENERATOR): if len(set(alphabet)) != len(alphabet): raise ValueError('Supplied alphabet cannot contain duplicates.') self.alphabet = tuple(alphabet) self.base = len(alphabet) self.length = length self.generator = generator self.maximum = self.base ** self.length - 1 self.prime = next_prime(int((self.maximum + 1) * self.generator)) # `next_prime` call on each initialized instance

可以看到， base實例的初始化需要調用next_prime函數； 正如我們在上面可視化的左下角矩形中看到的那樣，這非常重。

再來看看我的Hash類：

 class Hasher(object): @classmethod def from_model(cls, obj, klass=None): if obj.pk is None: return None return cls.make_hash(obj.pk, klass if klass is not None else obj) @classmethod def make_hash(cls, object_pk, klass): base36 = basehash.base36() # <-- initializing on each method call content_type = ContentType.objects.get_for_model(klass, for_concrete_model=False) return base36.hash('%(contenttype_pk)03d%(object_pk)06d' % { 'contenttype_pk': content_type.pk, 'object_pk': object_pk }) @classmethod def parse_hash(cls, obj_hash): base36 = basehash.base36() # <-- initializing on each method call unhashed = '%09d' % base36.unhash(obj_hash) contenttype_pk = int(unhashed[:-6]) object_pk = int(unhashed[-6:]) return contenttype_pk, object_pk @classmethod def to_object_pk(cls, obj_hash): return cls.parse_hash(obj_hash)[1]

如您所見，我標記了兩個方法，它們在每個方法調用上初始化一個base36實例，這並不是真正需要的。

由於散列是一個確定性過程，這意味著對於給定的輸入值，它必須始終生成相同的散列值，我們可以將其設為類屬性，而不必擔心它會破壞某些東西。 讓我們看看它將如何執行：

 class Hasher(object): base36 = basehash.base36() # <-- initialize hasher only once @classmethod def from_model(cls, obj, klass=None): if obj.pk is None: return None return cls.make_hash(obj.pk, klass if klass is not None else obj) @classmethod def make_hash(cls, object_pk, klass): content_type = ContentType.objects.get_for_model(klass, for_concrete_model=False) return cls.base36.hash('%(contenttype_pk)03d%(object_pk)06d' % { 'contenttype_pk': content_type.pk, 'object_pk': object_pk }) @classmethod def parse_hash(cls, obj_hash): unhashed = '%09d' % cls.base36.unhash(obj_hash) contenttype_pk = int(unhashed[:-6]) object_pk = int(unhashed[-6:]) return contenttype_pk, object_pk @classmethod def to_object_pk(cls, obj_hash): return cls.parse_hash(obj_hash)[1]

最終結果不到四秒，比我們開始的要小得多。 使用緩存可以進一步優化響應時間，但我不會在本文中討論它。

結論

性能優化是一個分析和發現的過程。 沒有適用於所有情況的硬性規則，因為每個項目都有自己的流程和瓶頸。 但是，您應該做的第一件事是分析您的代碼。 如果在這樣一個簡短的示例中，我可以將響應時間從 77 秒減少到 3.7 秒，那麼大型項目具有更大的優化潛力。

如果您有興趣閱讀更多與 Django 相關的文章，請查看 Toptal Django 開發人員 Alexandr Shurigin 的 Django 開發人員犯的 10 大錯誤。