mssql-python 应用的性能调优

mssql-python 驱动提供了多种功能和模式以优化 SQL Server 应用性能，包括连接池、查询优化和批量操作。

连接管理

使用连接池

内置连接池。当你调用 conn.close() 时，连接会返回连接池以供重用，而不是被销毁，因此后续对 connect() 的调用会跳过成本高昂的握手过程：

import mssql_python

def get_data():
    conn = mssql_python.connect(
        "Server=<server>.database.windows.net;Database=<database>;"
        "Authentication=ActiveDirectoryDefault;Encrypt=yes"
    )
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
        return cursor.fetchall()
    finally:
        conn.close()

为工作负载配置池大小

根据你的并发需求调整池大小。如果您的应用程序需要处理大量并发用户，请增大池大小。对于较轻的工作负载，较小的池可以节省服务器资源：

import mssql_python

mssql_python.pooling(
    max_size=50,      # Default is 100; reduce or increase for your workload
    idle_timeout=600  # Seconds before idle connections are recycled
)

在操作内重复使用连接

即使使用连接池机制，每次查询都创建一个新连接也会带来额外开销。相反，在逻辑操作期间保持单一连接：

# Bad: New connection per query
def bad_pattern(product_ids):
    for pid in product_ids:
        conn = mssql_python.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
        row = cursor.fetchone()
        process(row)
        conn.close()

# Good: Single connection for all queries
def good_pattern(product_ids):
    conn = mssql_python.connect(connection_string)
    cursor = conn.cursor()
    try:
        for pid in product_ids:
            cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE ProductID = %(pid)s", {"pid": pid})
            row = cursor.fetchone()
            process(row)
    finally:
        conn.close()

在长期运营的服务中保持连接畅通

持续运行的 Web 服务器、队列工作进程和定时作业应保持连接处于打开状态，而不是在每次操作时都重新连接再断开。开通连接涉及TCP握手、TLS协商和认证，具体取决于网络距离和认证方式，可能需要50-200毫秒。对于一个每小时处理数千条消息的队列工作者来说，这种开销会迅速累积起来。

在员工的整个生命周期内保持连接畅通，连接中断时再重新连接。在每次迭代之间稍作等待，以避免在队列为空时频繁请求服务器：

import mssql_python
import time

def run_worker(connection_string: str, poll_interval: float = 1.0):
    conn = None
    try:
        while True:
            try:
                if conn is None:
                    conn = mssql_python.connect(connection_string)
                cursor = conn.cursor()
                cursor.execute("SELECT TOP 1 * FROM dbo.JobQueue WHERE Status = 'Pending' ORDER BY CreatedDate")
                job = cursor.fetchone()
                if job:
                    try:
                        process_job(job)
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Done' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    except Exception:
                        cursor.execute("UPDATE dbo.JobQueue SET Status = 'Failed' WHERE JobID = %(job_id)s", {"job_id": job[0]})
                    conn.commit()
                else:
                    time.sleep(poll_interval)  # No work available, wait before polling again
            except mssql_python.OperationalError:
                # Connection lost, reconnect on next iteration
                conn = None
                time.sleep(poll_interval)
    finally:
        if conn is not None:
            conn.close()

启用连接池后（默认即启用），连接池会负责处理空闲连接。但如果你关闭了池化或使用单一专用连接，在你的连接字符串中设置Connection Timeout和Command Timeout，这样可以提前检测过时连接，而不是卡住。

查询优化

仅获取所需的数据

只选择应用使用的列可以减少网络传输、内存消耗和查询执行时间。

