使用具有Azure Database for PostgreSQL灵活服务器的缓存来提高性能

在Azure Database for PostgreSQL灵活服务器上生成应用程序时，添加缓存层是改进响应时间、减少数据库负载并提高复原能力的最有效方法之一。 通过从内存缓存中提供经常读取的数据，应用程序会向 PostgreSQL 发送更少的查询请求。 这意味着 CPU 和 IOPS 消耗量较低，因此可以在较小的计算层上运行、缩放读取而不纵向扩展服务器以及吸收流量峰值。 缓存还可以添加复原能力。 如果 PostgreSQL 短暂中断，则缓存中已命中数据的请求可以保持成功，因此读取路径在数据库恢复时保持可用。

本文可帮助你确定缓存何时帮助以及哪种模式适合应用程序。 然后，它借助 Azure Managed Redis、redis 和 psycopg 库以及 Microsoft Entra ID 身份验证，在 Python 中实现了四种缓存模式（旁路缓存、参考数据预提取、直写和事件驱动）。

何时添加缓存

PostgreSQL 已经缓存其缓冲区缓存中经常访问的数据页，并且还受益于操作系统的文件缓存。 这些缓存可以加快重复访问的速度，但它们与查询执行和其他进程共享分配给数据库服务器的内存。 在某些维护和故障转移操作之后，其中的内容也需要再次预热。

可以通过缩放到具有更多内存的计算选项来获取更多数据库缓存容量。 你还可以调整 PostgreSQL 的内存设置，但为缓冲区缓存分配更多内存，就会减少留给查询执行和操作系统的内存。 对内存更改进行仔细测试，以避免发生内存不足问题。

Azure Managed Redis 补充了这些原生缓存。 它将选定的应用程序数据和查询结果存储在数据库服务器之外，为时间敏感读取路径提供较低的延迟访问，并且可以在 PostgreSQL 恢复或预热缓存时保持缓存读取可用。 当这些收益足以证明增加应用逻辑和运营另一个服务是值得的时候，再使用它。 它不会取代 PostgreSQL 作为事实来源。

当您的工作负载具有以下特征时，添加 Azure Managed Redis 等外部缓存最有帮助：

• 读取密集型的访问模式。 相同的行的读取频率比更改要高得多，例如产品目录、用户配置文件、配置数据或引用表。
• 成本高昂或重复的查询。 聚合、联接或计算结果，这些结果成本高昂，但在短时间内保持稳定。
• 延迟敏感型端点。 面向用户的操作，其中内存中读取（子毫秒）优于数据库往返。
• 可预测的峰值。 季节性流量或事件驱动型流量中，缓存可以吸收原本会迫使你扩展计算资源的负载。
• 维护和故障转移敏感性。 在 PostgreSQL 实例于维护或故障转移操作后恢复或进行缓存预热期间，读取路径需要稳定的响应时间。

对于写入密集型工作负载、必须始终保持事务一致性的数据，或者本身就很快且很少重复的查询，缓存的帮助较小。

缓存模式

本文以零售店铺为贯穿全文的示例。 应用的不同部分受益于不同的缓存模式。 以下各节在 Python 中实现前四种模式。 本文介绍会话和状态卸载和多区域缓存模式，但不为它们提供代码实现。

Prerequisites

• 拥有有效订阅的 Azure 帐户。 创建试用帐户。
• 启用了Microsoft Entra身份验证的Azure Database for PostgreSQL灵活服务器实例。 若要创建一个服务器，请参阅创建Azure Database for PostgreSQL灵活服务器。
• 一个Azure Managed Redis 实例。 若要创建一个，请参阅创建Azure托管 Redis 实例。 在 PostgreSQL 服务器所在的同一区域中创建它（对于生产环境，同一虚拟网络），以最大程度地减少延迟。
• 在这两项服务上为你的开发人员帐户或应用程序标识配置数据访问权限：PostgreSQL 服务器上的 Microsoft Entra 管理员或角色，以及缓存上的 Redis 访问策略分配。 请参阅适用于 Azure Database for PostgreSQL 的 Microsoft Entra 身份验证和将 Microsoft Entra ID 用于 Azure Managed Redis 身份验证。
• Python 3.10 或更高版本。
• Azure CLI。 若要安装它，请参阅如何安装Azure CLI。

步骤 1：安装客户端库

安装 Redis 和 PostgreSQL 客户端库，以及用于Microsoft Entra身份验证的Azure标识库。

