数据注入吞吐量最佳实践

本页面介绍将数据注入 Cloud Healthcare API 时优化数据吞吐量的最佳实践。这些建议面向具有管理大型系统数据吞吐量的经验的从业者。

数据吞吐量

数据吞吐量是 FHIR 资源或 DICOM 实例等资源的数量，或 Cloud Healthcare API 每秒注入的字节数。

数据吞吐量限制

以下列表说明了数据吞吐量可能受限的原因：

您并未计划处理会导致流量高峰的大量请求。
带宽限制会限制在短时间内发送的大数据量。
多个并发事务会更改同一 Cloud Healthcare API 资源，这会导致数据争用。
发出的较小请求过多。如需了解详情，请参阅避免较小的导入和导出请求。
同时运行的长时间运行的操作 (LRO) 过多，且带宽有限。
同时调度多个 LRO 会导致失败。

重试失败的请求

如果客户端在失败后快速反复重试请求，则可能会超出 Cloud Healthcare API 配额。以下部分介绍了如何高效地重试失败的请求。

将指数退避算法与抖动和永久性重试队列搭配使用

引入了抖动的指数退避算法是适用于网络应用的标准错误处理策略。客户端定期重试，在重试次数之间呈指数级增加，且延迟时间短时随机出现小延迟。

请确保每次重试的指数退避算法实现都遵循幂等原则，尤其是在您使用自定义逻辑来绕过失败条件时。如需了解详情，请参阅 HTTP 规范中的 9.2.2 幂等方法。

大多数编程语言都提供库，以简化指数退避算法和类似的重试策略的实现。对于长期或多进程重试，请实现持久重试队列。如果超出退避时间上限，此队列可以重置重试机制。

重试以下请求时使用指数退避算法：

用于修改 FHIR 资源或 FHIR 资源包的操作。
同步 LRO 请求。如果 LRO 启动时发生错误或 LRO 失败，请重试。

LRO 具有独特的错误，可能需要您实施以下重试策略：
- 请使用单独的软件包来存储未能导入或创建操作的数据。
- 对同步失败的数据使用同步请求。

指数退避算法示例

指数退避算法以指数方式重试请求（不断增加各次重试之间的等待时间，直到达到最大退避时间）。以下算法使用抖动来实现截断指数退避算法：

向 Cloud Healthcare API 发送请求。
如果请求失败，请等待 1 + random-fraction 秒，然后重试请求。
如果请求失败，请等待 2 + random-fraction 秒，然后重试请求。
如果请求失败，请等待 4 + random-fraction 秒，然后重试请求。
继续此模式，每次重试后等待 2ⁿ + random-fraction 秒，最多 maximum-backoff 次。
deadline 秒后，停止重试该请求。

在实现算法时，请使用以下值：

每次重试前，等待时间均为 min((2ⁿ + random-fraction), maximum-backoff)，n 从 0 开始，且每次重试都会递增 1。
将 random-fraction 替换为小于或等于 1 的随机小数值。每次重试使用不同的值。添加此随机值可防止客户端同时同步并发送多次重试。
将 maximum-backoff 替换为重试之间的最长等待时间（以秒为单位）。典型值为 32 或 64（2⁵ 或 2⁶）秒。请选择最适合您的使用场景的值。
将 deadline 替换为持续发送重试的秒数上限。请选择一个可反映您的使用场景的值。

客户端可以在达到 maximum-backoff 时间后使用与退避时间相同的值重试。例如，如果 maximum-backoff 时间为 64 秒，则每 64 秒重试一次。确保客户端不会无限期重试。

实施“根据流量情况调整投放速度”这一客户端速率限制

速率限制可防止大规模系统因过多的请求而超负荷运转。如果客户端速率限制还不够，则 Cloud Healthcare API 配额系统可能会限制数据吞吐量。如需了解详情，请参阅配额管理最佳实践。

如果您有其他要求（例如保证重试次数之间传递），则重试失败的请求中的策略可能不足。根据流量情况调整投放速度是一种限制技术，可确保客户端请求的速率在带宽限制范围内。这会将负载峰值分散到小时或分钟，从而提高吞吐量。配额受限时，“调整流量”设置可实现比单独使用重试次数更高的吞吐量，因为它可以避免推送并跟踪工作器单元。

