当前位置：首页 > news >正文

有关爬虫中数据库的封装——单线程爬虫

news 来源：原创 2025/4/29 7:37:54

在编写爬虫程序时，数据存储是一个非常重要的环节。尤其是在单线程爬虫中，虽然并发量不大，但如果每次操作数据库都直接连接、查询、关闭，不仅代码臃肿，还容易出错，甚至影响爬虫的稳定性。

因此，封装数据库连接成为了一个非常有必要的优化步骤。今天，我就带大家详细了解一下，在单线程爬虫项目中，如何优雅地封装数据库操作！

为什么要封装数据库？

在单线程爬虫中，通常一边发送请求，一边将爬取到的数据写入数据库。直接写数据库容易遇到以下问题：

重复代码：每次都要手写连接、关闭，冗余且易错。
资源浪费：频繁打开、关闭数据库连接，效率低下。
异常处理：连接超时、插入失败等问题处理复杂。
维护困难：后期改动数据库逻辑时，修改的地方过多。

而封装之后，我们只需要简单调用统一的接口，大大提高了代码整洁性、健壮性和扩展性。

封装目标

我们的数据库封装类需要满足以下要求：

支持自动连接、重连；
支持执行查询和执行写入；
支持最大空闲时间控制（避免连接超时）；
捕获异常并打印日志；
尽量简单，适合单线程场景，无需加锁。

数据库封装代码示例

这里，我给出一个实用的封装示例（以 pymysql 为例，连接的是MySQL数据库）。

from ezpymysql import Connection# 1. 创建数据库连接
db = Connection(host="127.0.0.1",   # 数据库地址database="test_db", # 你的数据库名字user="root",        # 数据库用户名password="your_password",  # 数据库密码port=3306           # 可选，如果不是标准3306端口
)# 2. 查询数据
print("查询表中所有数据:")
rows = db.query("SELECT * FROM users")
for row in rows:print(row)# 3. 插入数据
print("\n插入一条数据:")
new_user = {"username": "alice","email": "alice@example.com","age": 25
}
insert_id = db.table_insert("users", new_user)
print(f"插入成功，ID：{insert_id}")# 4. 检查某条数据是否存在
print("\n检查用户是否存在:")
user = db.table_has("users", "username", "alice")
if user:print("用户已存在:", user)
else:print("用户不存在")# 5. 更新数据
print("\n更新用户年龄:")
db.table_update(table_name="users",updates={"age": 26},field_where="username",value_where="alice"
)
print("更新成功！")# 6. 查询更新后的数据
print("\n查询更新后的用户:")
updated_user = db.get("SELECT * FROM users WHERE username=%s", "alice")
print(updated_user)# 7. 关闭连接
db.close()
print("\n数据库连接已关闭。")

如何在爬虫中使用？

举个简单例子：

# 1. 创建数据库连接
db = Connection(host="127.0.0.1",   # 数据库地址database="test", # 你的数据库名字user="root",        # 数据库用户名password="5201314",  # 数据库密码port=3306           # 可选，如果不是标准3306端口
)# 2. 查询数据
print("查询表中所有数据:")
rows = db.query("SELECT * FROM account")
for row in rows:print(row)# 3. 插入数据
print("\n插入一条数据:")
new_user = {"name": "alice","money": 25
}
insert_id = db.table_insert("account", new_user)
print(f"插入成功，ID：{insert_id}")# 4. 检查某条数据是否存在
print("\n检查用户是否存在:")
user = db.table_has("account", "name", "张三")
if user:print("用户已存在:", user)
else:print("用户不存在")# 5. 更新数据
print("\n更新用户年龄:")
db.table_update(table_name="account",updates={"name": '账号'},field_where="name",value_where="张三"
)
print("更新成功！")# 6. 查询更新后的数据
print("\n查询更新后的用户:")
updated_user = db.get("SELECT * FROM account WHERE name=%s", "账号")
print(updated_user)# 7. 关闭连接
db.close()
print("\n数据库连接已关闭。")