使用 Spring Webflux 和新的 CosmosDB API v3 实现完全响应式开发

完全采取被动应对？

读了这篇关于 Spring Boot 和 MongoDB 的博客文章后，我决定将其移植到其他平台，以实现“完全响应式”：

• 从标准的 Spring Web 技术栈迁移到 Spring Webflux，后者使用 Reactor 项目来实现响应式 API。
• 将 CosmosDB 与 MongoDB API 的集成迁移到最新的 CosmosDB SDK（使用 SQL API）。如果您想了解更多关于 CosmosDB 的信息，请参阅此处的文档。

本文的目的是让我们从数据库到 Web 层实现完全响应式设计，以便研究所需的 API，并了解该架构的性能和可扩展性。

试用一下新的 Azure CosmosDB SDK

请注意，本文使用的是最新的 Azure Cosmos DB SDK，该 SDK 尚未最终完成，因此本文也是该 SDK 的全球首发预览版，方便我们进行测试和讨论。我（Julien Dubois）与微软的 SDK 团队直接联系，如果您有任何意见或问题，请随时在本文章下方留言或直接联系我！

• 这个新的 SDK 可在https://github.com/Azure/azure-cosmosdb-java/tree/v3上找到。
• 截至撰写本文时，相关文档和示例应用程序尚未准备就绪。
• 如果您正在使用之前的 SDK，请注意 MavenartifactId已更改，并且此 SDK 现在可在com.microsoft.azure:azure-cosmos获取。

由于这是新代码，所以可能会出现一些问题。例如，我在撰写这篇博文时就发现了这个问题，并在这里提出了一个修复方案。但正如您将在博文中看到的那样，这个新的 API 比之前的版本好得多，而且运行非常稳定。

使用新的 CosmosDB SDK 执行 CRUD 操作

新版 CosmosDB SDK 最令人振奋的消息是，它使用了 Project Reactor（类似于 Spring Webflux），而不是之前版本中老旧的 RxJava v1 API。这意味着它将返回 ` Object`Mono和Flux`Object` 对象，而这正是 Spring Webflux 所需要的，因此两者的集成将非常流畅便捷。

整个演示项目可在https://github.com/jdubois/spring-webflux-cosmosdb-sql/上找到，但我们重点关注存储库层，因为 CosmosDB SDK 的所有神奇功能都存在于此。

CosmosDB 配置和连接

我们创建了一个专门的Spring Boot 配置属性类来保存配置信息。该类用于我们的存储库层，其结构如下：

ConnectionPolicy connectionPolicy = new ConnectionPolicy();
connectionPolicy.connectionMode(ConnectionMode.DIRECT);

client = CosmosClient.builder()
    .endpoint(accountHost)
    .key(accountKey)
    .connectionPolicy(connectionPolicy)
    .build();

关键在于它使用了“直接模式”连接策略，而不是默认的“网关”策略：在我们的测试中，这带来了显著的差异，因为我们的响应式代码可能对网关来说效率过高，导致网关很快不堪重负。因此，我们遇到了大量的网关连接错误，而切换到直接模式后，这些错误立即消失了：如果条件允许，强烈建议在这种“响应式”场景下使用直接模式。

在本文提供的方法中init()（可在此处查看），我们还进行了两次阻塞调用来创建数据库及其容器。这里有一些有趣的调整：

• 我们的容器没有设置索引策略，因为我们使用的是默认设置indexingPolicy.automatic(false);。默认情况下，CosmosDB 会对所有存储对象的所有字段建立索引，这在插入数据时会带来显著的性能开销。虽然我们的测试不需要这样做，但我们也认为这种做法过于激进，应该针对每个具体用例进行优化。
• 容器默认使用 400 RU/s 创建database.createContainerIfNotExists(containerSettings, 400)。请谨慎设置此值，因为如果设置过高，可能会迅速产生大量费用。奇怪的是，1000MongoDB API 的默认设置是 400 RU/s，而400SQL API 的默认设置是 400 RU/s。无论如何，这是一个非常重要的设置，最好手动修正，而不是依赖默认值。
• 在执行操作时new CosmosContainerProperties(CONTAINER_NAME, "/id");，我们使用了我们的数据id作为分区键。这就是为什么使用以下方法可以获取项目container.getItem(id, id)：第一个参数是数据id，第二个参数是分区键，而分区键恰好也是数据id。这种方法运行良好，并且在我们的演示中，数据Project确实应该用于对所有内容进行分区，因此这在业务上是合理的。

