Python 项目提速技巧：连接复用
The following article is from 质量价值 Author 质量价值
为什么需要连接复用
接口、UI的测试用例中都会有大量的IO操作，比如HTTP、RPC调用、数据库查询等，这是典型的IO密集型任务，对自动化效率有追求的测试工程师应该思考一个问题：如何让用例执行更加地有效率（快）？
抛出的这个问题其实很大，从验证策略、用例设计、IO优化、用例分发方式等角度都可以讲，我不准备在这篇文章里完整的阐述，只挑出一个点：连接复用。
这里的连接可以存在于以下地方：
HTTP连接 RPC连接（http、socket都可能） 中间件连接（数据库、缓存服务等连接，可简化为TCP） UI自动化的Appium、Selenium对象（webdriver协议）
连接复用（以TCP为例）的好处可以大幅度降低TCP三次握手、四次挥手的次数以实现对用例消耗时间的降低，举一个很简单的例子：比如一个mysql client的建链跟关闭连接各需要10ms，当你存在10000多条用例，并且平均每个用例需要2次mysql查询操作，那整个用例执行时间可以降低400秒。对于做惯了UI自动化测试的童鞋而言，UI自动化执行时间往往以分钟、小时为计量单位，这400秒时间的减少似乎并不明显。这点我承认，但是对于下沉至接口层的自动化，完全可以相信一个业务场景用例能在一秒内验证完成，能压榨出400秒时间就是非常大的优化。
而且我相信，你在每一点上都比别人多想一点多做一点，这些点点滴滴的积累、沉淀就会变成你的绝对优势。
不经意来了碗鸡汤，回到正题：连接复用。
一般操作
对于测试人员而言，要实现『连接复用』最简单的办法对高度抽象的应用对象的复用，你不用过多去考虑实现层面的细节，比如连接池等。比如我之前在接口封装的基石：requests.Session介绍过通过requests.Session来实现HTTP连接的复用，当你所有的HTTP接口调用都基于同一个requests.Session来调用的话，那其实就实现了全局的『HTTP连接复用』能力。
HTTP调用是有状态的，所以是否应该使用同一个requests.Session来调用，要视实际情况来判断，本文不多展开。
下文我以mysql的连接复用（使用pymysql库）来作介绍。
先看一个简单的例子：
import pymysql
conn = pymysql.Connect(host="your_host", user="root", password="your_password", database="your_db")
with conn.cursor() as curosr:
curosr.execute("select * from user limit 1")
ret = curosr.fetchone()
conn.close()
当你在测试用例里需要进行SQL查询时，可以copy上面的代码去做相关的操作，一个两个用例还好，但是用例成千上百时，我就算不讲『连接复用』概念，我也相信你也觉得这样的代码很臃肿，需要优化。
大部分测试人员会使用这个办法：在测试启动时，连接一次数据库（pymysql.Connect），然后把返回的pymysql.Connection作为一个全局对象供其他用例使用，这就是连接复用的思路。
现实问题
但往往我们实际的应用场景可能更加丰富、复杂，比如：
需要访问同一数据库实例的不同database 需要不同账号访问同一数据库实例（权限问题） 需要访问不同数据库实例
第一种情况还好，访问不同database可以共用一个连接，只需要使用use <db>来切换。另外两种呢？如果按照上面提到的思路也有办法：在测试启动时，建立不同账号建立对不同数据库实例的连接，都是作为『全局的数据连接』，而在使用时（用例逻辑层）去挑选适合你当前用例的连接对象。
按照上面办法的需要注意：因为需要用例设计者人工去选择合适的pymysql.Connection对象，当对象较多时，用例设计者很可能选错，导致用例失败。
我这里更推荐另外种做法——懒加载，你不需要测试一开始就建立所有的mysql连接，而是在你的用例里需要去查询数据库时，显式地传入连接信息（地址、用户名等）去建立连接，这样就可以避免使用了错误的数据库连接信息了，如：
def test_user():
conn = pymysql.Connect(host="host1", user="root", password="pwd", database="ddd")
with conn.cursor() as curosr:
curosr.execute("select * from user limit 1")
ret = curosr.fetchone()
assert ret
def test_tag():
conn = pymysql.Connect(host="host1", user="root", password="pwd", database="ddd")
with conn.cursor() as curosr:
curosr.execute("select * from tag")
ret = curosr.fetchone()
assert ret
但这样就带出来问题了：明明要讲连接复用，为什么还要在每一个用例里去初始化数据库连接？
单例模式
上面一大段其实就为了引出设计模式里非常重要的一种——单例模式：单例模式确保某一个类只有一个实例，而且自行实例化并向整个系统提供这个实例，这个类称为单例类，它提供全局访问的方法。
也就是说会存在以下的逻辑：
单例模式的实现办法有很多种，比如：
def singleton(cls):
instances = dict()
@functools.wraps(cls)
def _singleton(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return _singleton
@singleton
class MySQLConnectionProxy:
def __init__(self, *args, **kwargs):
self._conn = pymysql.Connect(*args, **kwargs)
def __getattr__(self, item):
return getattr(self._conn, item)
上面的例子还用到了代理模式，之后会有更详细的讲解
对应的测试用例可以改成这种方式：
def test_user():
conn = MySQLConnectionProxy(host="host1", user="root", password="pwd", database="ddd")
with conn.cursor() as curosr:
curosr.execute("select * from user limit 1")
ret = curosr.fetchone()
assert ret
再结合我们上一讲的如何让用例支持多环境？，我们可以把数据库连接信息抽象出来，从而变成：
def test_user():
conn = MySQLConnectionProxy(**entrypoints.mysql)
with conn.cursor() as curosr:
curosr.execute("select * from user limit 1")
ret = curosr.fetchone()
assert ret
单例模式的变种
但上面单例模式的代码其实并没有解决多用户、多数据库连接的问题，该怎么解决呢？思路稍微变通下不难发现：应该只对使用相同连接信息的调用使用单例模式。
这话说点有点抽象，具象一点就是：当数据库host、端口、用户名、密码相同时，返回一个已建立的pymysql.Connection，也可以用下图来加深理解：
所以可以进一步优化上面的代码：
def singleton_mysql_instance(cls):
instances = dict()
@functools.wraps(cls)
def _singleton(*args, **kwargs):
conn_params = (kwargs.get("host"), kwargs.get("port"), kwargs.get("user"), kwargs.get("password"))
p = hash(conn_params)
if p not in instances:
instances[p] = cls(*args, **kwargs)
return instances[p]
return _singleton
为了方便理解，我简化了实现，也尽量少去使用inspect、magic method这些能力
连接复用的注意点
单例模式下全局只维护了一个实例，这个时候一定要慎重考虑一个问题：如果该对象被执行了析构函数或者像mysql的连接被关闭了（不管是主动还是被动），如何能够发现或者重新构造？
另外还有一个问题，全局只维护了一个实例，在多线程模型下，是否能够保证对它的操作是线程安全的？（thread safety）
受限于篇幅，这两个问题这边不展开讨论了，感兴趣的可以留言一起讨论。
优质文章，推荐阅读：