使用Python和Django进行性能测试和优化的指南 _教程合集

本文概述

介绍我们的应用
性能优化就是衡量
1.优化数据库查询
2.优化你的代码
总结

唐纳德·克努思(Donald Knuth)说：” 过早的优化是万恶之源。” 但是有时候, 通常是在高负载的成熟项目中, 不可避免地需要进行优化。在本文中, 我想谈谈优化Web项目代码的五种常用方法。我将使用Django, 但其他框架和语言的原理应相似。在本文中, 我将使用这些方法将查询的响应时间从77秒减少到3.7秒。

文章图片
该示例代码改编自与我合作过的真实项目, 并演示了性能优化技术。如果你希望自己继续学习并查看结果, 可以在GitHub上获取其初始状态的代码, 并在进行后续操作时进行相应的更改。我将使用Python 2, 因为某些第三方软件包尚不适用于Python 3。
介绍我们的应用我们的网络项目仅跟踪每个国家的房地产报价。因此, 只有两种模型：

# houses/models.py from utils.hash import Hasherclass 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.nameclass 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提供了从其继承了哈希属性的任何模型, 该属性包含实例的主键和模型的内容类型。通过用哈希替换敏感数据, 例如实例ID, 可以隐藏它们。当你的项目具有多个模型并且你需要一个集中的地方来散列并决定如何处理不同类的不同模型实例时, 它也可能很有用。请注意, 对于我们的小型项目, 实际上不需要散列, 因为没有散列就可以进行处理, 但这将有助于演示一些优化技术, 因此我将其保留在那里。
这是Hasher类：

# utils/hash.py import basehashclass 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 = Nonedef get_queryset(self): country = get_object_or_404(Country, pk=self.country) queryset = self.model.objects.filter(country=country) return querysetdef 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)

现在, 我们用一些数据填充数据库(使用工厂男孩生成的总数为100, 000个房屋实例：一个国家为50, 000, 另一个国家为40, 000, 第三国为10, 000), 并准备测试我们的应用程序的性能。
性能优化就是衡量我们可以在项目中衡量以下几项：

执行时间处理时间
代码行数
函数调用次数
分配的内存
等等。

但是, 并非所有这些指标都与衡量我们的项目执行情况有关。一般来说, 有两个最重要的主要指标：某项执行的时间和所需的内存。
在Web项目中, 响应时间(服务器接收由某个用户的操作生成的请求, 处理该请求并返回结果所需要的时间)通常是最重要的指标, 因为它不会使用户在等待时感到无聊进行响应, 然后切换到其浏览器中的另一个标签。
在编程中, 分析项目绩效称为分析。为了描述API端点的性能, 我们将使用Silk包。安装它并进行/ api / v1 / houses /？country = 5T22RI调用(对应于具有50, 000个房屋条目的国家的哈希), 我们得到以下信息：
200个
/ api / v1 / houses /
整体77292ms
15854ms查询
50004个查询
总体响应时间为77秒, 其中16秒用于数据库中的查询, 该数据库中总共进行了50, 000个查询。数量如此之多, 还有很大的改进空间, 所以让我们开始吧。
1.优化数据库查询关于性能优化的最常见技巧之一是确保对数据库查询进行优化。这种情况也不例外。此外, 我们可以对查询做几件事以优化响应时间。
1.1一次提供所有数据
仔细研究一下这50, 000个查询是什么, 你可以看到这些都是关于houses_country表的多余查询：
200个
/ api / v1 / houses /
整体77292ms
15854ms查询
50004个查询

At	桌子	Execution Time (ms)
+0:01:15.874374	” houses_country”	0.176
+0:01:15.873304	” houses_country”	0.218
+0:01:15.872225	” houses_country”	0.218
+0:01:15.871155	” houses_country”	0.198
+0:01:15.870099	” houses_country”	0.173
+0:01:15.869050	” houses_country”	0.197
+0:01:15.867877	” houses_country”	0.221
+0:01:15.866807	” houses_country”	0.203
+0:01:15.865646	” houses_country”	0.211
+0:01:15.864562	” houses_country”	0.209
+0:01:15.863511	” houses_country”	0.181
+0:01:15.862435	” houses_country”	0.228
+0:01:15.861413	” houses_country”	0.174

这个问题的根源在于, 在Django中, 查询集是惰性的。这意味着查询集不会被评估, 并且在你真正需要获取数据之前不会访问数据库。同时, 它仅获取你告诉它的数据, 如果需要任何其他数据, 则进行后续请求。
这就是我们的情况。通过House.objects.filter(country = country)获取查询集时, Django将获取给定国家/地区中所有房屋的列表。但是, 序列化房屋实例时, HouseSerializer需要房屋的国家/地区实例, 才能计算序列化器的国家/地区字段。由于查询集中没有国家/地区数据, 因此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