您可以针对同步创建、删除、更新和删除 (CRUD) 操作（包括 fhir.executeBundle）实现流量塑形。

“根据流量情况调整投放速度”功能的要求

要实现“根据流量情况调整投放速度”功能，您的系统必须实现以下各项：

具有冗余的存储支持的处理队列，以避免磁盘故障。
从处理队列拉取的协同工作器。
使用总体检测来根据配额限制调整工作器数量及其处理速度。
基于存储支持的处理队列的灾难恢复。如果发生灾难，您的系统必须能够完全清除或恢复队列。
降低了高峰时段的 LRO。如需了解详情，请参阅高效规划和使用配额以及将 LRO 加入队列和管理。

在以下情况下，可能只需要在单个流水线阶段调整流量：

限制从上一个流水线步骤拉取的工作器数量。
分别限制每个 worker。
使用工作器池协调器调整处理各个工作单元（如每秒查询次数 (QPS) 或注入的字节数）的速率。

在系统的其他方面实现速率限制

您可以使用现有的编程语言和框架来实现流量调整。考虑以下开源项目和预构建解决方案：

Apache Beam 中的客户端限制。如需了解如何使用 numWorkers 和 maxNumWorkers 标志来控制限制，请参阅横向自动扩缩。
Google Guava 核心 Java 库集中的 Java RateLimiter 类。
Python ratelimiter 模块。

对于流控制，请使用高级 Pub/Sub 客户端库。

选择异步处理或同步处理

封装了 Cloud Healthcare API 的客户端代理层（如处理多个层的错误中所示）还可以控制在使用 Cloud Healthcare API 的服务中的节流。根据所需的流量调整类型，使用以下选项之一：

异步

使用异步处理将请求加入队列并控制 worker。代理层会将传入的请求写入队列，并在每个请求加入队列后返回 200 OK 响应。这最适合写入请求，但也可用作 LRO 框架中的读取请求（如果客户端可以接收读取结果）。

同步

如果工作单元依赖于前一个单元完成，同步处理将提供一种简单的反馈机制。代理层会根据 QPS 或字节吞吐量限制延迟出站请求，而客户端会阻塞并等待代理层的响应。

代理层可以根据实例数调整速率限制，也可以与每隔几秒调整速率限制的控制器进程协调。为了让代理层跟踪实例数量及其速率限制，每个代理实例都可以定期读取文件或进行已编码的远程过程调用 (RPC)。

同步处理有时具有以下缺点：

在客户端屏蔽并等待响应时，客户端和代理层中的资源不可用。这可能会导致错误、超时和数据吞吐量降低，从而使其更难以扩缩。
如果客户端与代理层断开连接，则需要执行更多工作才能确保数据按照请求被修改。

使用 Cloud Tasks

使用 Cloud Tasks 将请求分流到队列。Cloud Tasks 会自动设置和监控以下 Google Cloud 配额：

使用 RateLimits 对象爆发大小上限和最大请求并发数
使用 RetryConfig 对象重试限制

请参阅创建队列，在 Cloud Tasks 中创建队列。Queue 资源显示您可以对队列设置的选项。例如，您可以使用 RetryConfig 对象来实现指数退避算法。请参阅 Cloud Tasks 客户端库，了解特定语言的库。

使用 Cloud Tasks 时，请考虑以下事项：

Cloud Tasks 不保证只传送一次。传送一次后，所有包含重复数据的请求都会被识别为重复项，并被服务器忽略。如需了解详情，请参阅“Lambda 之后：在 Dataflow 中仅处理一次”，第 1 部分。
任务大小上限可能远低于 Cloud Healthcare API 中的大小上限 FHIR 软件包大小。如需了解详情，请参阅 Cloud Tasks 配额和限制以及 Cloud Healthcare API 配额和限制。
Cloud Tasks 存在问题和限制。

将 FHIR 软件包与速率限制器结合使用

使用指数退避算法和速率限制器重试 FHIR 捆绑包有助于保持高数据吞吐量和管理负载峰值。

客户端可以将批量 FHIR 捆绑包和事务 FHIR 捆绑包发送到 Cloud Tasks，后者会将捆绑包中的请求发送到 Cloud Healthcare API。如果速率限制器已满或超出配额（因为它已达到队列大小上限）并耗尽磁盘空间，则客户端可以实现指数退避算法来将捆绑包排队。

监控以下资源，防止速率限制器队列已满：

