手把手教你学 Dapr - 5. 状态管理

作者：MASA技术团队

2022 年 1 月 19 日
本文字数：4437 字
阅读完需：约 15 分钟

介绍

使用状态管理，您的应用程序可以将数据作为键/值对存储在支持的状态存储中。

您的应用程序可以使用 Dapr 的状态管理 API 使用状态存储组件来保存和读取键/值对，如下图所示。例如，通过使用 HTTP POST，您可以保存键/值对，通过使用 HTTP GET，您可以读取键并返回其值。

特性

可插拔状态存储

Dapr 数据存储被建模为组件，可以在不更改代码的情况下更换它。例如：MySQL、Redis、Azure CosmosDB 等。

可配置的状态存储行为

Dapr 允许开发人员将额外的元数据附加到状态操作请求中，用以描述请求的处理方式。如：

并发要求
一致性要求

默认情况下，您的应用程序应假定数据存储最终一致并使用最后写入获胜的并发模式

并发

Dapr 支持使用 ETags 的乐观并发控制 (OCC)。当请求状态时，Dapr 总是将 ETag 属性附加到返回的状态。当用户代码尝试更新或删除状态时，应该通过请求正文附加 ETag 以进行更新或通过 If-Match 标头进行删除。只有当提供的 ETag 与状态存储中的 ETag 匹配时，写操作才能成功。建议您在使用 ETag 时使用重试策略来补偿此类冲突。

如果您的应用程序在写入请求时省略 ETag，则 Dapr 在处理请求时会跳过 ETag 检查。与使用 ETag 的先写赢模式相比，这实质上启用了最后写赢模式。

自动加密

Dapr 支持应用程序状态的自动客户端加密，并支持密钥轮换。这是一项预览功能，所有 Dapr 状态存储都支持。

一致性

Dapr 支持强一致性和最终一致性，最终一致性作为默认行为。

当使用强一致性时，Dapr 在确认写入请求之前等待所有副本（或指定的仲裁）确认。
当使用最终一致性时，一旦底层数据存储接受写入请求，Dapr 就会立即返回，即使这是单个副本。

批量操作

Dapr 支持两种类型的批量操作 - 批量（bulk）或多（multi）。

注：bulk 与 multi 的区别在于 bulk 不是事务性的，multi 是事务处理。

Actor 状态

事务状态存储可用于存储 Actor 状态。要指定用于 Actor 的状态存储，请在状态存储组件的元数据部分中将属性 actorStateStore 的值指定为 true。

注：Actors 状态以特定方案存储在事务状态存储中允许一致的查询。所以只能有一个状态存储组件被用于所有的 Actor。

直接查询状态存储

Dapr 无需任何转换即可保存和检索状态值。您可以直接从底层状态存储查询和聚合状态。

例如，要在 Redis 中获取与应用程序 ID “myApp” 关联的所有状态键，请使用：

KEYS "myApp*"

复制代码

查询 Actor 状态

如果数据存储支持 SQL 查询，您可以使用 SQL 查询查询参与者的状态。例如使用：

SELECT * FROM StateTable WHERE Id='<app-id>||<actor-type>||<actor-id>||<key>'

复制代码

您还可以跨 Actor 实例执行聚合查询，避免 Actor 框架常见的基于回合的并发限制。例如，要计算所有温度计 Actor 的平均温度，请使用：

SELECT AVG(value) FROM StateTable WHERE Id LIKE '<app-id>||<thermometer>||*||temperature'

复制代码

保存并获取状态

状态管理是任何应用程序最常见的需求之一：新的或遗留的、单体或微服务。处理不同的数据库、测试、处理重试和故障可能既费时又费力。

先决条件

准备好 Dapr 运行环境可以看之前的文章

手把手教你学Dapr - 3. 使用Dapr运行第一个.Net程序

设置状态存储

Windows 打开目录%USERPROFILE%\.dapr\components

创建文件statestore.yaml
使用redis作为状态存储的数据库

   apiVersion: dapr.io/v1alpha1   kind: Component   metadata:     name: statestore   spec:     type: state.redis     version: v1     metadata:     - name: redisHost       value: localhost:6379     - name: redisPassword       value: ""     - name: actorStateStore       value: "true"

复制代码

注：这个 yaml 已经通过 actorStateStore 开启了 Actor 状态

保存和检索单个状态

注：设置 app-id 很重要，因为状态键以该值作为前缀。如果您不设置它，则在运行时为您生成一个，下次运行该命令时将生成一个新的，您将无法再访问以前保存的状态。换句话说，如果你要共享状态可以自定义一个保留 app-id 作为共享状态而不是留空。

运行 Dapr Sidecar

运行一个空的 Sidecar，因为我们只用它来帮助访问状态存储，所以与之前不同的是，dapr run 后面没有接 dotnet run 去作为某一个程序的 Sidecar

dapr run --app-id myapp --dapr-http-port 3500 --dapr-grpc-port 50001

复制代码

创建客户端

创建控制台程序，添加Dapr.Client NuGet 包引用。

修改 Program.cs

using Dapr.Client;
var storeName = "statestore";var key = "myFirstKey";var value = "myFirstValue";
var client = new DaprClientBuilder().Build();await client.SaveStateAsync(storeName, key, value);Console.WriteLine("State has been stored");
var data = await client.GetStateAsync<string>(storeName, key);Console.WriteLine($"Got value: {data}");
Console.ReadKey();