让我们看看这如何影响性能：
200个
/ api / v1 / houses /
整体35979毫秒
102ms查询
4查询
总体响应时间降至36秒, 仅花费4个查询就在数据库中花费的时间约为100ms！这是个好消息, 但我们可以做得更多。
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

这样可以将查询所花费的时间减少一半, 这是很好的, 但是50ms并没有那么多。总体时间也略有下降, 但仍有更多空间可以削减。
200个
/ api / v1 / houses /
总共33111ms
52ms查询
4查询
2.优化你的代码你无法无限优化数据库查询, 而我们的最后结果表明了这一点。即使我们假设将查询所花费的时间减少到0, 我们仍将面临等待半分钟才能获得响应的现实。现在该切换到另一个优化级别：业务逻辑。
2.1简化代码
有时, 第三方软件包会为简单任务带来很多开销。这样的例子之一就是我们返回序列化房屋实例的任务。
Django REST框架很棒, 提供了许多有用的功能。但是, 我们现在的主要目标是减少响应时间, 因此它是进行优化的理想选择, 尤其是序列化的对象非常简单。
为此, 我们编写一个自定义序列化程序。为简单起见, 我们将使用一个静态方法来完成这项工作。实际上, 你可能希望具有相同的类和方法签名, 以便能够互换使用序列化程序：

# 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 = Nonedef get_queryset(self): country = get_object_or_404(Country, pk=self.country) queryset = self.model.objects.filter(country=country) return querysetdef 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 = http://www.srcmini.com/self.plain_serializer_class.serialize_data(queryset)# < -- serializereturn Response(data)

200个
/ api / v1 / houses /
整体17312毫秒
38ms查询
4查询
现在看起来好多了。由于我们没有使用DRF序列化器代码, 因此响应时间几乎减少了一半。
另一个可测量的结果(在请求/响应周期内进行的函数调用总数)从15, 859, 427个调用(从上面1.2节中的请求)下降到9, 257, 469个调用。这意味着所有函数调用中约有1/3是由Django REST Framework进行的。
2.2更新/替代第三方程序包
上面描述的优化技术是最常见的, 你无需进行深入分析和思考即可完成这些优化技术。但是, 17秒仍然感觉很长；为了减少此数量, 我们将需要更深入地研究代码并分析幕后情况。换句话说, 我们将需要分析我们的代码。
你可以使用内置的Python探查器自己进行分析, 也可以使用一些第三方程序包(使用内置的Python探查器)。由于我们已经使用了Silk, 它可以对代码进行概要分析并生成一个二进制概要文件, 我们可以进一步对其进行可视化。有几种可视化程序包可以将二进制配置文件转换为一些有见地的可视化文件。我将使用snakeviz软件包。
这是上面最后一个请求的二进制配置文件的可视化图, 与视图的分派方法相关联：

文章图片
从上到下是调用堆栈, 显示文件名, 方法/函数名称及其行号以及在该方法中花费的相应累积时间。现在, 可以很容易地看出, 大部分时间都用于计算哈希值(紫色的__init__.py和primes.py矩形)。
目前, 这是我们代码中的主要性能瓶颈, 但同时它并不是我们的代码, 它是第三方程序包。
在这种情况下, 我们可以做的事情有限：

检查软件包的新版本(希望它具有更好的性能)。
寻找另一个在我们需要的任务上表现更好的软件包。
编写我们自己的实现, 它将击败我们当前使用的程序包的性能。

对我来说幸运的是, 有一个负责哈希处理的basehash软件包的较新版本。该代码使用v.2.1.0, 但是有一个v.3.0.4。当你能够更新到软件包的较新版本时, 这种情况在你处理现有项目时更可能发生。
查看v.3的发行说明时, 这句话听起来很有希望：

使用素数算法进行了大修。如果gmpy2在系统上可用, 则包括对gmpy2的支持(增加)。

让我们找出答案！

pip install -U basehash gmpy2

200个
/ api / v1 / houses /
整体7738毫秒
59ms查询
4查询
我们将响应时间从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

如你所见, 基本实例的初始化需要调用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]

200个
/ api / v1 / houses /
整体3766毫秒
38ms查询
4查询
最终结果不到4秒, 比我们开始时要小得多。使用缓存可以进一步优化响应时间, 但是在本文中我不会解决。
总结性能优化是分析和发现的过程。没有适用于所有情况的硬性规则, 因为每个项目都有自己的流程和瓶颈。但是, 你应该做的第一件事就是分析你的代码。如果在这样一个简短的示例中, 我可以将响应时间从77秒减少到3.7秒, 那么大型项目就具有更大的优化潜力。
【使用Python和Django进行性能测试和优化的指南】如果你有兴趣阅读更多与Django相关的文章, 请查看srcmini Django开发人员Alexandra Shurigin撰写的Django开发人员的十大错误。