创建、查找和删除项目

对于简单的操作，当我们拥有一个项目id（及其分区键）时，可以直接使用简单的 API，例如用于创建：

public Mono<Project> save(Project project) {
    project.setId(UUID.randomUUID().toString());
    return container.createItem(project)
        .map(i -> {
            Project savedProject = new Project();
            savedProject.setId(i.item().id());
            savedProject.setName(i.properties().getString("name"));
            return savedProject;
        });
}

由于没有提供 ORM，我们需要手动将返回结果映射到我们的领域对象。对于较大的对象来说，这会产生相当多的样板代码，但这在此类技术中很常见。当然，好消息是，在这种情况下我们可以轻松返回一个 `Object` Mono<Project>，这正是 Spring Webflux 所需要的。

查询

执行 SQL 查询稍微复杂一些，我们遇到了两个问题：

• 由于我们的id也是分区键，因此我们必须允许跨分区查询才能获取所有数据options.enableCrossPartitionQuery(true);。当然，这会造成性能损失。
• 由于我们需要分页数据，因此我们TOP 20在 SQL 查询中使用了 `--get_page_ ...

以下是生成的代码：

FeedOptions options = new FeedOptions();
options.enableCrossPartitionQuery(true);

return container.queryItems("SELECT TOP 20 * FROM Project p", options)
    .map(i -> {
        List<Project> results = new ArrayList<>();
        i.results().forEach(props -> {
            Project project = new Project();
            project.setId(props.id());
            project.setName(props.getString("name"));
            results.add(project);
        });
        return results;
    });

尝试限制返回值数量时要格外小心FeedOptions，因为你可能会想使用 `.` 来配置实例options.maxItemCount(20)。这样做行不通，而且非常棘手：

• 查询返回分页值，maxItemCount实际上指的是每页的值的数量。这源自 CosmosDB API（实际上，这是执行查询时底层使用的 HTTP 标头的名称），因此这个名称有一定的逻辑，但它确实容易造成误导，因此可能会引发问题。例如，如果您将其设置为 `false` 20，则意味着您仍然会获得完整的项目列表，只是分页显示，这将非常耗费资源。
• 请注意，文档中没有说明默认值maxItemCount是多少，但它被硬编码为 100。
• 正是由于这个 API，我们的查询返回的是一个页面流Flux<List<Project>>，而不是一个流Flux<Project>：我们得到的是页面流，而不仅仅是一个流。

性能测试

终于到了性能测试环节！我们将采用类似于之前关于 Spring Boot 和 MongoDB 的博文中的方法，以便您可以查看两种测试结果。但请注意，两者并不完全相同，因为这个应用是用JHipster构建的。JHipster 目前还不完全支持响应式编程，所以 Spring Webflux 是手动编写的，因此与之前的版本有很大不同：

• JHipster 应用具备安全、审计和指标功能：这些都会消耗大量性能，但如果您想在生产环境中部署真正的应用，它们是必不可少的。我们的 Spring Weblux 演示则要简单得多。
• 此外，JHipster 还提供了 Spring Webflux 演示中没有的一些性能调整，例如使用Afterburner。

因此，虽然比较这两个应用程序很有趣，但请记住它们并不完全是一对一的。

投入生产

正如我们在 Spring Boot/MongoDB 博客文章中所做的那样，我们使用提供的 Maven 插件将应用程序部署到Azure Web Apps上（请参阅我们 pom.xml 中的相关内容）。

测试场景

我们的测试场景是用 Gatling 构建的，可以在https://github.com/jdubois/spring-webflux-cosmosdb-sql/blob/master/src/test/gatling/user-files/simulations/ProjectGatlingTest.scala获取。这是一个简单的脚本，模拟用户使用 API 的过程：创建、查询和删除项目。

运行于 100 个用户

我们的第一次测试有 100 个用户，正如预期的那样，一切运行良好，因为并发请求并不多：

没什么好看的，我们继续吧！

用户数即将达到 500 人

用户数达到 500 人很有意思：它仍然运行良好，但我们出现了 3 个错误：

这是因为在 CosmosDB 上删除项目是一项开销很大的操作（会占用略多于 5 个 RU 的空间），因此在应用程序满负荷运行时执行此操作意味着我们达到了 API 调用限制。这是因为我们的应用程序比传统的 Spring Web 框架性能更高、更稳定：我们对后端的负载更高，我们需要考虑到这一点。

用户数达到 1000 人

为了应对超过 500 个用户，我们需要提高 CosmosDB 的 RU/s 值，就像我们之前对 Spring Web 所做的那样。这里 1200 RU/s 似乎足够了，但说实话，我们将其提升到了 50000 RU/s，这样在接下来的测试中就不用担心这个问题了。

一切进展顺利，没有任何问题，让我们扩大规模吧！

10,000 名用户

用户数达到 10,000 时出现了一个有趣的副作用：客户端的 Gatling 测试开始失败。因此，我们不得不增加ulimit负载测试机器的负载：这种情况很常见，但 Spring Web 却没有出现过，这再次表明完全响应式设计确实会产生影响，因为它的运行速度对于我们的负载测试机器来说太快了。尽管如此，我们仍然遇到了一些客户端错误，因为 Gatling 找不到我们服务器的主机名：很遗憾，这就是我们无法将用户数提升到 20,000 的原因……

用户数达到 5000 后，我们也开始遇到一些服务器错误：这些错误与客户端的错误基本相同，都是由于服务器上打开的文件过多导致的。由于我们使用的是Azure Web 应用，因此无法对服务器进行任何修改，但我们可以轻松地进行横向扩展。根据我们的测试，似乎 2 到 3 台服务器就足够了，但为了确保万无一失，我们使用了 5 台。请注意，在使用 Spring Web 时，我们使用了 20 台服务器：再次强调，这两个测试并非完全等效，还需要进一步完善，但很明显，使用响应式技术栈可以节省资源。

另请注意，我们的第 99 百分位性能非常出色，而且我们在一分钟内轻松扩展到每秒 10000 次请求，图表也非常清晰：

个人资料分析

由于我们的图表看起来一切都很好，但是我们的负载测试工具阻止了我们继续进行下去，所以我们决定使用YourKit进行一些性能分析，以确保没有任何东西阻碍或妨碍我们继续进行下去。

在本地机器上以 5000 个用户运行测试，我们发现没有线程阻塞：

此外，我们的CPU使用率极低，线程数稳定，内存使用率也很低且稳定：

我们还运行了一些 YourKit 分析，以查找瓶颈、锁或占用大量内存的对象：由于我们没有找到任何问题，所以我们就不赘述细节了！

结论与结语

通过采取“完全被动式”策略，我们获得了许多优势：

• 应用程序启动速度更快，占用CPU和内存更少。
• 它的吞吐量非常稳定。
• 它易于扩展。

然而，一切并非完美无缺：

• 还有更多代码，这些代码相当复杂，需要良好的技术背景。
• 所有操作都必须是非阻塞的：在这个简单的用例中，这很棒，但在实际应用中就复杂得多。例如，我喜欢使用 Spring Cache：它易于使用，而且使用 Memcached 或 Redis 服务器的成本可能比扩展 Cosmos DB 要低得多。但由于这是一个阻塞操作，所以我们不能在这里使用它！
• 只有当用户数量庞大时，才有必要采用“完全响应式”方案。如果每秒只有 500 个请求，那很可能是过度设计了。

我们也首次体验了 CosmosDB SDK 的 v3 版本：我们证明它在高负载下表现极其出色，而且幸运的是，它还能与 Spring Webflux 相同的响应式框架协同工作。
当然，目前仍然存在一些 bug，API 也需要改进，例如：

• 它不使用 setter/getter 方法，例如，我们以前会用 setter 和 getteroptions.maxItemCount(20)来设置最大物品数量，以及options.maxItemCount()获取该数量。我个人觉得这种方式不太好用。
• 我觉得很奇怪，创建条目时只需传入一个 POJO 对象container.createItem(project)，但如果需要读取该条目，却会收到一个返回值CosmosItemResponse，然后需要手动创建 POJO 对象。我认为我们可以像 MongoDB 那样，添加一个自动 POJO 映射器。
• 对于查询，我们可以使用流畅的查询构建器。

由于该 API 还有改进的空间，请花些时间阅读示例代码并提供反馈：在本帖下方留言、发送 Twitter 私信……请不要犹豫，我非常乐意将您的反馈转达给微软的 SDK 团队。