复制代码

删除单个状态

await client.DeleteStateAsync(storeName, key);

复制代码

通过事务保存和检索多个状态

Dapr 还允许您在同一个调用中保存和检索多个状态。

var lst = new List<StateTransactionRequest>(){    new StateTransactionRequest("test1", System.Text.Encoding.UTF8.GetBytes("value1"), StateOperationType.Upsert),    new StateTransactionRequest("test2", System.Text.Encoding.UTF8.GetBytes("value2"), StateOperationType.Upsert),};await client.ExecuteStateTransactionAsync(storeName, lst);
var datas = await client.GetBulkStateAsync(storeName, lst.Select(r => r.Key).ToList(), 0);Console.WriteLine($"Got items: {string.Join(",", datas.Select(d => $"{d.Key}={d.Value}"))}");

复制代码

强一致性

使用强一致性时，Dapr 将确保底层状态存储在写入或删除状态之前，一旦数据被写入到所有副本或收到来自 quorum 的 ack，就会返回响应。

对于 GET 请求，Dapr 将确保存储在副本之间一致地返回最新数据。默认为最终一致性，除非在对状态 API 的请求中另有说明。

await client.SaveStateAsync(storeName, key, value, new StateOptions() { Consistency = ConsistencyMode.Strong });
var etagData = await client.GetStateAndETagAsync<string>(storeName, key, ConsistencyMode.Strong);Console.WriteLine($"ETag:{etagData.etag}");
await client.DeleteStateAsync(storeName, key, new StateOptions() { Consistency = ConsistencyMode.Strong });

复制代码

先写赢和最后写赢

Dapr 允许开发人员在使用数据存储时选择两种常见的并发模式：首先写入获胜和``最后写入获胜`。 First-Write-Wins 在您有多个应用程序实例的情况下很有用，所有实例都同时写入同一个键。

Dapr 的默认模式是最后写入获胜。

下面的例子展示了如何获取一个 ETag，然后使用它来保存状态，然后删除状态：

await client.SaveStateAsync(storeName, key, value, new StateOptions() { Concurrency = ConcurrencyMode.FirstWrite });var firstWriteWinData = await client.GetStateAndETagAsync<string>(storeName, key);var etag = firstWriteWinData.etag;
await client.TrySaveStateAsync(storeName, key, DateTime.Now.Ticks.ToString(), etag, new StateOptions() { Concurrency = ConcurrencyMode.FirstWrite });var firstWriteWinDeleteSucceeded = await client.TryDeleteStateAsync(storeName, key, etag);Console.WriteLine($"First write wins delete:{firstWriteWinDeleteSucceeded}");
firstWriteWinData = await client.GetStateAndETagAsync<string>(storeName, key);firstWriteWinDeleteSucceeded = await client.TryDeleteStateAsync(storeName, key, firstWriteWinData.etag);Console.WriteLine($"First write wins delete:{firstWriteWinDeleteSucceeded}");

复制代码

注：这里演示了 ETag 在更新后尝试删除失败的例子，最后再重新获取新的状态以修正 ETag 再删除

在不同的应用程序之间共享状态

为了实现状态共享，Dapr 支持以下键前缀策略

appid - 这是默认策略。 appid 前缀允许状态只能由具有指定 appid 的应用程序管理。所有状态键都将以 appid 为前缀，并以应用程序为范围。
name - 此设置使用状态存储组件的名称作为前缀。对于给定的状态存储，多个应用程序可以共享相同的状态。
none - 此设置不使用前缀。多个应用程序在不同的状态存储之间共享状态

举个例子：要指定前缀策略，请在状态组件上添加名为 keyPrefix 的元数据键

apiVersion: dapr.io/v1alpha1kind: Componentmetadata:  name: statestore  namespace: productionspec:  type: state.redis  version: v1  metadata:  - name: keyPrefix    value: <key-prefix-strategy>

复制代码

注：此示例演示相对较复杂，思路大概是使用多个 statestore.yaml，然后根据不同的 storename 切换不同策略即可。感兴趣的小伙伴可以自行尝试。

自动加密状态并管理密钥轮换

注：截止目前，这个功能是个预览版，感兴趣的小伙伴可以自行尝试

应用程序状态通常需要静态加密，以在企业工作负载或受监管环境中提供更强的安全性。 Dapr 提供基于 AES256 的自动客户端加密。

状态的生存时间（TTL）

Dapr 为每个状态在请求时设置生存时间 (TTL)。这意味着应用程序可以为每个存储的状态设置生存时间，并且这些状态在到期后无法检索。

注：只有一部分 Dapr 状态存储组件与状态 TTL 兼容。对于支持的状态存储，只需在发布消息时设置 ttlInSeconds 元数据。其他状态存储将忽略此值。

await client.SaveStateAsync(storeName, key, value, metadata: new Dictionary<string, string>() { { "ttlInSeconds", "3" } });var ttlData = await client.GetStateAsync<string>(storeName, key);Console.WriteLine($"TTL Data:{ttlData}");
Thread.Sleep(5000);ttlData = await client.GetStateAsync<string>(storeName, key);Console.WriteLine($"TTL Data:{ttlData}");

复制代码