Cloud Healthcare API 中的 FHIR 操作配额
速率限制器
速率限制器错误

如果速率限制器队列已满，您的系统必须提醒用户，并阻止客户端发送请求。

使用 HTTP 持久（可重复使用的 keep-alive）连接

默认情况下，Cloud Healthcare API 会为每个 CRUD 请求打开新的 TCP 连接。这需要 TCP 握手，这可能会导致开销并降低性能。为了提高性能，请使用 HTTP keep-alive 使 TCP 连接对多个请求保持打开状态。

如需在 HTTP/1.1 中使用 HTTP keep-alive，请将 Connection 标头设置为 keep-alive：

Connection: keep-alive

HTTP/2 使用一个 TCP 连接进行依序请求和并发请求，从而自动避免开销。

Python requests 库默认使用 HTTP keep-alive。如果您使用的是 Node.js，请在创建 http.Agent 对象时将 keepAlive 设置为 true，然后在请求中传递该对象。

使用测试框架

测试框架可确保您的代码能够正常运行并帮助您执行以下操作：

为应用或流水线中的突发流量高峰做好准备。
测试指数退避算法和客户端速率限制是否提高性能。测试可以显示这些实现是否会造成必须单独处理的任务积压。
分离和控制高优先级流量。例如，如果用户正在等待响应，则可以减少后台处理任务的工作负载，以确保用户体验不会变差。
测试用于控制同步流的同步和异步排队策略，或测试代理层是否处理回退。
制定灾难恢复计划。这通常需要重置传入流量，或在灾难结束后使用队列恢复流量。

使用 Cloud Monitoring

使用 Cloud Monitoring 监控测试和生产环境。请遵循以下建议：

将 Cloud Tasks 与其他 Google Cloud 日志记录和监控服务（例如 Cloud Audit Logs）集成。
使用 Cloud Monitoring API 创建自定义指标，以跟踪重试、队列大小和队列存在时间等关键指标。
为您的环境创建服务等级目标 (SLO) 和服务等级指标 (SLI)。如需查看建议，请参阅 SLI 简介。
使用 Google Cloud 的运维套件创建提醒政策。提醒政策会通知您的问题，例如您的系统是否处于压力状态或需要人工干预。
创建操作策略方案，以便系统管理员了解在提醒政策发送通知时该怎么办。
在临时环境中使用操作策略方案来响应以下场景：
- 因速率限制导致的积压
- 因超出配额限制而导致回滚
- 传入的流量高峰

防止 `429 Resource Exhausted operation_too_costly` 个错误

每天对 FHIR 资源进行数千个并行更新可能会导致锁争用和延迟，并阻止事务完成。无法完成的事务可能会导致 429 Resource Exhausted operation_too_costly 错误积压：

HTTP/1.1 429 Too many requests
...

{
  "issue": [
    {
      "code": "too-costly",
      "details": {
        "text": "operation_too_costly"
      },
      "diagnostics": "aborted due to lock contention while executing transactional bundle. Resource type: FHIR_RESOURCE_TYPE",
      "severity": "error"
    }
  ],
  "resourceType": "OperationOutcome"
}

在本示例中，“费用”是指资源用量和数据吞吐量，而不是结算费用。

429 Too Many Requests 错误并不一定表示存在配额问题。当 Cloud Healthcare API FHIR 服务器检测到数据库记录存在过多的锁争用时，可能会发生此错误。这可能是因为 FHIR 软件包中的许多操作或 CRUD 操作的组合。

请考虑以下场景：

用于更新患者资源和其他 FHIR 资源的 FHIR 事务软件包会锁定患者资源，直到事务完成。
多个 FHIR 软件包尝试并行更新患者资源，并发生锁争用。错误响应包括一个内容为 Resource type: PATIENT 的 diagnostics 字段。

您可以尝试使用指数退避算法更新患者资源，但长时间锁定争用可能会导致超时、减少吞吐量并增加资源使用量。
Cloud Healthcare API FHIR 服务器最终会通过返回 operation_too_costly 错误来检测积压的事务和负载变化。这样可以限制流量，并避免进一步出错。

operation_too_costly 错误会限制 Google Cloud 项目中的所有 FHIR CRUD 操作，这会影响连接到项目的所有应用。

排查 `429 Too Many Requests` 错误

