为何要研究SQLAlchemy 的内存消耗问题？因为SQLAlchemy在应用中，绝大多数问题体现在应用人员对SQLAlchemy 的内存消耗问题不认知、不重视、不处理，最终造成整个系统的大问题，使SQLAlchemy 的性能大打折扣，最终影响了SQLAlchemy的在您手中的可用性。
通过以下解决问题的手法，可以有效控制 SQLAlchemy 的内存消耗，提高应用程序的性能和稳定性。
1. 连接池相关内存消耗
原理
SQLAlchemy 使用连接池来管理数据库连接，连接池会在内存中维护一定数量的数据库连接，以避免频繁创建和销毁连接带来的开销。连接池的大小、超时时间等配置会影响内存消耗。
示例
from sqlalchemy import create_engine# 创建一个连接池大小为 10 的数据库引擎
engine = create_engine('mysql+pymysql://user:password@host/dbname', pool_size=10)在这个例子中，连接池会在内存中保留 10 个数据库连接，每个连接会占用一定的内存空间。连接池越大，占用的内存就越多。
应对策略
不要试图对数据库进行长连接，例如：终端程序启动就连接数据库，终端程序退出才关闭连接，这是最不可取的，这会导致大量的数据库长连接。如果您不是使用SQLAlchemy，而是手动管理数据库连接，并进行了长连接，那么系统的噩梦很可能就此开始。
数据库的连接使用应该是：需要数据库操作时连接数据库，数据库操作完毕后就管理闲置的连接。SQLAlchemy可以自动的利用数据库连接池中的空闲连接。根据实际业务需求合理配置连接池大小。如果并发访问量较小，可以适当减小连接池大小；如果并发访问量较大，可以增加连接池大小，但要注意不要过度分配内存。
2. 对象管理导致的内存消耗
原理
当使用 SQLAlchemy 从数据库中查询数据时，会将查询结果映射为 Python 对象。这些对象会在内存中占用一定的空间，尤其是当查询返回大量数据时，内存消耗会显著增加。如果您不小心返回了大量数据（尤其是在处理大数据时），您的这样一次无心之失，足以让整个系统死机。
示例
from sqlalchemy.orm import sessionmaker
from your_model_module import User # 假设 User 是一个 SQLAlchemy 模型类Session = sessionmaker(bind=engine)
session = Session()# 查询所有用户，如果User表的记录条数很多（超过1万条会极度影响性能，超过10万条会迟滞系统，100万条直接死机）
users = session.query(User).all()应对策略
- 分批查询：对于大量数据的查询，采用分批查询的方式，每次只查询一部分数据进行处理，处理完后释放相关内存。例如：
batch_size = 100
offset = 0
while True:users = session.query(User).limit(batch_size).offset(offset).all()if not users:break# 处理当前批次的用户数据for user in users:# 处理逻辑passoffset += batch_size- 及时释放对象：在使用完对象后，及时释放对它们的引用，让 Python 的垃圾回收机制能够回收相关内存。例如，在处理完一批数据后，将列表置为
None。
3. 查询操作中的内存消耗
原理
复杂的查询操作，尤其是涉及大量数据的连接查询、子查询等，可能会导致 SQLAlchemy 在内存中进行复杂的计算和数据处理，从而增加内存消耗。
示例
from sqlalchemy.orm import joinedload# 进行一个复杂的连接查询
orders = session.query(Order).options(joinedload(Order.user)).all() 在这个例子中，使用 joinedload 进行连接查询，SQLAlchemy 会将 Order 对象和关联的 User 对象一次性加载到内存中，但如果 Order 和 User 表的数据量都很大，内存消耗会明显增加（此时要么通过with_entities削减加载的数据量，要么采用流式查询）。
应对策略
- 优化查询语句：尽量避免不必要的连接和子查询，只查询需要的字段。例如，使用
with_entities方法指定要查询的字段：
orders = session.query(Order).with_entities(Order.id, Order.amount).all()- 使用流式查询：对于大数据量的查询，可以使用流式查询，避免将所有数据一次性加载到内存中。例如：
for order in session.query(Order).yield_per(100):# 处理每个订单pass4. 关联关系处理的内存消耗
原理
在处理对象之间的关联关系时，SQLAlchemy 可能会自动加载关联对象，这会增加内存消耗。例如，当使用 relationship 定义关联关系时，如果没有合理配置加载策略，可能会导致大量关联对象被加载到内存中。
示例
class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)orders = relationship("Order")user = session.query(User).first()
# 访问用户的订单，可能会导致所有订单对象被加载到内存中
for order in user.orders:pass应对策略
- 合理配置加载策略：使用
joinedload、subqueryload等加载策略来控制关联对象的加载时机和方式。例如，使用joinedload一次性加载关联对象：
user = session.query(User).options(joinedload(User.orders)).first()- 延迟加载：对于不需要立即使用的关联对象，可以配置为延迟加载，只有在实际访问时才加载到内存中。例如，在
relationship中使用lazy='dynamic'：
class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)orders = relationship("Order", lazy='dynamic')5.分批操作和流式操作
在 SQLAlchemy 里，分批查询和流式查询都是用于处理大量数据查询的技术手段，它们在实现方式、内存使用、性能表现、适用场景等方面存在一定差异。
(1)实现方式
分批查询
分批查询是将大数据集分割成多个较小的数据批次进行查询。一般通过设置 limit 和 offset 参数来实现，每次查询返回固定数量的记录，处理完这批记录后，再调整 offset 值进行下一批次的查询，直至查询完所有数据。
示例代码
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from your_model_module import User # 假设 User 是模型类engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()batch_size = 100
offset = 0
while True:users = session.query(User).limit(batch_size).offset(offset).all()if not users:break# 处理当前批次的用户数据for user in users:print(user)offset += batch_sizesession.close()流式查询
流式查询借助 yield_per 方法，以流式方式逐行处理查询结果。数据库游标会逐行从数据库中读取数据，每读取一定数量（yield_per 指定的数量）的记录后就将其返回，而不是一次性把所有数据加载到内存中。
示例代码
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from your_model_module import User # 假设 User 是模型类engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()for user in session.query(User).yield_per(100):print(user)session.close()(2)内存使用
分批查询
分批查询每次仅将一批数据加载到内存中，相较于一次性加载全量数据，能显著减少内存占用。不过，每批数据仍需全部加载到内存后再进行处理，若批次大小设置不合理（过大），仍可能导致内存占用过高。
流式查询
流式查询以逐行方式处理数据，每次只将少量数据加载到内存中，内存占用非常低，即使处理超大规模数据集，也能有效避免内存溢出问题。
(3)性能表现
分批查询
分批查询需要多次与数据库交互，每次查询都有一定的开销，如建立连接、执行查询语句等。而且，随着 offset 值的增大，查询效率可能会逐渐降低，因为数据库需要跳过大量记录来定位到指定偏移量的位置。
流式查询
流式查询与数据库保持持续连接，逐行读取数据，减少了多次查询的开销，在处理大数据集时性能优势明显。但流式查询依赖数据库的游标机制，若数据库游标性能不佳，可能会影响整体查询效率。
(4)适用场景
分批查询
- 适用于需要对每一批数据进行批量处理的场景，例如批量更新、批量插入等操作。
- 当需要对数据进行分页展示时，分批查询可以方便地实现分页逻辑。
流式查询
- 适合处理超大规模数据集，尤其是在内存资源有限的情况下，流式查询能有效避免内存问题。
- 适用于实时数据处理场景，如实时数据分析、日志处理等，可在获取数据的同时立即进行处理，无需等待全量数据加载完成。
分批查询和流式查询各有优劣，在实际应用中，需要根据数据规模、内存资源、处理需求等因素综合考虑，选择合适的查询方式。
6.一次性加载关联 vs 动态加载关联
在 SQLAlchemy 中，joinedload 和 dynamic 是两种不同的关联对象加载策略，它们在资源消耗方面各有特点，具体哪种更省资源取决于具体的使用场景，下面从内存、数据库查询、CPU 等资源消耗维度详细分析。
(1)joinedload 加载策略
原理
joinedload 是一种立即加载策略，它会使用 SQL 的 JOIN 操作在一次数据库查询中同时获取主对象和关联对象的数据。查询结果会被一次性加载到内存中，并且关联对象会被直接关联到主对象上。
示例代码
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, joinedload
from sqlalchemy.ext.declarative import declarative_base# 创建数据库引擎，使用 SQLite 内存数据库
engine = create_engine('sqlite:///:memory:')
# 创建基类
Base = declarative_base()# 定义 User 类
class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(50))# 定义与 Order 的关联关系orders = relationship("Order")# 定义 Order 类
class Order(Base):__tablename__ = 'orders'id = Column(Integer, primary_key=True)order_number = Column(String(20))user_id = Column(Integer, ForeignKey('users.id'))# 创建表
Base.metadata.create_all(engine)# 创建会话
Session = sessionmaker(bind=engine)
session = Session()# 使用 joinedload 一次性加载用户及其关联的订单
users = session.query(User).options(joinedload(User.orders)).all()
for user in users:for order in user.orders:print(f"User: {user.name}, Order: {order.order_number}")session.close()资源消耗情况
- 内存消耗：由于
joinedload会一次性将主对象和关联对象的数据都加载到内存中，如果关联对象数量较多或者数据量较大，会占用较多的内存。例如，一个用户关联了大量的订单记录，使用joinedload会将所有订单数据都加载到内存中。 - 数据库查询消耗：只进行一次数据库查询，减少了与数据库的交互次数，降低了数据库的负载。但是，如果关联表的数据量很大，查询语句可能会变得复杂，导致查询时间增加。
- CPU 消耗：由于只进行一次查询和数据处理，CPU 在查询和数据转换方面的计算量相对较小。
(2)dynamic 加载策略
原理
dynamic 是一种延迟加载策略，它返回一个可查询对象（Query 对象），而不是直接加载关联对象。当需要访问关联对象时，会根据具体的查询条件进行按需查询，每次只查询需要的数据。
示例代码
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base# 创建数据库引擎，使用 SQLite 内存数据库
engine = create_engine('sqlite:///:memory:')
# 创建基类
Base = declarative_base()# 定义 User 类
class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(50))# 定义与 Order 的关联关系，使用 dynamic 加载策略orders = relationship("Order", lazy='dynamic')# 定义 Order 类
class Order(Base):__tablename__ = 'orders'id = Column(Integer, primary_key=True)order_number = Column(String(20))user_id = Column(Integer, ForeignKey('users.id'))# 创建表
Base.metadata.create_all(engine)# 创建会话
Session = sessionmaker(bind=engine)
session = Session()# 查询用户
users = session.query(User).all()
for user in users:# 按需查询用户的订单user_orders = user.orders.filter(Order.order_number.like('123%')).all()for order in user_orders:print(f"User: {user.name}, Order: {order.order_number}")session.close()资源消耗情况
- 内存消耗：由于是按需查询，只有在实际访问关联对象时才会加载数据到内存中，不会一次性加载大量数据，因此内存消耗相对较低。例如，只需要查看部分订单记录时，不会将所有订单数据都加载到内存中。
- 数据库查询消耗：可能会进行多次数据库查询，增加了与数据库的交互次数，提高了数据库的负载。但是，每次查询的数据量较小，查询语句相对简单，查询时间可能会较短。
- CPU 消耗：由于需要多次进行查询和数据处理，CPU 在查询和数据转换方面的计算量相对较大。
(3)对比总结
- 当关联对象数据量较小且需要一次性访问所有关联对象时：
joinedload更省资源。因为它只进行一次数据库查询，减少了数据库的交互次数，虽然会占用一定的内存，但整体的资源消耗相对较低。 - 当关联对象数据量较大且只需要访问部分关联对象时：
dynamic更省资源。它按需查询，避免了一次性加载大量数据到内存中，降低了内存消耗，虽然会增加数据库的交互次数，但每次查询的数据量较小。
选择 joinedload 还是 dynamic 加载策略需要根据具体的业务场景和数据特点来决定，以达到最优的资源利用效果。
![[LVGL] 在VC_MFC中移植LVGL](https://i-blog.csdnimg.cn/direct/53c7f1b2a72442f694d54ebab1e3bf1d.png)
![[RabbitMQ] RabbitMQ常见面试题](https://i-blog.csdnimg.cn/direct/7c0071eb6c214ee18562419f2a3fbfb5.png)