pip install "redis>=5.0,<6.0" "psycopg[binary]>=3.1,<4.0" "azure-identity>=1.17,<2.0"

步骤 2：使用 Microsoft Entra ID 进行连接

使用Microsoft Entra ID身份验证，而不是访问密钥。 Microsoft Entra ID无需在应用程序中存储机密，并允许集中管理访问权限。

以下代码创建 Redis 客户端和 PostgreSQL 连接，通过托管标识或开发人员凭据 DefaultAzureCredential进行身份验证。 此示例使用与此示例使用的 OSS 群集策略匹配的群集感知 RedisCluster 客户端。 如果缓存使用企业群集策略，请改用标准 redis.Redis 客户端。

import os
import json
import redis
from redis.cluster import RedisCluster
import psycopg
from azure.identity import DefaultAzureCredential

REDIS_HOST = os.environ["REDIS_HOST"]  # for example, mycache.chinaeast.redis.chinacloudapi.cn
REDIS_PORT = 10000
PG_HOST = os.environ["PG_HOST"]        # for example, myserver.postgres.database.chinacloudapi.cn
PG_DATABASE = os.environ["PG_DATABASE"]

credential = DefaultAzureCredential()

# Acquire a token for Azure Managed Redis and use it as the password.
redis_token = credential.get_token("https://redis.azure.com/.default")

# The username is the object ID of the Microsoft Entra identity.
redis_client = RedisCluster(
    host=REDIS_HOST,
    port=REDIS_PORT,
    ssl=True,
    ssl_check_hostname=False,  # cluster nodes are reached by IP; the certificate chain is still validated
    username=os.environ["REDIS_USER_OBJECT_ID"],
    password=redis_token.token,
    decode_responses=True,
)

# Acquire a token for Azure Database for PostgreSQL and use it as the password.
pg_token = credential.get_token("https://ossrdbms-aad.database.windows.net/.default")

pg_conn = psycopg.connect(
    host=PG_HOST,
    dbname=PG_DATABASE,
    user=os.environ["PG_USER"],
    password=pg_token.token,
    sslmode="require",
)

步骤 3：缓存端

旁路缓存是最常见的模式，在此示例中用于产品数据读取。 应用程序首先检查 Redis，未命中时会查询 PostgreSQL，并将结果写入缓存，并设置生存时间（TTL）。 少数热门产品带来了大部分阅读量，因此命中率很高。

由于大多数读取请求都直接由内存提供，因此旁路缓存会将持续的读取负载从 PostgreSQL 上移开。 这种减少意味着连接更少、缓冲区缓存变动率减少、CPU 和 IOPS 降低。 可以在不扩容服务器或添加读副本的情况下吸收读取流量峰值。 只有在未命中时（首次访问，或 TTL 过期后）才会查询 PostgreSQL。 请参阅 “查找要缓存的内容 ”以标识值得缓存的查询。

CACHE_TTL_SECONDS = 300  # 5 minutes

def get_product(product_id: int) -> dict | None:
    cache_key = f"product:{product_id}"

    # 1. Try the cache first.
    cached = redis_client.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    # 2. On a miss, read from PostgreSQL.
    with pg_conn.cursor() as cursor:
        cursor.execute("SELECT id, name, price FROM products WHERE id = %s", (product_id,))
        row = cursor.fetchone()
    if row is None:
        return None

    product = {"id": row[0], "name": row[1], "price": float(row[2])}

    # 3. Populate the cache with a TTL, then return.
    redis_client.set(cache_key, json.dumps(product), ex=CACHE_TTL_SECONDS)
    return product

步骤 4：参考数据预取

不断读取但很少更改（例如类别、品牌或发货配置）的稳定数据不需要等待缓存丢失。 预先将其加载到缓存中，并在源发生变化时刷新它。 在 PostgreSQL 架构中，这些表通常是联接到多个查询中的小型查找和维度表。 从内存中提供它们会从数据库中删除大量的重复联接和查找。 与缓存端不同，没有每个请求未命中，也没有 TTL 争用。 在发生更改时刷新，因此读取始终保持热状态。

def prefetch_categories() -> None:
    with pg_conn.cursor() as cursor:
        cursor.execute("SELECT id, name FROM categories ORDER BY name")
        categories = [{"id": r[0], "name": r[1]} for r in cursor.fetchall()]
    redis_client.set("ref:categories", json.dumps(categories))  # no TTL; refreshed on change

