GraphQL 教程(六)—— N+1问题和缓存等问题

N+1问题

考虑这样一个常见应用场景,前端页面上需要展示一个文章列表,其中包括了文章的标题,并且会同时显示每篇文章的作者名。那么我们可能会按下面几种方案来设计我们的API。

方案一: 对于Restful,设计下面两个接口,客户端总共需要请求1+N次接口,查询数据库1+N次。

1
2
/articles	# 文章列表接口
/articles/{article_id}/author # 获取文章作者信息接口

方案二: 对于Restful,为避免多次频繁请求获取文章作者信息的接口,可以采取下面两种方式,不过都得修改后端代码,总共需要请求1次接口,查询数据库1+N次。

1
2
/articles?withauthor=true		# 在原接口的基础上添加一个参数表示是否同时获取文章的作者信息
/articlesWithAuthor # 直接添加一个新的能够通过获取作者信息的文章列表接口

方案三:对于GraphQL,我们不需要做任何的修改,因为我们早已经定义好这样的Schema将article与author进行了关联:

1
2
3
4
5
6
7
8
9
10
11
# schemas/article.py
class ArticleSchema(SQLAlchemyObjectType):
author = graphene.Field("schemas.AuthorSchema", description="文章作者信息")

def resolve_author(self, info):
# 下面这一行请自行在demo中去掉注释
return AuthorManager.get_one(id=self.author_id)

class Meta:
model = ArticleModel
description = "文章Schema"

默认是打开了SQL日志的,所以我们可以看到最终的结果仍然是1次请求,数据库查询却有1+N次:

1
2
3
4
5
6
SELECT * FROM `articles` LIMIT 0, 20;
SELECT * FROM `authors` WHERE `id` = 1;
SELECT * FROM `authors` WHERE `id` = 2;
SELECT * FROM `authors` WHERE `id` = 3;
...
SELECT * FROM `authors` WHERE `id` = N;

为了防止N+1问题,社区为GraphQL提供了一个解决方案: DataLoader。其原理就是,在需要查询数据库的时候将查询进行延迟,等到拿到所有的查询需求之后再一次性查询出来。在graphene里面,批量查询可以这样写:

1
2
3
4
5
6
# managers/author.py
class AuthorsDataLoader(DataLoader):
def batch_load_fn(self, ids):
query = DBSession().query(AuthorModel).filter(AuthorModel.id.in_(ids))
articles = dict([(article.id, article) for article in query.all()])
return Promise.resolve([articles.get(id, None) for id in ids])

最终,仅需要两次数据库查询就完成了两个批量查询,即:

1
2
SELECT * FROM `articles` LIMIT 0, 20;
SELECT * FROM `authors` WHERE `id` IN (1, 2, 3, ..., N);

至此,我们通过批量查询的方式完成了减少冗余查询的功能。但是性能问题依然存在。

缓存问题

仔细看上面的两条SQL,第一条查询由于是Graphene框架自己帮我完成了解析然后进行查询,所以其实并不是SELECT *,但是多级查询的时候,后面的查询逻辑是我自己写的,为了方便我就写的SELECT *,大家都应该知道这样子查询数据库会有很大的性能隐患,访问量一旦大了数据库压力会加倍增长。不过好处是,DataLoader本身具有缓存结果的功能,它缓存的是SELECT * FROM authors WHERE id =N的结果,而不是批量查询的结果,所以,下一次即使是不一样的IN列表,依然会有部分能够使用缓存。DataLoader默认本身是开启缓存的,如果想自己用Redis等来实现对象的缓存,可以在DataLoader初始化的时候将cache设置为False:

1
2
3
4
5
6
7
8
9
10
# schemas/.__init__.py
def get_dataloaders():
return {
"ArticlesDataLoader": ArticlesDataLoader(cache=False),
"AuthorsDataLoader": AuthorsDataLoader(cache=False),
"CommentsDataLoader": CommentsDataLoader(),
"ArticleCommentsDataLoader": ArticleCommentsDataLoader(),
"OrdinaryWritersDataLoader": OrdinaryWritersDataLoader(cache=True),
"ProfessionalWritersDataLoader": ProfessionalWritersDataLoader(cache=False),
}

验证问题/限流问题

由于只有一个查询入口,不能像Restful那样针对每个接口进行单独的验证或者限流。但是,正如我之前所说的,一个简单的方法就是强制让用户传入规范命名的查询语句,通过命名来进行后续的判断,就能方便地达到我们的要求。

分页查询问题

这个有很多种实现方式,大多数是用类似数据库limit begin, end的方式,但是为了兼容Restful的习惯,我使用的仍然是Restful风格的分页方式,返回结果也与Restful类似,可以参考代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# schemas/__init__.py
class PageInfoSchema(ObjectType):
"""
专用于分页的schema
"""
total = graphene.Int(description="总条数")
current_page = graphene.Int(description="当前页码")
per_page = graphene.Int(description="每页数量")
total_pages = graphene.Int(description="总共页码数量")

@staticmethod
def paginate(total: int, current_page: int, per_page: int):
"""
:param total: 总共条数
:param current_page: 当前页码
:param per_page: 每页数量
:return:
"""
return PageInfoSchema(
total=int(total),
current_page=int(current_page),
per_page=int(per_page),
total_pages=math.ceil(int(total) / int(per_page)),
)

GraphQL 教程demo地址
GraphQL 教程(一)——What’s GraphQL
GraphQL 教程(二)—— GraphQL 生态
GraphQL 教程(三)—— GraphQL 原理
GraphQL 教程(四)—— Python Demo搭建
GraphQL 教程(五)—— 增删改查语法及类型系统
GraphQL 教程(六)—— N+1问题和缓存等问题