# Bad: Select all columns
cursor.execute("SELECT * FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})

# Good: Select specific columns
cursor.execute("""
    SELECT SalesOrderID, OrderDate, TotalDue
    FROM Sales.SalesOrderHeader
    WHERE CustomerID = %(customer_id)s
""", {"customer_id": 1})

使用适当的获取方法

驱动程序提供了几种获取方法。使用与你结果尺寸相匹配的那个：

fetchval() 返回单个标量值，且开销极低。
fetchall() 将整个结果集加载到内存中，这对小表来说效果很好。
fetchmany(n) 分批检索行，从而在处理大型结果集时保持内存使用恒定。

fetchmany() 的合适批处理大小取决于行宽。对于窄行（几个小列，每个大约1KB），1000行让每批大约有1MB内存。对于较宽的行和大字符串或二进制列，使用较小的批次。从1000个开始，根据你的数据进行调整。

def process_batch(rows):
    # Example: print each row. Replace with your own logic.
    for row in rows:
        print(row)

# Single value
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval()

# Small result set
cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
categories = cursor.fetchall()

# Large result set - process in batches
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader")
while True:
    batch = cursor.fetchmany(1000)
    if not batch:
        break
    process_batch(batch)

使用服务器端分页

不要先获取所有行再在 Python 中切片，而应使用 OFFSET/FETCH NEXT 仅检索所需的那一页数据。

def get_page(cursor, page: int, page_size: int = 50) -> list:
    """Get paginated results efficiently."""
    offset = (page - 1) * page_size

    cursor.execute("""
        SELECT ProductID, Name, ListPrice
        FROM Production.Product
        ORDER BY ProductID
        OFFSET %(offset)s ROWS
        FETCH NEXT %(page_size)s ROWS ONLY
    """, {"offset": offset, "page_size": page_size})

    return cursor.fetchall()

使用SET NOCOUNT开启

默认情况下，SQL Server 在每个 DML 语句后都会发送“受影响行”消息。 SET NOCOUNT ON 抑制这些消息并减少网络流量。这是一个会话级别的设置，所以连接后设置一次，而不是每次查询都嵌入。

# Set once after connecting
cursor.execute("SET NOCOUNT ON")

# All subsequent statements on this connection skip the row-count message
cursor.execute(
    "INSERT INTO Log (Message) VALUES (%(message)s)",
    {"message": "Log entry"}
)

选择合适的插入方式

驱动程序提供了三种数据插入方式，每种方式适用于不同的尺度：

方法	行计数	为什么
`execute()`	每次通话1行	适用于单行操作，例如表单提交或 API 处理函数，在这些场景中你需要立即获取插入后的 ID。
`executemany()`	约 10 到 1,000 行	采用列级参数绑定，吞吐量优于循环。将每行发送为参数化语句。
`bulkcopy()`	数百行及以上	采用TDS大批量插入协议，效率远高于逐行插入。最适合数据加载、迁移和批处理。

更多细节和示例请参见数据加载与移动模式。

使用 execute() 的单条插入

适用于需要立即获得结果的一次性插入操作。 Production.Product 有多个没有默认值的 NOT NULL 列，因此在 INSERT 语句中列出了所有这些列：

from datetime import datetime

cursor.execute(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (%(name)s, %(number)s, %(safety)s, %(reorder)s,
            %(cost)s, %(price)s, %(days)s, %(start)s)
    """,
    {
        "name": "Widget", "number": "WG-1001",
        "safety": 100, "reorder": 75,
        "cost": 12.50, "price": 19.99,
        "days": 1, "start": datetime(2024, 1, 1),
    },
)
conn.commit()

使用 executemany() 进行批量插入

executemany() 按列绑定参数并高效发送参数。用它处理中等批量，而不是循环调用 execute() 。请注意，executemany()需要使用元组列表的位置?标记，而execute()同时支持?和使用字典的命名%(name)s参数。有关每种样式的详细信息，请参见参数化查询。

rows = [
    ("Widget A", "WG-1001", 100, 75, 12.50, 19.99, 1, datetime(2024, 1, 1)),
    ("Widget B", "WG-1002", 100, 75, 15.00, 24.99, 1, datetime(2024, 1, 1)),
    ("Widget C", "WG-1003", 100, 75, 18.00, 29.99, 1, datetime(2024, 1, 1)),
]

cursor.executemany(
    """
    INSERT INTO Production.Product
        (Name, ProductNumber, SafetyStockLevel, ReorderPoint,
         StandardCost, ListPrice, DaysToManufacture, SellStartDate)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """,
    rows,
)
conn.commit()

适用于大负载的批量复制

当吞吐量比每行控制更重要时，切换到 bulkcopy()。它通过 TDS 批量插入协议以流式方式传输数据行，避免了参数化语句逐行处理的开销。 bulkcopy() 优于 executemany() 的确切临界点取决于行宽和网络延迟，但通常在数百行出头。对于非常小的批次，executemany() 更简单，因为 bulkcopy() 会创建单独的内部连接并自动提交。

与 execute() 和 executemany() 不同，bulkcopy() 是按位置将值映射到列中，而不是通过 INSERT 列列表。通过传递 column_mappings 来指定要加载到的目标列，这样源元组数据就会与正确的列对齐，而不是与表中靠前的标识列对齐：

result = cursor.bulkcopy(
    "Production.Product",
    rows,
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
)
print(f"Copied {result['rows_copied']} rows")

对于非常大的加载，使用生成器避免将整个数据集加载到内存中，并设置为 batch_size 定期提交：

import csv

def csv_rows(path):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield tuple(row)

cursor.bulkcopy(
    "Production.Product",
    csv_rows("products.csv"),
    column_mappings=["Name", "ProductNumber", "SafetyStockLevel", "ReorderPoint",
                     "StandardCost", "ListPrice", "DaysToManufacture", "SellStartDate"],
    batch_size=5000,
)

缓存策略

对于很少变化的参考数据（类别、查找表、配置），可以在应用中缓存结果，而不是每次请求都查询。

Python functools.lru_cache 提供简单的记忆化功能，但它会无限缓存，直到进程重启。如果底层数据可能发生变化，请使用 cachetools.TTLCache 在设定的时间间隔后自动刷新：

from cachetools import TTLCache, cached

category_cache = TTLCache(maxsize=1, ttl=300)  # Refresh every 5 minutes

@cached(category_cache)
def get_categories(connection_string: str) -> tuple:
    conn = mssql_python.connect(connection_string)
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT ProductCategoryID, Name FROM Production.ProductCategory")
        return cursor.fetchall()
    finally:
        conn.close()

网络优化

减少往返次数

每次查询都是一次网络往返服务器的过程。将相关查询合并到单个批次中，并使用 nextset() 遍历各个结果集：

# Bad: Multiple round trips
cursor.execute("SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
customer = cursor.fetchone()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": 1})
orders = cursor.fetchall()
cursor.execute("SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s)", {"customer_id": 1})
detail_count = cursor.fetchval()

# Good: Single round trip
cursor.execute("""
    SELECT CustomerID, AccountNumber FROM Sales.Customer WHERE CustomerID = %(customer_id)s;
    SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s;
    SELECT COUNT(*) FROM Sales.SalesOrderDetail WHERE SalesOrderID IN (SELECT SalesOrderID FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s);
""", {"customer_id": 1})

customer = cursor.fetchone()
cursor.nextset()
orders = cursor.fetchall()
cursor.nextset()
detail_count = cursor.fetchval()

使用服务器端处理复杂逻辑

把聚合和过滤推送到SQL Server，而不是直接抓取原始行并用Python处理。服务器返回的是单个摘要行，而非可能的数千个详细信息行：

cursor.execute("""
    SELECT p.Name, COUNT(sod.SalesOrderDetailID) AS OrderCount, SUM(sod.LineTotal) AS TotalSales
    FROM Production.Product p
    JOIN Sales.SalesOrderDetail sod ON p.ProductID = sod.ProductID
    WHERE p.ProductID = %(product_id)s
    GROUP BY p.Name
""", {"product_id": 707})

避免交错光标操作

mssql-python驱动不支持多重主动结果集（MARS）。每个连接一次只能有一个游标执行活动查询。在运行下一次查询前，完整获取第一个结果集，或者使用第二个连接：

connection_string = (
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

# Option 1: Fetch first, then query (single connection)
# Warning: This is an N+1 pattern. Each iteration is a round trip.
# Use this only when the JOIN in Option 2 is not possible.
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("SELECT ProductID FROM Production.Product WHERE ProductSubcategoryID = 1")
product_ids = [row[0] for row in cursor.fetchall()]

for pid in product_ids:
    cursor.execute("SELECT ProductID, LocationID, Quantity FROM Production.ProductInventory WHERE ProductID = %(product_id)s", {"product_id": pid})
    inventory = cursor.fetchone()

conn.close()

# Option 2: Use a JOIN instead of N+1 queries (preferred)
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
    SELECT p.ProductID, p.Name, i.Quantity
    FROM Production.Product p
    LEFT JOIN Production.ProductInventory i ON p.ProductID = i.ProductID
    WHERE p.ProductSubcategoryID = 1
""")
results = cursor.fetchall()
conn.close()

内存管理

分块处理大型结果集

将包含数百万行的表加载到列表中，所消耗的内存量与整个结果集的大小成正比。使用 OFFSET 和 FETCH NEXT 对数据进行服务器端分页，并一次处理一个数据块。

def quote_id(identifier: str) -> str:
    """Quote a possibly schema-qualified SQL identifier to prevent SQL injection."""
    return ".".join("[" + part.replace("]", "]]") + "]" for part in identifier.split("."))

def process_large_table(cursor, table: str, columns: list[str], key_column: str, processor, chunk_size: int = 10000):
    """Process large table without loading all data."""
    safe_table = quote_id(table)
    safe_key = quote_id(key_column)
    col_list = ", ".join(quote_id(c) for c in columns)
    cursor.execute(f"SELECT COUNT(*) FROM {safe_table}")
    total = cursor.fetchval()

    offset = 0
    while offset < total:
        cursor.execute(f"""
            SELECT {col_list} FROM {safe_table}
            ORDER BY {safe_key}
            OFFSET ? ROWS
            FETCH NEXT ? ROWS ONLY
        """, (offset, chunk_size))

        chunk = cursor.fetchall()
        processor(chunk)

        offset += chunk_size
        print(f"Processed {min(offset, total)}/{total}")

# key_column must be unique, otherwise rows can be duplicated or skipped across pages
process_large_table(
    cursor,
    "Production.TransactionHistory",
    ["TransactionID", "ProductID", "Quantity", "ActualCost"],
    "TransactionID",
    lambda chunk: None,  # replace with your row-processing logic
)

使用生成器进行直播

对 fetchmany() 进行封装的 Python 生成器无论表的大小如何，都能使内存占用保持恒定。调用者逐行迭代，但不加载完整结果集。对于超大的源数据，使用 UNION ALL 合并表，并以相同方式流式传输合并后的结果。

def stream_query(cursor, query: str, params: dict = None, batch_size: int = 1000):
    cursor.execute(query, params or {})
    
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        for row in batch:
            yield row

# Union the live and archive transaction tables into one extra-large result set
query = """
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistory
    UNION ALL
    SELECT TransactionID, ProductID, Quantity, ActualCost FROM Production.TransactionHistoryArchive
"""

count = 0
for row in stream_query(cursor, query, batch_size=5000):
    count += 1
print(f"Streamed {count} rows")

及时清理资源

未关闭连接会占用服务器资源，并可能耗尽连接池。使用上下文管理器保证即使出现异常也能清理。

from contextlib import contextmanager

@contextmanager
def database_connection(connection_string: str):
    conn = mssql_python.connect(connection_string)
    try:
        yield conn
    finally:
        conn.close()

with database_connection(connection_string) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
    data = cursor.fetchall()

监视内存使用量

大型结果集、长寿命缓存和连接对象都会消耗内存。如果你的应用程序作为服务运行，未关闭的光标或无界缓存导致的内存泄漏最终可能导致操作系统或容器运行时终止进程。

使用 Python 的tracemalloc模块来快照内存并找到最大的内存分配。

import tracemalloc

tracemalloc.start()

# ... run your workload ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:10]:
    print(stat)

常见的意外记忆增长来源包括：

对返回数百万行结果的查询调用 fetchall()。请改用 fetchmany() 或生成器。
缓存查询结果时未设置 maxsize 或 TTL。缓存会不断增长，直到进程被重新启动。
在循环中创建光标却未将其关闭。每个打开的光标都会将其结果集保存在内存中。

索引和查询计划优化

检查服务器端查询性能

使用 SET STATISTICS TIME ON 和 SET STATISTICS IO ON 来查看查询在服务器上花费多长时间以及读取了多少数据。逻辑读取次数高通常表明缺少索引。在 SQL Server Management Studio 或 Visual Studio Code 的 MSSQL 扩展中运行这些语句，输出会显示在消息面板中：

SET STATISTICS TIME ON;
SET STATISTICS IO ON;

SELECT * FROM Production.Product WHERE ProductSubcategoryID = 1;

SET STATISTICS TIME OFF;
SET STATISTICS IO OFF;

应会看到如下所示的输出：

Table 'Product'. Scan count 1, logical reads 3
SQL Server Execution Times: CPU time = 0 ms, elapsed time = 1 ms.

如果你看到高逻辑读取或表扫描，可以考虑添加索引。

使用查询提示作为战术性解决方案

查询提示覆盖查询优化器的索引和连接策略选择。在生产环境中，当查询性能突然回退时，它们可作为一种快速、低风险的应急修补方案，十分有价值。你可以立即在应用代码中部署提示，以稳定查询，同时调查根本原因（缺失的索引、过时的统计或模式变更）。

避免让暗示永久存在。当数据分布或模式发生变化时，硬编码的提示可能会让情况更糟。将这些问题视为暂时的，等根本问题解决后再回来：

cursor.execute("""
    SELECT * FROM Production.Product WITH (INDEX(AK_Product_Name))
    WHERE ProductSubcategoryID = %(subcategory_id)s
""", {"subcategory_id": 1})

使用OPTION（重新编译）绕过坏缓存计划

SQL Server 根据它看到的第一个参数值集合来缓存查询计划。如果数据分布在不同调用间差异很大，缓存计划在某些值下表现可能较差。这种问题称为参数嗅探，通常表现为“曾经很快”的查询突然需要几秒或几分钟。

OPTION (RECOMPILE)它会强制 SQL Server 为每次执行构建新的计划，这是一种有效的即时修复方案，你可以部署而无需服务器端的任何更改。代价在于每次调用都会带来少量编译开销，但对于不常运行或返回大小可变结果集的查询来说，与执行一个糟糕的执行计划相比，这点开销几乎可以忽略不计。

一旦问题稳定下来，你可以慢慢进行永久修复，比如重写查询、添加过滤索引或使用计划指南：

cursor.execute("""
    SELECT * FROM Sales.SalesOrderHeader
    WHERE OrderDate > %(start_date)s
    OPTION (RECOMPILE)
""", {"start_date": start_date})

性能监控

请安排好你的提问时间

要查找耗时操作，请使用 time.perf_counter() 包装查询：

import time

start = time.perf_counter()
cursor.execute("SELECT SalesOrderID, OrderDate, TotalDue FROM Sales.SalesOrderHeader WHERE CustomerID = %(customer_id)s", {"customer_id": customer_id})
rows = cursor.fetchall()
elapsed = time.perf_counter() - start

print(f"Query returned {len(rows)} rows in {elapsed:.3f}s")

如果你想更全面地了解你的应用所花费的时间，可以使用 Python 内置cProfile的模块：

python -m cProfile -s cumtime my_app.py

此视图显示每个函数调用的累计时间，有助于你判断慢是出在查询执行、数据处理还是网络延迟。

使用查询存储进行服务器端分析

客户端时序告诉你从应用角度看查询所需时间，但它结合了网络延迟、服务器执行时间和客户端处理。查询存储会在服务器上捕捉执行计划和运行时统计数据，因此你可以准确看到 SQL Server 如何执行每个查询、运行频率以及性能随时间的变化。

查询存储特别适合识别参数嗅探、计划回归以及消耗最多服务器资源的查询。你可以直接查询 sys.query_store_runtime_stats 和 sys.query_store_plan view，或者使用 SQL Server Management Studio 内置的查询存储报表。

使用性能仪表板报告

SQL Server Management Studio 中的性能仪表盘报告提供 SQL Server 健康状况的实时概览，包括当前等待类型、活跃且昂贵的查询以及 CPU/IO 趋势。用它们快速发现瓶颈，无需直接向DMV提交查询。

性能清单

连接

[ ] 启用连接池。
[ ] 根据你的工作量来计算池数。
[ ] 在操作中重复使用连接。
[ ] 在长期运行的服务中保持连接处于打开状态。

Queries

[ ] 只选择你需要的列。
[ ] 为每个查询使用合适的获取方法。
[ ] 实现服务器端分页。
[ ] 连接后设置 SET NOCOUNT ON 一次。
[ ] 通过批量处理查询来减少往返次数。

插入

[ ] 单行插入请使用 execute()。
[ ] 使用 executemany() 处理小到中等规模的批次（约 10–1,000 行）。
[ ] 当吞吐量比每行控制更重要时使用 bulkcopy() 。

Caching

[ ] 用TTL缓存引用数据以避免提供过时的结果。

Resources

[ ] 处理大块结果，或使用生成器。
[ ] 及时清理连接。
[ ] 在长期运行的服务中使用 tracemalloc 监控内存使用情况。

反馈

此页面是否有帮助？

Last updated on 2026-07-22