def get_categories() -> list[dict]:
    cached = redis_client.get("ref:categories")
    return json.loads(cached) if cached else []

步骤 5：直写

当必须立即看到更改时，请在同一操作中写入 PostgreSQL 和缓存，而不是等待 TTL 过期或使密钥失效。 例如，此模式用于更新示例中的定价信息。 PostgreSQL 始终是事实来源。 更改先提交，然后刷新缓存，因此写入后读取返回新值。

向两个不共享同一事务的系统写入数据时，数据库提交可能成功，而缓存刷新却失败，而且这个问题没有简单的解决办法。 以下代码片段展示了正常流程，省略了失败处理。在生产环境中，你需要决定如何处理刷新失败的情况，例如当失败看起来只是暂时性问题时进行重试，或者使该键失效，以便下一次读取时从 PostgreSQL 重新加载。 无论哪种方式，PostgreSQL 都保留正确的值，因此始终可恢复过时或缺失的缓存条目。 当缓存更新必须可靠生效时，请改为由数据库的变更流来驱动（请参阅事件驱动失效）。

def update_price(product_id: int, new_price: float) -> None:
    # 1. Write to PostgreSQL, the source of truth.
    with pg_conn.cursor() as cursor:
        cursor.execute("UPDATE products SET price = %s WHERE id = %s", (new_price, product_id))
    pg_conn.commit()

    # 2. Refresh the cached entry so reads see the new price right away.
    with pg_conn.cursor() as cursor:
        cursor.execute("SELECT id, name, price FROM products WHERE id = %s", (product_id,))
        row = cursor.fetchone()
    if row is not None:
        product = {"id": row[0], "name": row[1], "price": float(row[2])}
        redis_client.set(f"product:{product_id}", json.dumps(product), ex=CACHE_TTL_SECONDS)

步骤 6：事件驱动的失效

事件驱动的失效机制通过响应数据变更事件，使缓存与数据库保持一致。 它会在条目发生变化时更新或将其设为无效，而不是通过定时器使其过期。 写入方将事件追加到持久化的 Redis 流（一种仅追加日志）中，而一个或多个消费者会读取这些事件并更新缓存。 由于流仍然存在，因此事件在使用者重启后幸存下来。 消费者组会将每个事件分发给单个工作进程，跟踪确认情况以确保不会发生消息丢失或重复处理，并支持在多个工作进程之间扩展处理能力。

当缓存值派生自其他地方会发生变化的数据（例如状态、投影或聚合）时，应使用此模式，因为在这种情况下，TTL 要么会返回陈旧数据，要么会迫使系统频繁地重新计算。 在店面中，此模式通过履行驱动订单状态。 下订单将订单写入 PostgreSQL，缓存其初始状态，并将事件追加 placed 到流中：

ORDER_STREAM = "orders:events"

def place_order(product_id: int, quantity: int) -> int:
    with pg_conn.cursor() as cursor:
        cursor.execute(
            "INSERT INTO orders (product_id, quantity, status) VALUES (%s, %s, 'placed') RETURNING id",
            (product_id, quantity),
        )
        order_id = cursor.fetchone()[0]
    pg_conn.commit()

    redis_client.set(f"order:{order_id}:status", "placed", ex=86400)
    redis_client.xadd(ORDER_STREAM, {"order_id": order_id, "status": "placed"}, maxlen=10000, approximate=True)
    return order_id

履行工作线程运行使用者组：它读取新事件，在 PostgreSQL 中推进每个订单，刷新缓存 order:{id}:status 的投影，并确认事件。 订单页会读取该投影，因此状态检查会保持快速，并且永远不会接触数据库。 值保持不变，因为事件保持最新状态。

GROUP = "fulfillment"

def process_orders() -> None:
    try:
        redis_client.xgroup_create(ORDER_STREAM, GROUP, id="0", mkstream=True)
    except redis.exceptions.ResponseError:
        pass  # group already exists

    while True:
        events = redis_client.xreadgroup(GROUP, "worker-1", {ORDER_STREAM: ">"}, count=10, block=5000)
        for _stream, entries in events or []:
            for event_id, fields in entries:
                order_id = int(fields["order_id"])
                with pg_conn.cursor() as cursor:
                    cursor.execute("UPDATE orders SET status = 'shipped' WHERE id = %s", (order_id,))
                pg_conn.commit()
                redis_client.set(f"order:{order_id}:status", "shipped", ex=86400)
                redis_client.xack(ORDER_STREAM, GROUP, event_id)