如需排查 429 Too Many Requests 错误，请搜索 Cloud Logging。包含 operation_too_costly 的错误表示锁争用。如果错误是由资源耗尽引起的，请检查是否存在配额问题。

如果发生节流，则事务捆绑包可能会因锁定争用级别较高而失败，并产生以下错误：

HTTP/1.1 429 Too many requests
...

{
  "issue": [
    {
      "code": "too-costly",
      "details": {
        "text": "operation_too_costly"
      },
      "diagnostics": "aborted due to cumulative heavy load or lock contention in this project while executing transactional bundle, please see https://cloud.google.com/healthcare-api/docs/troubleshooting#fhir_transaction_bundle_heavy_load for more information",
      "severity": "error"
    }
  ],
  "resourceType": "OperationOutcome"
}

如需对此错误进行问题排查，请转到 diagnostics 字段中的 FHIR 事务包因累积大量负载而被取消。

避免创建较大的软件包

如果使用大型事务包，则更有可能出现 429 Too Many Requests 错误。任何规模的捆绑包都可能造成吞吐量瓶颈。测试不同的捆绑包以找到最合适的尺寸。

包含重试次数的大包内容可能会减少性能回报，并且更有可能发生多次故障。客户端应实现其他逻辑来管理在事务中失败的 FHIR 资源子集。

如果批量捆绑包较大或具有较高的 QPS，则可能会遇到 429 Too Many Requests 和 413 Request Entity Too Large 错误以及吞吐量瓶颈。

避免使用包含数千个事务的大型软件包。请改为执行以下操作：

使用支持数据一致性的较小事务包。如果 FHIR 资源互不干扰，请分别进行更新。例如，FHIR 资源可能并不依赖于同一软件包中另一个资源的特定版本。
使用捆绑包进行批量处理，并避免单个请求。批处理可以提高性能，但较大的批次可能会导致错误并降低数据吞吐量。大小类似的批量软件包争用更少，因为它们不会在 FHIR 资源更新中处于锁定状态。

小型事务捆绑包可避免争用，因为它们每次只持有几条锁，并且会快速完成。这有助于防止堆叠的事务积压。

LRO 吞吐量

请参阅 LRO 数据吞吐量。

FHIR 数据存储选项

如果您的 FHIR 数据量较小到中等，请使用 fhir.create 存储数据。如需存储大量 FHIR 资源，请使用 fhir.executeBundle 或 fhirStores.import。如需了解每种方法，请参阅 FHIR 导入选项。

导入 FHIR 资源

在决定是否使用 FHIR 导入功能时，请考虑以下事项：

FHIR 导入不会限制其导入的数据的总大小。如果 FHIR 软件包超过 50 MB，您可以将 FHIR 资源上传到 Cloud Storage 并将其导入。避免并发高延迟或大型导入，否则数据吞吐量可能会受到限制。
FHIR 导入的复杂度低于使用 FHIR 软件包。例如，您无需执行以下操作：
- 将较大的软件包拆分为较小的软件包
- 管理时间表
- 在资源级或软件包级重试暂时性错误
FHIR 导入不会强制执行参照完整性。如需了解详情，请参阅 FHIR 参照完整性。
当数据新鲜度为高优先级时，请勿使用 FHIR 导入。导入过程可能很快，但可能会延迟数小时或数天。
当 Google Cloud 项目中只有较少的 LRO 时，FHIR 导入的效果更好。
如果您的应用可以处理一部分资源的批量错误和故障，FHIR 导入可以实现较高的数据吞吐量。

使用 FHIR 软件包

在以下情况下使用 FHIR 软件包，而不是 FHIR 导入：

不管是在结算费用还是网络带宽方面，构建流水线将数据存储在 Cloud Storage 中并进行导入的成本过高。
必须强制执行参照完整性。
必须强制执行 FHIR 配置文件验证。
存储 FHIR 资源时，您需要发送 Pub/Sub 通知。FHIR 导入不支持 Pub/Sub 通知。
数据新鲜度是重中之重，数据注入必须为几秒钟或几分钟。但是，即使在设计完善的系统中，数据吞吐量也可能受到以下因素的限制：
- 处理流水线的上游延迟。流水线可能需要更多时间来准备数据，然后才能提取数据。
- 退避、重试和流量形成代理层。

FHIR 软件包具有以下限制：

