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 大错误。