此示例中的事件源是应用程序，该应用程序将 PostgreSQL 写入并在同一路径中追加该事件。 PostgreSQL 还可以自行发出变更：LISTEN/NOTIFY 用于轻量级通知，或通过逻辑解码（变更数据捕获）生成持久的、行级变更流。 由 PostgreSQL 自身的变更流来驱动缓存，意味着它会对每一次已提交的变更作出响应，甚至包括绕过应用程序的写入。

最佳做法

遵循这些做法，使缓存保持正确、高效且经济高效。

• 在任何存在数据陈旧风险的地方，都应设置 TTL。 如果失效操作失败，TTL 可限制数据的陈旧时间。 将 TTL 与应用程序可以接受的最大过期时间匹配。 仅当存在可靠的无效和刷新进程时，才对引用数据使用长 TTL 或无 TTL。
• 使用一致的密钥命名方案。 按服务、实体和标识符（如 product:42 或 user:1001:profile）的命名空间密钥。 当密钥格式或值架构可能发生变化时，请添加版本号。
• 缓存正确的粒度。 对于经常复用且易于使缓存失效的单个实体或小型结果集，应将其缓存。 不要缓存很少重用的数据。 过度缓存会浪费内存并降低命中率。
• 妥善处理缓存未命中和故障情况。 将缓存视为优化，而不是事实来源。 如果 Redis 不可用，请使用有界回退机制，切换到 PostgreSQL。 添加超时、断路器、退避和请求限制来保护 PostgreSQL。 如果 PostgreSQL 短暂中断服务，您仍可继续提供对缓存中已有数据的读取，同时让写入请求等待数据库恢复。
• 防止缓存雪崩。 当常用密钥过期时，许多请求可以同时命中数据库。 使用 TTL 抖动、请求合并、过时的重新验证或短分布式锁让一个请求重新填充条目。
• 正确调整缓存大小。 监视命中率、内存使用情况、逐出率、过期率、延迟、热键和密钥基数。 低命中率可以指示缓存太小、逐出过多、键选择差或访问模式不佳。 有关规模估算指导，请参阅 Azure 托管 Redis 层选择指南。
• 选择符合密钥的逐出策略。 对于仅缓存数据库，请从 allkeys-lru 或 allkeys-lfu 开始。 仅当同一数据库包含即将过期的缓存密钥和受保护的非过期密钥时，才使用 volatile-* 策略。 当没有密钥具有 TTL 时，可变策略可以停止逐出。 尽可能将缓存数据与受保护的数据分开。
• 高效序列化。 JSON 是可读且可移植的。 对于高吞吐量路径，请测试压缩的二进制格式以减少内存和网络开销。 更改格式之前，基准内存、CPU、延迟、架构演变和调试影响。

查找要缓存的内容

最有效的缓存目标是应用程序最常针对更改最少的数据运行的查询。 在启用相关功能时，在查询遥测中比较Azure Database for PostgreSQL可以收集的历史和当前视图，而不是猜测哪些查询适合此说明：

• 查询存储保留用于历史分析的查询执行统计信息。 使用调用计数和总计和平均执行时间查找在较长时间内持续主宰数据库负载的查询。 请参阅使用 查询存储 监视性能。
• Query Performance Insight 在 Azure 门户中可视化查询存储数据，以便可以发现频繁和资源密集型查询，并随时间推移比较其行为。 请参阅 查询性能见解。
• pg_stat_statements 公开数据库中的累积每语句统计信息，以获取当前观察窗口的直接视图。 可以重置其统计信息，因此，如果需要跨观察窗口保留历史记录，请使用查询存储。

在选择缓存候选项之前，请使用这两个视图。 短期峰值可能不会表示工作负荷的正常行为，而历史平均值可能会隐藏当前回归。 优先考虑 频繁、开销大且稳定的查询，也就是说，这类查询调用次数多、总执行时间长，而且结果不会因每次请求而发生变化。 这些查询提供最高的缓存命中率和数据库负载的最大下降。 持续运行但在数分钟内始终返回相同结果的查询（例如产品列表、类别树或定价表），是理想的适用对象。

相关内容

• 什么是 Azure 托管 Redis？
• 缓存指南
• 规划 Azure Database for PostgreSQL 灵活服务器部署以提升运行性能