配额和结算将应用于捆绑包中的每个操作，就好像每个操作都是单独执行的一样。例如，如果某个捆绑包中有 10 个 POST 操作、5 个 GET 操作和 1 个 DELETE 操作，则捆绑包的配额和结算方式与这些操作是单独执行的。
大型事务捆绑包更有可能出现导致锁争用的事务冲突。如需了解详情，请参阅防止 429 Resource Exhausted operation_too_costly 错误。
批处理软件包可以提高数据吞吐量，但不具有参照一致性等事务一致性功能。
大批量捆绑包可以降低吞吐量。如需了解详情，请参阅避免创建较大的集合。

DICOM 数据存储选项

将数据从照片归档和通信系统 (PACS) 发送到 Cloud Healthcare API 时，您可以使用以下方法实现高数据吞吐量：

使用 DICOM 消息服务元素 (DIMSE) 协议的开源 Cloud Healthcare API DICOM 适配器

当您将 PACS 与 Cloud Healthcare API 同步时，该适配器可优化数据吞吐量。在同步之前，请运行性能测试并验证适配器能否维持峰值数据吞吐量。

如果您无法使用 Storage Transfer Service 或其他转移选项将 DICOM 文件上传到 Cloud Storage，请使用此适配器。例如，您可能不符合以下 Storage Transfer Service 要求：

在代理池中托管代理的每个机器上装载文件系统，以检索来源数据。
如果您定期（而不是一次性批量加载）转移数据，则必须衡量数据大小随时间的变化，以发生变化。
最大限度地提高代理转移性能。
支付和分配 Cloud Storage 存储空间。
验证转移到 Cloud Storage 的数据。
在将数据导入 Cloud Healthcare API 后移除 Cloud Storage 资源，并修复任何导入错误。
根据临床系统的网络和存储容量安排批量注入间隔。

我们建议您使用 Storage Transfer Service 进行单次批量加载以填充 DICOM 存储区。定期使用 Storage Transfer Service 需要额外的工作，例如同步导入流水线。如需了解详情，请参阅 Storage Transfer Service 文件系统传输详细信息。

dicomStores.import

使用此方法可以存储大量 DICOM 数据。

DICOMweb 实体店交易

使用此方法以编程方式存储 DICOM 数据。

管理配额以优化数据吞吐量

以下各部分介绍了如何管理和规划配额以优化数据吞吐量。如需了解配额管理的一般最佳实践，请参阅配额管理最佳实践。

规划可预测流量的配额

规划配额要求，首先分析客户端应用的典型每日流量。即使流量是可预测的，您也应该规划更多配额，使其超出平均水平。这有助于避免错误，并在流量高峰或每日用量偶尔增加时提供安全外边距。

下图显示了向 Cloud Healthcare API 发送的请求，这些请求的大小一致，并且以可预测的模式发送：

配额用量与高峰时段的配额用量对比。 — **图 1.**Google Cloud 项目中的数据集和数据存储区的每小时 API 汇总负载。

为大批量请求规划配额

避免在高峰时段安排大批量作业。如需了解详情，请参阅一致地支持低容量事务。

下图显示了可预测的流量模式。但是，高峰流量高峰期间发生的批量批量请求会超出可用配额。这可能会导致项目中的所有请求出现 429 Resource Exhausted 错误。

比较高峰时段和典型高峰时段的配额用量。 — **图 2.**高峰时段因大批量作业而导致的资源使用量不规则分配。

如果您的系统有额外的灵活性配额，则小流量峰值不会造成错误，也不会导致可预测的峰值负载遇到错误。小的峰值必须在许多在 Google Cloud 项目中产生负载的数据存储区、应用和其他客户端之间分配。

为了防止单个大型作业导致流量高峰，请参阅避免大型捆绑包。

申请更多配额

如需保持高数据吞吐量并避免 429 Resource Exhausted 错误，请参阅本页面上的最佳实践，尤其是管理配额以优化数据吞吐量。这些最佳实践可确保您的客户端应用稳健可靠，并能根据请求量的变化而变化。在长期没有实施最佳实践的情况下，请求额外的配额不太可能防止出错。

如果您实施了最佳实践，但仍需要更多配额，请参阅申请更多配额的最佳实践。

数据注入吞吐量资源

如需详细了解数据注入吞吐量，请参阅在 Google Cloud 中管理工作负载的流量和负载。