This is the multi-page printable view of this section. Click here to print.
状态管理
创建长期运行的有状态服务
- 1: 状态管理概述
- 2: 操作指南：保存和获取状态
- 3: 操作方法：查询状态
- 4: 操作指南：构建有状态服务
- 5: How-To：启用事务性 Outbox 模式
- 6: 操作指南：在应用程序之间共享状态
- 7: 操作指南：加密应用状态
- 8: 使用后端状态存储
- 8.1: Azure Cosmos DB
- 8.2: Redis
- 8.3: SQL Server
- 9: 状态生存时间 (TTL)
1 - 状态管理概述
状态管理 API 构建块的概述
您的应用程序可以使用 Dapr 的状态管理 API 在受支持的状态存储中保存、读取和查询键值对。使用状态存储组件，您可以构建有状态的、长时间运行的应用程序，用于保存和检索其状态（例如购物车或游戏会话状态）。例如，在下图中：
- 使用 HTTP POST 来保存或查询键值对。
- 使用 HTTP GET 来读取特定键并获取其值。
以下概述视频和演示展示了 Dapr 状态管理的工作原理。
功能
通过状态管理 API 构建块，您的应用程序可以利用通常构建复杂且容易出错的特性，包括：
- 设置并发控制和数据一致性选项。
- 执行批量更新操作 CRUD，包括多个事务性操作。
- 查询和过滤键值数据。
以下是状态管理 API 提供的功能：
可插拔状态存储
Dapr 数据存储被建模为组件，可以在不更改服务代码的情况下进行替换。请参阅受支持的状态存储查看列表。
可配置的状态存储行为
使用 Dapr，您可以在状态操作请求中包含额外的元数据，用于描述您希望如何处理该请求。您可以附加：
- 并发要求
- 一致性要求
默认情况下，您的应用程序应假设数据存储是最终一致的，并使用**最后写入胜出（last-write-wins）**的并发模式。
并非所有存储都是平等的。为了确保应用程序的可移植性，您可以查询存储的元数据功能，并使代码适应不同的存储功能。
并发
Dapr 使用 ETags 支持乐观并发控制（OCC）。当请求状态值时，Dapr 始终将 ETag 属性附加到返回的状态。当用户代码：
- 更新状态时，预期会通过请求正文附加 ETag。
- 删除状态时，预期会通过
If-Match标头附加 ETag。
当提供的 ETag 与状态存储中的 ETag 匹配时，write 操作成功。
为什么 Dapr 选择乐观并发控制（OCC）
在许多应用程序中，数据更新冲突很少发生，因为客户端在业务上下文中自然分离以操作不同的数据。但是，如果您的应用程序选择使用 ETags，ETag 不匹配可能会导致请求被拒绝。建议您在代码中使用重试策略来补偿使用 ETags 时的冲突。
如果您的应用程序在写入请求中省略 ETags，Dapr 将在处理请求时跳过 ETag 检查。这将启用最后写入胜出模式，而使用 ETags 则是**首次写入胜出（first-write-wins）**模式。
关于 ETags 的说明
对于本身不支持 ETags 的存储，相应的 Dapr 状态存储实现应模拟 ETags，并在处理状态时遵循 Dapr 状态管理 API 规范。由于 Dapr 状态存储实现在技术上是底层数据存储的客户端，因此使用存储提供的并发控制机制进行模拟应该很简单。阅读 API 参考以了解如何设置并发选项。
一致性
Dapr 支持强一致性和最终一致性，默认行为是最终一致性。
- 强一致性：Dapr 等待所有副本（或指定的法定人数）确认后，才确认写入请求。
- 最终一致性：Dapr 在底层数据存储接受写入请求后立即返回，即使只是单个副本。
阅读 API 参考以了解如何设置一致性选项。
设置内容类型
状态存储组件可能根据内容类型以不同方式维护和操作数据。Dapr 支持在状态管理 API中传递内容类型作为请求元数据的一部分。
设置内容类型是_可选的_，组件决定是否使用它。Dapr 只提供将此信息传递给组件的手段。
- 使用 HTTP API：通过 URL 查询参数
metadata.contentType设置内容类型。例如，http://localhost:3500/v1.0/state/store?metadata.contentType=application/json。 - 使用 gRPC API：通过在请求元数据中添加键值对
"contentType" : <content type>来设置内容类型。
多操作
Dapr 支持两种类型的多读或多写操作：批量或事务。阅读 API 参考以了解如何使用批量和多选项。
批量读取操作
您可以将多个读取请求分组到批量（或批处理）操作中。在批量操作中，Dapr 将读取请求作为单独的请求提交给底层数据存储，并将它们作为单个结果返回。
事务操作
您可以将写入、更新和删除操作分组到一个请求中，然后作为原子事务处理。该请求将作为事务操作集成功或失败。
Actor 状态
事务状态存储可用于存储 actor 状态。要指定用于 actor 的状态存储，请在状态存储组件的元数据部分将属性 actorStateStore 的值指定为 true。Actor 状态以特定方案存储在事务状态存储中，从而允许一致的查询。所有 actor 只能使用单个状态存储组件作为状态存储。如果您的状态存储由分布式数据库支持，您必须确保它提供强一致性。阅读状态 API 参考和 actor API 参考以了解有关 actor 状态存储的更多信息。
Actor 状态的生存时间（TTL）
您在保存 actor 状态时，应始终设置 TTL 元数据字段（ttlInSeconds），或使用所选 SDK 中的等效 API 调用，以确保状态最终被删除。阅读 actor 概述以获取更多信息。
状态加密
Dapr 支持应用程序状态的自动客户端加密，并支持密钥轮换。这适用于所有 Dapr 状态存储。有关更多信息，请阅读如何：加密应用程序状态主题。
应用程序之间共享状态
不同的应用程序在共享状态方面的需求各不相同。在一种场景中，您可能希望将所有状态封装在给定的应用程序中，并让 Dapr 为您管理访问。在另一种场景中，您可能希望两个应用程序处理相同的状态以获取和保存相同的键。
Dapr 使状态能够：
- 隔离到应用程序。
- 在应用程序之间的状态存储中共享。
- 在不同状态存储之间的多个应用程序之间共享。
有关更多详细信息，请阅读如何：在应用程序之间共享状态
启用发件箱模式
Dapr 使开发者能够使用发件箱模式在事务状态存储和任何消息代理之间实现单个事务。有关更多信息，请阅读如何启用事务性发件箱消息传递
查询状态
有两种查询状态的方式：
- 使用 Dapr 运行时中提供的状态管理查询 API。
- 使用存储的原生 SDK 直接查询状态存储。
查询 API
使用_可选的_状态管理查询 API，您可以查询保存在状态存储中的键值数据，而不管底层数据库或存储技术如何。使用状态管理查询 API，您可以过滤、排序和分页键值数据。有关更多详细信息，请阅读如何：查询状态。
直接查询状态存储
Dapr 保存和检索状态值时不进行任何转换。您可以直接从底层状态存储查询和聚合状态。 例如，要在 Redis 中获取与应用程序 ID “myApp” 关联的所有状态键，请使用：
KEYS "myApp*"
关于直接查询的说明
由于您不是通过 Dapr 运行时调用，因此状态存储的直接查询不受 Dapr 并发控制约束。您看到的是已提交数据的快照，可用于跨多个 actor 的只读查询。写入应通过 Dapr 状态管理或 actor API 完成。查询 actor 状态
如果数据存储支持 SQL 查询，您可以使用 SQL 查询 actor 的状态。例如：
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'
状态生存时间（TTL）
Dapr 支持每个状态设置请求的生存时间（TTL）。这意味着应用程序可以为每个存储的状态设置生存时间，这些状态在过期后无法检索。
状态管理 API
状态管理 API 可在状态管理 API 参考中找到，该参考描述了如何通过提供键来检索、保存、删除和查询状态值。
尝试状态管理
快速入门和教程
想要测试 Dapr 状态管理 API？通过以下快速入门和教程了解状态管理的实际应用：
| 快速入门/教程 | 描述 |
|---|---|
| 状态管理快速入门 | 使用状态管理 API 创建有状态的应用程序。 |
| Hello World | 推荐 演示如何在本地运行 Dapr。突出显示服务调用和状态管理。 |
| Hello World Kubernetes | 推荐 演示如何在 Kubernetes 中运行 Dapr。突出显示服务调用和_状态管理_。 |
直接在应用程序中开始使用状态管理
想跳过快速入门？没问题。您可以直接在应用程序中试用状态管理构建块。安装 Dapr后，您可以从状态管理操作指南开始使用状态管理 API。
后续步骤
- 开始学习状态管理操作指南，从以下开始：
- 查看状态存储组件列表
- 阅读状态管理 API 参考
- 阅读actor API 参考
2 - 操作指南：保存和获取状态
使用键值对持久化状态
状态管理是任何新应用程序、遗留应用程序、单体应用程序或微服务应用程序最常见的需求之一。处理和测试不同的数据库库以及处理重试和故障可能既困难又耗时。
在本指南中，您将学习使用键/值状态 API 的基础知识，使应用程序能够保存、获取和删除状态。
下面的代码示例大致描述了一个处理订单的应用程序，该应用程序使用带有 Dapr 边车的订单处理服务。订单处理服务使用 Dapr 将状态存储在 Redis 状态存储中。
设置状态存储
状态存储组件代表 Dapr 用于与数据库通信的资源。
为了本指南的目的，我们将使用 Redis 状态存储，但支持列表中的任何状态存储都可以使用。
当您在自托管模式下运行 dapr init 时，Dapr 会创建一个默认的 Redis statestore.yaml 并在您的本地机器上运行 Redis 状态存储，位于：
- 在 Windows 上，位于
%UserProfile%\.dapr\components\statestore.yaml - 在 Linux/MacOS 上，位于
~/.dapr/components/statestore.yaml
使用 statestore.yaml 组件，您可以轻松替换底层组件，而无需更改应用程序代码。
要将其部署到 Kubernetes 集群，请在下面的 YAML 中填写状态存储组件的 metadata 连接详细信息，保存为 statestore.yaml，然后运行 kubectl apply -f statestore.yaml。
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
重要
设置一个app-id，因为状态键会以此值作为前缀。如果您不设置 app-id，系统会在运行时为您生成一个。下次运行该命令时，会生成一个新的 app-id，您将无法访问之前保存的状态。保存和检索单个状态
下面的示例展示了如何使用 Dapr 状态管理 API 保存和检索单个键/值对。
using System.Text;
using System.Threading.Tasks;
using Dapr.Client;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprClient();
var app = builder.Build();
var random = new Random();
//从依赖注入注册中解析 DaprClient
using var client = app.Services.GetRequiredService<DaprClient>();
while(true)
{
await Task.Delay(TimeSpan.FromSeconds(5));
var orderId = random.Next(1,1000);
//使用 Dapr SDK 保存和获取状态
await client.SaveStateAsync(DAPR_STORE_NAME, "order_1", orderId.ToString());
await client.SaveStateAsync(DAPR_STORE_NAME, "order_2", orderId.ToString());
var result = await client.GetStateAsync<string>(DAPR_STORE_NAME, "order_1");
Console.WriteLine($"Result after get: {result}");
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 dotnet run
//依赖项
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.State;
import io.dapr.client.domain.TransactionalStateOperation;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.util.Random;
import java.util.concurrent.TimeUnit;
//代码
@SpringBootApplication
public class OrderProcessingServiceApplication {
private static final Logger log = LoggerFactory.getLogger(OrderProcessingServiceApplication.class);
private static final String STATE_STORE_NAME = "statestore";
public static void main(String[] args) throws InterruptedException{
while(true) {
TimeUnit.MILLISECONDS.sleep(5000);
Random random = new Random();
int orderId = random.nextInt(1000-1) + 1;
DaprClient client = new DaprClientBuilder().build();
//使用 Dapr SDK 保存和获取状态
client.saveState(STATE_STORE_NAME, "order_1", Integer.toString(orderId)).block();
client.saveState(STATE_STORE_NAME, "order_2", Integer.toString(orderId)).block();
Mono<State<String>> result = client.getState(STATE_STORE_NAME, "order_1", String.class);
log.info("Result after get" + result);
}
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 mvn spring-boot:run
#依赖项
import random
from time import sleep
import requests
import logging
from dapr.clients import DaprClient
from dapr.clients.grpc._state import StateItem
from dapr.clients.grpc._request import TransactionalStateOperation, TransactionOperationType
#代码
logging.basicConfig(level = logging.INFO)
DAPR_STORE_NAME = "statestore"
while True:
sleep(random.randrange(50, 5000) / 1000)
orderId = random.randint(1, 1000)
with DaprClient() as client:
#使用 Dapr SDK 保存和获取状态
client.save_state(DAPR_STORE_NAME, "order_1", str(orderId))
result = client.get_state(DAPR_STORE_NAME, "order_1")
logging.info('Result after get: ' + result.data.decode('utf-8'))
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 -- python3 OrderProcessingService.py
// 依赖项
import (
"context"
"log"
"math/rand"
"strconv"
"time"
dapr "github.com/dapr/go-sdk/client"
)
// 代码
func main() {
const STATE_STORE_NAME = "statestore"
rand.Seed(time.Now().UnixMicro())
for i := 0; i < 10; i++ {
orderId := rand.Intn(1000-1) + 1
client, err := dapr.NewClient()
if err != nil {
panic(err)
}
defer client.Close()
ctx := context.Background()
err = client.SaveState(ctx, STATE_STORE_NAME, "order_1", []byte(strconv.Itoa(orderId)), nil)
if err != nil {
panic(err)
}
result, err := client.GetState(ctx, STATE_STORE_NAME, "order_1", nil)
if err != nil {
panic(err)
}
log.Println("Result after get:", string(result.Value))
time.Sleep(2 * time.Second)
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 go run OrderProcessingService.go
//依赖项
import { DaprClient, HttpMethod, CommunicationProtocolEnum } from '@dapr/dapr';
//代码
const daprHost = "127.0.0.1";
var main = function() {
for(var i=0;i<10;i++) {
sleep(5000);
var orderId = Math.floor(Math.random() * (1000 - 1) + 1);
start(orderId).catch((e) => {
console.error(e);
process.exit(1);
});
}
}
async function start(orderId) {
const client = new DaprClient({
daprHost,
daprPort: process.env.DAPR_HTTP_PORT,
communicationProtocol: CommunicationProtocolEnum.HTTP,
});
const STATE_STORE_NAME = "statestore";
//使用 Dapr SDK 保存和获取状态
await client.state.save(STATE_STORE_NAME, [
{
key: "order_1",
value: orderId.toString()
},
{
key: "order_2",
value: orderId.toString()
}
]);
var result = await client.state.get(STATE_STORE_NAME, "order_1");
console.log("Result after get: " + result);
}
function sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
main();
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 npm start
启动 Dapr 边车：
dapr run --app-id orderprocessing --dapr-http-port 3601
在单独的终端中，将一个键/值对保存到您的状态存储中：
curl -X POST -H "Content-Type: application/json" -d '[{ "key": "order_1", "value": "250"}]' http://localhost:3601/v1.0/state/statestore
现在获取您刚刚保存的状态：
curl http://localhost:3601/v1.0/state/statestore/order_1
重启您的边车并尝试再次检索状态，以观察状态独立于应用程序持久存在。
启动 Dapr 边车：
dapr --app-id orderprocessing --dapr-http-port 3601 run
在单独的终端中，将一个键/值对保存到您的状态存储中：
Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '[{"key": "order_1", "value": "250"}]' -Uri 'http://localhost:3601/v1.0/state/statestore'
现在获取您刚刚保存的状态：
Invoke-RestMethod -Uri 'http://localhost:3601/v1.0/state/statestore/order_1'
重启您的边车并尝试再次检索状态，以观察状态独立于应用程序持久存在。
删除状态
以下是利用 Dapr SDK 删除状态的代码示例。
using Dapr.Client;
using System.Threading.Tasks;
const string DAPR_STORE_NAME = "statestore";
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprClient();
var app = builder.Build();
//从依赖注入注册中解析 DaprClient
using var client = app.Services.GetRequiredService<DaprClient>();
//使用 DaprClient 删除状态
await client.DeleteStateAsync(DAPR_STORE_NAME, "order_1", cancellationToken: cancellationToken);
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 dotnet run
//依赖项
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import org.springframework.boot.autoconfigure.SpringBootApplication;
//代码
@SpringBootApplication
public class OrderProcessingServiceApplication {
public static void main(String[] args) throws InterruptedException{
String STATE_STORE_NAME = "statestore";
//使用 Dapr SDK 删除状态
DaprClient client = new DaprClientBuilder().build();
String storedEtag = client.getState(STATE_STORE_NAME, "order_1", String.class).block().getEtag();
client.deleteState(STATE_STORE_NAME, "order_1", storedEtag, null).block();
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 mvn spring-boot:run
#依赖项
from dapr.clients.grpc._request import TransactionalStateOperation, TransactionOperationType
#代码
logging.basicConfig(level = logging.INFO)
DAPR_STORE_NAME = "statestore"
#使用 Dapr SDK 删除状态
with DaprClient() as client:
client.delete_state(store_name=DAPR_STORE_NAME, key="order_1")
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 -- python3 OrderProcessingService.py
//依赖项
import (
"context"
dapr "github.com/dapr/go-sdk/client"
)
//代码
func main() {
STATE_STORE_NAME := "statestore"
//使用 Dapr SDK 删除状态
client, err := dapr.NewClient()
if err != nil {
panic(err)
}
defer client.Close()
ctx := context.Background()
if err := client.DeleteState(ctx, STATE_STORE_NAME, "order_1"); err != nil {
panic(err)
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 go run OrderProcessingService.go
//依赖项
import { DaprClient, HttpMethod, CommunicationProtocolEnum } from '@dapr/dapr';
//代码
const daprHost = "127.0.0.1";
var main = function() {
const STATE_STORE_NAME = "statestore";
//使用 Dapr SDK 保存和获取状态
const client = new DaprClient({
daprHost,
daprPort: process.env.DAPR_HTTP_PORT,
communicationProtocol: CommunicationProtocolEnum.HTTP,
});
await client.state.delete(STATE_STORE_NAME, "order_1");
}
main();
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 npm start
使用与上面相同的 Dapr 实例运行，执行以下命令：
curl -X DELETE 'http://localhost:3601/v1.0/state/statestore/order_1'
尝试再次获取状态。请注意，没有返回任何值。
使用与上面相同的 Dapr 实例运行，执行以下命令：
Invoke-RestMethod -Method Delete -Uri 'http://localhost:3601/v1.0/state/statestore/order_1'
尝试再次获取状态。请注意，没有返回任何值。
保存和检索多个状态
以下是利用 Dapr SDK 保存和检索多个状态的代码示例。
using Dapr.Client;
using System.Threading.Tasks;
const string DAPR_STORE_NAME = "statestore";
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprClient();
var app = builder.Build();
//从依赖注入注册中解析 DaprClient
using var client = app.Services.GetRequiredService<DaprClient>();
IReadOnlyList<BulkStateItem> multipleStateResult = await client.GetBulkStateAsync(DAPR_STORE_NAME, new List<string> { "order_1", "order_2" }, parallelism: 1);
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 dotnet run
上面的示例返回一个 BulkStateItem，其中包含您保存到状态的值的序列化格式。如果您希望 SDK 在每个批量响应项中对值进行反序列化，可以改为使用以下代码：
using Dapr.Client;
using System.Threading.Tasks;
const string DAPR_STORE_NAME = "statestore";
var builder = WebApplication.CreateBuilder(args);
builder.Serivces.AddDaprClient();
var app = builder.Build();
//从依赖注入注册中解析 DaprClient
using var client = app.Services.GetRequiredService<DaprClient>();
IReadOnlyList<BulkStateItem<Widget>> mulitpleStateResult = await client.GetBulkStateAsync<Widget>(DAPR_STORE_NAME, new List<string> { "widget_1", "widget_2" }, parallelism: 1);
record Widget(string Size, string Color);
//依赖项
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.State;
import java.util.Arrays;
//代码
@SpringBootApplication
public class OrderProcessingServiceApplication {
private static final Logger log = LoggerFactory.getLogger(OrderProcessingServiceApplication.class);
public static void main(String[] args) throws InterruptedException{
String STATE_STORE_NAME = "statestore";
//使用 Dapr SDK 检索多个状态
DaprClient client = new DaprClientBuilder().build();
Mono<List<State<String>>> resultBulk = client.getBulkState(STATE_STORE_NAME,
Arrays.asList("order_1", "order_2"), String.class);
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 mvn spring-boot:run
#依赖项
from dapr.clients import DaprClient
from dapr.clients.grpc._state import StateItem
#代码
logging.basicConfig(level = logging.INFO)
DAPR_STORE_NAME = "statestore"
orderId = 100
#使用 Dapr SDK 保存和检索多个状态
with DaprClient() as client:
client.save_bulk_state(store_name=DAPR_STORE_NAME, states=[StateItem(key="order_2", value=str(orderId))])
result = client.get_bulk_state(store_name=DAPR_STORE_NAME, keys=["order_1", "order_2"], states_metadata={"metakey": "metavalue"}).items
logging.info('Result after get bulk: ' + str(result))
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 -- python3 OrderProcessingService.py
// 依赖项
import (
"context"
"log"
"math/rand"
"strconv"
"time"
dapr "github.com/dapr/go-sdk/client"
)
// 代码
func main() {
const STATE_STORE_NAME = "statestore"
rand.Seed(time.Now().UnixMicro())
for i := 0; i < 10; i++ {
orderId := rand.Intn(1000-1) + 1
client, err := dapr.NewClient()
if err != nil {
panic(err)
}
defer client.Close()
ctx := context.Background()
err = client.SaveState(ctx, STATE_STORE_NAME, "order_1", []byte(strconv.Itoa(orderId)), nil)
if err != nil {
panic(err)
}
keys := []string{"key1", "key2", "key3"}
items, err := client.GetBulkState(ctx, STATE_STORE_NAME, keys, nil, 100)
if err != nil {
panic(err)
}
for _, item := range items {
log.Println("Item from GetBulkState:", string(item.Value))
}
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 go run OrderProcessingService.go
//依赖项
import { DaprClient, HttpMethod, CommunicationProtocolEnum } from '@dapr/dapr';
//代码
const daprHost = "127.0.0.1";
var main = function() {
const STATE_STORE_NAME = "statestore";
var orderId = 100;
//使用 Dapr SDK 保存和检索多个状态
const client = new DaprClient({
daprHost,
daprPort: process.env.DAPR_HTTP_PORT,
communicationProtocol: CommunicationProtocolEnum.HTTP,
});
await client.state.save(STATE_STORE_NAME, [
{
key: "order_1",
value: orderId.toString()
},
{
key: "order_2",
value: orderId.toString()
}
]);
result = await client.state.getBulk(STATE_STORE_NAME, ["order_1", "order_2"]);
}
main();
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 npm start
使用与上面相同的 Dapr 实例运行，将两个键/值对保存到您的状态存储中：
curl -X POST -H "Content-Type: application/json" -d '[{ "key": "order_1", "value": "250"}, { "key": "order_2", "value": "550"}]' http://localhost:3601/v1.0/state/statestore
现在获取您刚刚保存的状态：
curl -X POST -H "Content-Type: application/json" -d '{"keys":["order_1", "order_2"]}' http://localhost:3601/v1.0/state/statestore/bulk
使用与上面相同的 Dapr 实例运行，将两个键/值对保存到您的状态存储中：
Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '[{ "key": "order_1", "value": "250"}, { "key": "order_2", "value": "550"}]' -Uri 'http://localhost:3601/v1.0/state/statestore'
现在获取您刚刚保存的状态：
Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '{"keys":["order_1", "order_2"]}' -Uri 'http://localhost:3601/v1.0/state/statestore/bulk'
执行状态事务
注意
状态事务需要支持多项目事务的状态存储。有关完整列表，请参阅支持的状态存储页面。以下是利用 Dapr SDK 执行状态事务的代码示例。
using Dapr.Client;
using System.Threading.Tasks;
const string DAPR_STORE_NAME = "statestore";
var builder = WebApplication.CreateBuilder(args);
builder.Serivces.AddDaprClient();
var app = builder.Build();
//从依赖注入注册中解析 DaprClient
using var client = app.Services.GetRequiredService<DaprClient>();
var random = new Random();
while (true)
{
await Task.Delay(TimeSpan.FromSeconds(5));
var orderId = random.Next(1, 1000);
var requests = new List<StateTransactionRequest>
{
new StateTransactionRequest("order_3", JsonSerializer.SerializeToUtf8Bytes(orderId.ToString()), StateOperationType.Upsert),
new StateTransactionRequest("order_2", null, StateOperationType.Delete)
};
var cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = cancellationTokenSource.Token;
//使用 DaprClient 执行状态事务
await client.ExecuteStateTransactionAsync(DAPR_STORE_NAME, requests, cancellationToken: cancellationToken);
Console.WriteLine($"Order requested: {orderId}");
Console.WriteLine($"Result: {result}");
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 dotnet run
//依赖项
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.State;
import io.dapr.client.domain.TransactionalStateOperation;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
//代码
@SpringBootApplication
public class OrderProcessingServiceApplication {
private static final Logger log = LoggerFactory.getLogger(OrderProcessingServiceApplication.class);
private static final String STATE_STORE_NAME = "statestore";
public static void main(String[] args) throws InterruptedException{
while(true) {
TimeUnit.MILLISECONDS.sleep(5000);
Random random = new Random();
int orderId = random.nextInt(1000-1) + 1;
DaprClient client = new DaprClientBuilder().build();
List<TransactionalStateOperation<?>> operationList = new ArrayList<>();
operationList.add(new TransactionalStateOperation<>(TransactionalStateOperation.OperationType.UPSERT,
new State<>("order_3", Integer.toString(orderId), "")));
operationList.add(new TransactionalStateOperation<>(TransactionalStateOperation.OperationType.DELETE,
new State<>("order_2")));
//使用 Dapr SDK 执行状态事务
client.executeStateTransaction(STATE_STORE_NAME, operationList).block();
log.info("Order requested: " + orderId);
}
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 mvn spring-boot:run
#依赖项
import random
from time import sleep
import requests
import logging
from dapr.clients import DaprClient
from dapr.clients.grpc._state import StateItem
from dapr.clients.grpc._request import TransactionalStateOperation, TransactionOperationType
#代码
logging.basicConfig(level = logging.INFO)
DAPR_STORE_NAME = "statestore"
while True:
sleep(random.randrange(50, 5000) / 1000)
orderId = random.randint(1, 1000)
with DaprClient() as client:
#使用 Dapr SDK 执行状态事务
client.execute_state_transaction(store_name=DAPR_STORE_NAME, operations=[
TransactionalStateOperation(
operation_type=TransactionOperationType.upsert,
key="order_3",
data=str(orderId)),
TransactionalStateOperation(key="order_3", data=str(orderId)),
TransactionalStateOperation(
operation_type=TransactionOperationType.delete,
key="order_2",
data=str(orderId)),
TransactionalStateOperation(key="order_2", data=str(orderId))
])
client.delete_state(store_name=DAPR_STORE_NAME, key="order_1")
logging.basicConfig(level = logging.INFO)
logging.info('Order requested: ' + str(orderId))
logging.info('Result: ' + str(result))
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 -- python3 OrderProcessingService.py
// 依赖项
package main
import (
"context"
"log"
"math/rand"
"strconv"
"time"
dapr "github.com/dapr/go-sdk/client"
)
// 代码
func main() {
const STATE_STORE_NAME = "statestore"
rand.Seed(time.Now().UnixMicro())
for i := 0; i < 10; i++ {
orderId := rand.Intn(1000-1) + 1
client, err := dapr.NewClient()
if err != nil {
panic(err)
}
defer client.Close()
ctx := context.Background()
err = client.SaveState(ctx, STATE_STORE_NAME, "order_1", []byte(strconv.Itoa(orderId)), nil)
if err != nil {
panic(err)
}
result, err := client.GetState(ctx, STATE_STORE_NAME, "order_1", nil)
if err != nil {
panic(err)
}
ops := make([]*dapr.StateOperation, 0)
data1 := "data1"
data2 := "data2"
op1 := &dapr.StateOperation{
Type: dapr.StateOperationTypeUpsert,
Item: &dapr.SetStateItem{
Key: "key1",
Value: []byte(data1),
},
}
op2 := &dapr.StateOperation{
Type: dapr.StateOperationTypeDelete,
Item: &dapr.SetStateItem{
Key: "key2",
Value: []byte(data2),
},
}
ops = append(ops, op1, op2)
meta := map[string]string{}
err = client.ExecuteStateTransaction(ctx, STATE_STORE_NAME, meta, ops)
log.Println("Result after get:", string(result.Value))
time.Sleep(2 * time.Second)
}
}
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 go run OrderProcessingService.go
//依赖项
import { DaprClient, HttpMethod, CommunicationProtocolEnum } from '@dapr/dapr';
//代码
const daprHost = "127.0.0.1";
var main = function() {
for(var i=0;i<10;i++) {
sleep(5000);
var orderId = Math.floor(Math.random() * (1000 - 1) + 1);
start(orderId).catch((e) => {
console.error(e);
process.exit(1);
});
}
}
async function start(orderId) {
const client = new DaprClient({
daprHost,
daprPort: process.env.DAPR_HTTP_PORT,
communicationProtocol: CommunicationProtocolEnum.HTTP,
});
const STATE_STORE_NAME = "statestore";
//使用 Dapr SDK 保存和检索多个状态
await client.state.transaction(STATE_STORE_NAME, [
{
operation: "upsert",
request: {
key: "order_3",
value: orderId.toString()
}
},
{
operation: "delete",
request: {
key: "order_2"
}
}
]);
}
function sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
main();
要为上述示例应用程序启动 Dapr 边车，请运行类似以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 npm start
使用与上面相同的 Dapr 实例运行，执行两个状态事务：
curl -X POST -H "Content-Type: application/json" -d '{"operations": [{"operation":"upsert", "request": {"key": "order_1", "value": "250"}}, {"operation":"delete", "request": {"key": "order_2"}}]}' http://localhost:3601/v1.0/state/statestore/transaction
现在查看您的状态事务的结果：
curl -X POST -H "Content-Type: application/json" -d '{"keys":["order_1", "order_2"]}' http://localhost:3601/v1.0/state/statestore/bulk
使用与上面相同的 Dapr 实例运行，将两个键/值对保存到您的状态存储中：
Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '{"operations": [{"operation":"upsert", "request": {"key": "order_1", "value": "250"}}, {"operation":"delete", "request": {"key": "order_2"}}]}' -Uri 'http://localhost:3601/v1.0/state/statestore/transaction'
现在查看您的状态事务的结果：
Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '{"keys":["order_1", "order_2"]}' -Uri 'http://localhost:3601/v1.0/state/statestore/bulk'
后续步骤
3 - 操作方法：查询状态
使用查询 API 查询状态存储
alpha
状态查询 API 目前处于 alpha 阶段。通过状态查询 API，你可以检索、筛选和排序存储在状态存储组件中的键/值数据。查询 API 不是完整查询语言的替代品。
尽管状态存储是键/值存储，但 value 可能是一个具有自身层次结构、键和值的 JSON 文档。查询 API 允许你使用这些键/值来检索相应的文档。
查询状态
通过 HTTP POST/PUT 或 gRPC 提交查询请求。请求体是包含 3 个条目的 JSON 映射：
filtersortpage
filter
filter 以树的形式指定查询条件，其中每个节点代表一元或多操作数操作。
支持以下操作：
| 操作符 | 操作数 | 描述 |
|---|---|---|
EQ | key:value | key == value |
NEQ | key:value | key != value |
GT | key:value | key > value |
GTE | key:value | key >= value |
LT | key:value | key < value |
LTE | key:value | key <= value |
IN | key:[]value | key == value[0] OR key == value[1] OR … OR key == value[n] |
AND | []operation | operation[0] AND operation[1] AND … AND operation[n] |
OR | []operation | operation[0] OR operation[1] OR … OR operation[n] |
操作数中的 key 类似于 JSONPath 表示法。键中的每个点表示嵌套的 JSON 结构。例如，考虑以下结构：
{
"shape": {
"name": "rectangle",
"dimensions": {
"height": 24,
"width": 10
},
"color": {
"name": "red",
"code": "#FF0000"
}
}
}
要比较颜色代码的值，键将是 shape.color.code。
如果省略 filter 部分，查询将返回所有条目。
sort
sort 是一个有序的 key:order 对数组，其中：
key是状态存储中的键order是一个可选字符串，指示排序顺序："ASC"表示升序"DESC"表示降序
如果省略，默认为升序。
page
page 包含 limit 和 token 参数。
limit设置页面大小。token是组件返回的迭代令牌，用于后续查询。
在后台，此查询请求被转换为原生查询语言并由状态存储组件执行。
示例数据和查询
让我们看一些从简单到复杂的实际示例。
作为数据集，考虑一个包含员工 ID、组织、州和城市的 员工记录集合。请注意，该数据集是一个键/值对数组，其中：
key是唯一 IDvalue是包含员工记录的 JSON 对象。
为了更好地说明功能，组织名称 和员工 ID 是一个嵌套的 JSON person 对象。
首先创建一个 MongoDB 实例作为你的状态存储。
docker run -d --rm -p 27017:27017 --name mongodb mongo:5
接下来，启动一个 Dapr 应用程序。请参阅 组件配置文件，该文件指示 Dapr 使用 MongoDB 作为其状态存储。
dapr run --app-id demo --dapr-http-port 3500 --resources-path query-api-examples/components/mongodb
使用员工数据集填充状态存储，以便稍后查询。
curl -X POST -H "Content-Type: application/json" -d @query-api-examples/dataset.json http://localhost:3500/v1.0/state/statestore
填充完成后，你可以检查状态存储中的数据。在下图中，MongoDB UI 的一部分显示了员工记录。
每个条目都有 _id 成员作为连接的对象键，以及包含 JSON 记录的 value 成员。
查询 API 允许你从此 JSON 结构中选择记录。
现在你可以运行示例查询了。
示例 1
首先，查找加利福尼亚州的所有员工，并按员工 ID 降序排序。
这是 查询：
{
"filter": {
"EQ": { "state": "CA" }
},
"sort": [
{
"key": "person.id",
"order": "DESC"
}
]
}
此查询的 SQL 等效形式为：
SELECT * FROM c WHERE
state = "CA"
ORDER BY
person.id DESC
使用以下命令执行查询：
curl -s -X POST -H "Content-Type: application/json" -d @query-api-examples/query1.json http://localhost:3500/v1.0-alpha1/state/statestore/query | jq .
Invoke-RestMethod -Method Post -ContentType 'application/json' -InFile query-api-examples/query1.json -Uri 'http://localhost:3500/v1.0-alpha1/state/statestore/query'
查询结果是按请求顺序排列的匹配键/值对数组：
{
"results": [
{
"key": "3",
"data": {
"person": {
"org": "Finance",
"id": 1071
},
"city": "Sacramento",
"state": "CA"
},
"etag": "44723d41-deb1-4c23-940e-3e6896c3b6f7"
},
{
"key": "7",
"data": {
"city": "San Francisco",
"state": "CA",
"person": {
"id": 1015,
"org": "Dev Ops"
}
},
"etag": "0e69e69f-3dbc-423a-9db8-26767fcd2220"
},
{
"key": "5",
"data": {
"state": "CA",
"person": {
"org": "Hardware",
"id": 1007
},
"city": "Los Angeles"
},
"etag": "f87478fa-e5c5-4be0-afa5-f9f9d75713d8"
},
{
"key": "9",
"data": {
"person": {
"org": "Finance",
"id": 1002
},
"city": "San Diego",
"state": "CA"
},
"etag": "f5cf05cd-fb43-4154-a2ec-445c66d5f2f8"
}
]
}
示例 2
现在，查找来自 “Dev Ops” 和 “Hardware” 组织的所有员工。
这是 查询：
{
"filter": {
"IN": { "person.org": [ "Dev Ops", "Hardware" ] }
}
}
此查询的 SQL 等效形式为：
SELECT * FROM c WHERE
person.org IN ("Dev Ops", "Hardware")
使用以下命令执行查询：
curl -s -X POST -H "Content-Type: application/json" -d @query-api-examples/query2.json http://localhost:3500/v1.0-alpha1/state/statestore/query | jq .
Invoke-RestMethod -Method Post -ContentType 'application/json' -InFile query-api-examples/query2.json -Uri 'http://localhost:3500/v1.0-alpha1/state/statestore/query'
与上一个示例类似，结果是匹配键/值对的数组。
示例 3
在此示例中，查找：
- “Dev Ops” 部门的所有员工。
- 居住在华盛顿州和加利福尼亚州的 “Finance” 部门的员工。
此外，首先按州按字母降序排序，然后按员工 ID 升序排序。让我们一次处理最多 3 条记录。
这是 查询：
{
"filter": {
"OR": [
{
"EQ": { "person.org": "Dev Ops" }
},
{
"AND": [
{
"EQ": { "person.org": "Finance" }
},
{
"IN": { "state": [ "CA", "WA" ] }
}
]
}
]
},
"sort": [
{
"key": "state",
"order": "DESC"
},
{
"key": "person.id"
}
],
"page": {
"limit": 3
}
}
此查询的 SQL 等效形式为：
SELECT * FROM c WHERE
person.org = "Dev Ops" OR
(person.org = "Finance" AND state IN ("CA", "WA"))
ORDER BY
state DESC,
person.id ASC
LIMIT 3
使用以下命令执行查询：
curl -s -X POST -H "Content-Type: application/json" -d @query-api-examples/query3.json http://localhost:3500/v1.0-alpha1/state/statestore/query | jq .
Invoke-RestMethod -Method Post -ContentType 'application/json' -InFile query-api-examples/query3.json -Uri 'http://localhost:3500/v1.0-alpha1/state/statestore/query'
成功执行后，状态存储会返回一个 JSON 对象，其中包含匹配记录列表和分页令牌：
{
"results": [
{
"key": "1",
"data": {
"person": {
"org": "Dev Ops",
"id": 1036
},
"city": "Seattle",
"state": "WA"
},
"etag": "6f54ad94-dfb9-46f0-a371-e42d550adb7d"
},
{
"key": "4",
"data": {
"person": {
"org": "Dev Ops",
"id": 1042
},
"city": "Spokane",
"state": "WA"
},
"etag": "7415707b-82ce-44d0-bf15-6dc6305af3b1"
},
{
"key": "10",
"data": {
"person": {
"org": "Dev Ops",
"id": 1054
},
"city": "New York",
"state": "NY"
},
"etag": "26bbba88-9461-48d1-8a35-db07c374e5aa"
}
],
"token": "3"
}
分页令牌在 后续查询 中"按原样"使用，以获取下一批记录：
{
"filter": {
"OR": [
{
"EQ": { "person.org": "Dev Ops" }
},
{
"AND": [
{
"EQ": { "person.org": "Finance" }
},
{
"IN": { "state": [ "CA", "WA" ] }
}
]
}
]
},
"sort": [
{
"key": "state",
"order": "DESC"
},
{
"key": "person.id"
}
],
"page": {
"limit": 3,
"token": "3"
}
}
curl -s -X POST -H "Content-Type: application/json" -d @query-api-examples/query3-token.json http://localhost:3500/v1.0-alpha1/state/statestore/query | jq .
Invoke-RestMethod -Method Post -ContentType 'application/json' -InFile query-api-examples/query3-token.json -Uri 'http://localhost:3500/v1.0-alpha1/state/statestore/query'
此查询的结果为：
{
"results": [
{
"key": "9",
"data": {
"person": {
"org": "Finance",
"id": 1002
},
"city": "San Diego",
"state": "CA"
},
"etag": "f5cf05cd-fb43-4154-a2ec-445c66d5f2f8"
},
{
"key": "7",
"data": {
"city": "San Francisco",
"state": "CA",
"person": {
"id": 1015,
"org": "Dev Ops"
}
},
"etag": "0e69e69f-3dbc-423a-9db8-26767fcd2220"
},
{
"key": "3",
"data": {
"person": {
"org": "Finance",
"id": 1071
},
"city": "Sacramento",
"state": "CA"
},
"etag": "44723d41-deb1-4c23-940e-3e6896c3b6f7"
}
],
"token": "6"
}
通过这种方式，你可以在查询中更新分页令牌并遍历结果，直到不再返回记录。
限制
状态查询 API 具有以下限制：
- 要查询存储在状态存储中的 actor 状态，你需要使用特定数据库的查询 API。请参阅 查询 actor 状态。
- 该 API 不适用于 Dapr 加密状态存储功能。由于加密是由 Dapr 运行时完成的并作为加密数据存储，因此这实际上阻止了服务器端查询。
你可以在 相关链接 部分找到其他信息。
相关链接
- 请参阅 查询 API 参考。
- 查看 实现查询支持的状态存储组件。
- 查看 状态存储查询 API 实现指南。
- 了解如何 查询 Redis 状态存储。
4 - 操作指南：构建有状态服务
使用状态管理与扩展的复制服务
在本文中，你将学习如何使用可选的并发和一致性模型创建可水平扩展的有状态服务。使用状态管理 API 可以让开发者无需处理复杂的状态协调、冲突解决和故障处理。
设置状态存储
状态存储组件代表 Dapr 用于与数据库通信的资源。在本指南中，我们将使用默认的 Redis 状态存储。
使用 Dapr CLI
在自托管模式下运行 dapr init 时，Dapr 会创建一个默认的 Redis statestore.yaml，并在本地机器上运行一个 Redis 状态存储，位于：
- 在 Windows 上，位于
%UserProfile%\.dapr\components\statestore.yaml - 在 Linux/MacOS 上，位于
~/.dapr/components/statestore.yaml
使用 statestore.yaml 组件，你可以轻松地交换底层组件，而无需更改应用程序代码。
请参阅支持的状态存储列表。
Kubernetes
强一致性和最终一致性
使用强一致性时，Dapr 确保底层状态存储：
- 在数据写入所有副本后返回响应。
- 在写入或删除状态之前从法定数量获取 ACK。
对于 get 请求，Dapr 确保存储返回副本之间一致的最新数据。默认值为最终一致性，除非在状态 API 的请求中另有指定。
以下示例说明了如何使用强一致性保存、获取和删除状态。该示例使用 Python 编写，但适用于任何编程语言。
保存状态
import requests
import json
store_name = "redis-store" # name of the state store as specified in state store component yaml file
dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
stateReq = '[{ "key": "k1", "value": "Some Data", "options": { "consistency": "strong" }}]'
response = requests.post(dapr_state_url, json=stateReq)
获取状态
import requests
import json
store_name = "redis-store" # name of the state store as specified in state store component yaml file
dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
response = requests.get(dapr_state_url + "/key1", headers={"consistency":"strong"})
print(response.headers['ETag'])
删除状态
import requests
import json
store_name = "redis-store" # name of the state store as specified in state store component yaml file
dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
response = requests.delete(dapr_state_url + "/key1", headers={"consistency":"strong"})
如果未指定 concurrency 选项，则默认为最后写入并发模式。
先写优先和后写优先
Dapr 允许开发者在处理数据存储时选择两种常见的并发模式：
- 先写优先（First-write-wins）：适用于有多个应用程序实例同时写入同一键的情况。
- 后写优先（Last-write-wins）：Dapr 的默认模式。
Dapr 使用版本号来确定特定键是否已更新。你可以：
- 读取键的数据时保留版本号。
- 在更新（如写入和删除）期间使用版本号。
如果自获取版本号以来版本信息已更改，则会抛出错误，要求你执行另一次读取以获取最新版本信息和状态。
Dapr 利用 ETag 来确定状态的版本号。ETag 从状态请求中的 ETag 头返回。使用 ETag，你的应用程序可以通过在 ETag 不匹配时出错来知道资源自上次检查以来已更新。
以下示例说明如何：
- 获取 ETag。
- 使用 ETag 保存状态。
- 删除状态。
以下示例使用 Python 编写，但适用于任何编程语言。
import requests
import json
store_name = "redis-store" # name of the state store as specified in state store component yaml file
dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
response = requests.get(dapr_state_url + "/key1", headers={"concurrency":"first-write"})
etag = response.headers['ETag']
newState = '[{ "key": "k1", "value": "New Data", "etag": {}, "options": { "concurrency": "first-write" }}]'.format(etag)
requests.post(dapr_state_url, json=newState)
response = requests.delete(dapr_state_url + "/key1", headers={"If-Match": "{}".format(etag)})
处理版本不匹配失败
在以下示例中，你将看到当版本已更改时如何重试保存状态操作：
import requests
import json
# This method saves the state and returns false if failed to save state
def save_state(data):
try:
store_name = "redis-store" # name of the state store as specified in state store component yaml file
dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
response = requests.post(dapr_state_url, json=data)
if response.status_code == 200:
return True
except:
return False
return False
# This method gets the state and returns the response, with the ETag in the header -->
def get_state(key):
response = requests.get("http://localhost:3500/v1.0/state/<state_store_name>/{}".format(key), headers={"concurrency":"first-write"})
return response
# Exit when save state is successful. success will be False if there's an ETag mismatch -->
success = False
while success != True:
response = get_state("key1")
etag = response.headers['ETag']
newState = '[{ "key": "key1", "value": "New Data", "etag": {}, "options": { "concurrency": "first-write" }}]'.format(etag)
success = save_state(newState)
5 - How-To：启用事务性 Outbox 模式
在状态存储与发布订阅消息代理之间提交单个事务
事务性 outbox 模式是一种众所周知的设计模式，用于发送与应用程序状态变更相关的通知。事务性 outbox 模式使用一个横跨数据库与消息代理的单一事务来投递通知。
开发者在尝试自行实现此模式时面临着许多困难的技术挑战，通常需要编写容易出错的核心协调管理器，最多只能支持一两组数据库与消息代理的组合。
例如，你可以使用 outbox 模式：
- 向账户数据库写入新的用户记录。
- 发送一条表示账户已成功创建的通知消息。
借助 Dapr 的 outbox 支持，当调用 Dapr 的事务 API创建或更新应用程序状态时，你可以通知订阅者。
下图是从高层次概述 outbox 功能的工作原理：
- 服务 A 使用事务将状态保存/更新到状态存储。
- 在同一事务下向消息代理写入一条消息。当消息成功投递到消息代理后，事务完成，从而确保状态与消息作为一个整体完成事务。
- 消息代理将消息主题投递给任何订阅者 — 本例中为服务 B。
Outbox 在底层的工作原理
Dapr outbox 在两个流程中处理请求：用户请求流程与后台消息流程。二者共同确保状态与事件保持一致。
交互顺序如下：
应用程序调用 Dapr 状态管理 API，以事务方式写入状态。
这是业务数据（例如订单或资料更新）提交进行持久化的入口点。Dapr 向一个内部 outbox 主题发布一条带有唯一事务 ID 的意向消息。
这条持久化记录确保在任何数据库提交发生之前，事件意向就已经存在。状态与一个事务标记以原子方式写入同一个状态存储。
业务数据与标记在同一事务中提交，防止部分写入。事务提交后，应用程序收到成功响应。
此时应用程序可以继续执行，已知状态已保存，事件意向得到保证。后台订阅者读取意向消息。
当 outbox 启用时，Dapr 会启动消费者来处理内部 outbox 主题。订阅者在状态存储中验证事务标记。
此项检查确认数据库提交已成功，然后才会进行外部发布。验证后的业务事件被发布到外部发布订阅主题。
事件被发送到已配置的代理（Kafka、RabbitMQ 等），其他服务可以消费该事件。标记从状态存储中清理（删除）。
事件成功投递后，这可防止数据库无限制地增长。消息被确认并从内部主题中移除
如果发布或清理失败，Dapr 会重试，确保可靠的至少一次投递。
要求
outbox 功能需要 Dapr 支持的事务型状态存储。
了解更多关于你可以使用的事务方法。任何 Dapr 支持的发布订阅代理均可与 outbox 功能一起使用。
注意
建议使用支持竞争消费者模式的消息代理（例如 Apache Kafka），以降低重复事件的可能性。内部 outbox 主题
当启用 outbox 时，Dapr 会使用以下命名约定创建一个内部主题：{namespace}{appID}{topic}outbox，其中：namespace：Dapr 应用程序命名空间（如果已配置）appID：Dapr 应用程序标识符topic：在outboxPublishTopic元数据中指定的值
通过这种方式，每个 outbox 主题在每个应用程序和外部主题范围内被唯一标识，防止多租户环境中的路由冲突。
注意
确保主题已提前创建，或者 Dapr 在启动时拥有足够的权限来创建该主题。
启用 outbox 模式
要启用 outbox 功能，请在状态存储组件上添加以下必填和可选字段：
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: mysql-outbox
spec:
type: state.mysql
version: v1
metadata:
- name: connectionString
value: "<CONNECTION STRING>"
- name: outboxPublishPubsub # 必填
value: "mypubsub"
- name: outboxPublishTopic # 必填
value: "newOrder"
- name: outboxPubsub # 可选
value: "myOutboxPubsub"
- name: outboxDiscardWhenMissingState #可选。默认为 false
value: false
元数据字段
| 名称 | 必填 | 默认值 | 描述 |
|---|---|---|---|
| outboxPublishPubsub | 是 | N/A | 设置发布订阅组件的名称，用于在发布状态变更时投递通知 |
| outboxPublishTopic | 是 | N/A | 设置在用 outboxPublishPubsub 配置的发布订阅上接收状态变更的主题。消息正文将是针对 insert 或 update 操作的状态事务项 |
| outboxPubsub | 否 | outboxPublishPubsub | 设置 Dapr 用于协调状态与发布订阅事务的发布订阅组件。如果未设置，则使用用 outboxPublishPubsub 配置的发布订阅组件。如果你想将用于发送通知状态变更的发布订阅组件与用于协调事务的组件分开，此设置会很有用 |
| outboxDiscardWhenMissingState | 否 | false | 通过将 outboxDiscardWhenMissingState 设置为 true，如果 Dapr 无法在数据库中找到状态，它将丢弃该事务并且不再重试。如果状态存储数据在 Dapr 能够投递消息之前因任何原因被删除，并且你希望 Dapr 从发布订阅中删除这些项并停止重试以获取状态，则此设置会很有用 |
其他配置
在同一个状态存储上组合使用 outbox 与非 outbox 消息
如果你希望使用同一个状态存储来发送 outbox 与非 outbox 消息，只需定义两个连接到同一个状态存储的状态存储组件，其中一个启用 outbox 功能，另一个不启用。
不带 outbox 的 MySQL 状态存储
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: mysql
spec:
type: state.mysql
version: v1
metadata:
- name: connectionString
value: "<CONNECTION STRING>"
带 outbox 的 MySQL 状态存储
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: mysql-outbox
spec:
type: state.mysql
version: v1
metadata:
- name: connectionString
value: "<CONNECTION STRING>"
- name: outboxPublishPubsub # 必填
value: "mypubsub"
- name: outboxPublishTopic # 必填
value: "newOrder"
定制 outbox 模式消息
你可以通过设置另一个事务来覆盖发布到发布订阅代理的 outbox 模式消息，该事务不会被保存到数据库，并且被明确标记为投影（projection）。该事务会添加一个名为 outbox.projection 的元数据键，其值设置为 true。当添加到在事务中保存的状态数组时，写入状态时会忽略此负载，并将该数据用作发送到上游订阅者的负载。
若要正确使用，状态存储上的操作与消息投影之间的 key 值必须匹配。如果键不匹配，整个事务将失败。
如果你为同一个键启用了两个或更多 outbox.projection 状态项，则使用第一个定义的项，忽略其他项。
在以下状态事务的 Python SDK 示例中，值 "2" 被保存到数据库，而值 "3" 被发布到最终用户主题。
DAPR_STORE_NAME = "statestore"
async def main():
client = DaprClient()
client.execute_state_transaction(
store_name=DAPR_STORE_NAME,
operations=[
# 定义第一个状态操作以保存值 "2"
TransactionalStateOperation(
key='key1', data='2', metadata={'outbox.projection': 'false'}
),
# 定义第二个状态操作以发布带有元数据的值 "3"
TransactionalStateOperation(
key='key1', data='3', metadata={'outbox.projection': 'true'}
),
],
)
print("State transaction executed.")
通过将元数据项 "outbox.projection" 设置为 "true" 并确保 key 值匹配（key1）：
- 第一个操作被写入状态存储，不会向消息代理写入消息。
- 第二个操作值被发布到已配置的发布订阅主题。
在以下状态事务的 JavaScript SDK 示例中，值 "2" 被保存到数据库，而值 "3" 被发布到最终用户主题。
const { DaprClient, StateOperationType } = require('@dapr/dapr');
const DAPR_STORE_NAME = "statestore";
async function main() {
const client = new DaprClient();
// 定义第一个状态操作以保存值 "2"
const op1 = {
operation: StateOperationType.UPSERT,
request: {
key: "key1",
value: "2"
}
};
// 定义第二个状态操作以发布带有元数据的值 "3"
const op2 = {
operation: StateOperationType.UPSERT,
request: {
key: "key1",
value: "3",
metadata: {
"outbox.projection": "true"
}
}
};
// 创建状态操作列表
const ops = [op1, op2];
// 执行状态事务
await client.state.transaction(DAPR_STORE_NAME, ops);
console.log("State transaction executed.");
}
main().catch(err => {
console.error(err);
});
通过将元数据项 "outbox.projection" 设置为 "true" 并确保 key 值匹配（key1）：
- 第一个操作被写入状态存储，不会向消息代理写入消息。
- 第二个操作值被发布到已配置的发布订阅主题。
在以下状态事务的 .NET SDK 示例中，值 "2" 被保存到数据库，而值 "3" 被发布到最终用户主题。
public class Program
{
private const string DAPR_STORE_NAME = "statestore";
public static async Task Main(string[] args)
{
var client = new DaprClientBuilder().Build();
// 定义第一个状态操作以保存值 "2"
var op1 = new StateTransactionRequest(
key: "key1",
value: Encoding.UTF8.GetBytes("2"),
operationType: StateOperationType.Upsert
);
// 定义第二个状态操作以发布带有元数据的值 "3"
var metadata = new Dictionary<string, string>
{
{ "outbox.projection", "true" }
};
var op2 = new StateTransactionRequest(
key: "key1",
value: Encoding.UTF8.GetBytes("3"),
operationType: StateOperationType.Upsert,
metadata: metadata
);
// 创建状态操作列表
var ops = new List<StateTransactionRequest> { op1, op2 };
// 执行状态事务
await client.ExecuteStateTransactionAsync(DAPR_STORE_NAME, ops);
Console.WriteLine("State transaction executed.");
}
}
通过将元数据项 "outbox.projection" 设置为 "true" 并确保 key 值匹配（key1）：
- 第一个操作被写入状态存储，不会向消息代理写入消息。
- 第二个操作值被发布到已配置的发布订阅主题。
在以下状态事务的 Java SDK 示例中，值 "2" 被保存到数据库，而值 "3" 被发布到最终用户主题。
public class Main {
private static final String DAPR_STORE_NAME = "statestore";
public static void main(String[] args) {
try (DaprClient client = new DaprClientBuilder().build()) {
// 定义第一个状态操作以保存值 "2"
State<String> state1 = new State<>(
"key1",
"2",
null, // etag
null // 并发与一致性选项
);
// 定义第二个状态操作以发布带有元数据的值 "3"
Map<String, String> metadata = new HashMap<>();
metadata.put("outbox.projection", "true");
State<String> state2 = new State<>(
"key1",
"3",
null, // etag
metadata,
null // 并发与一致性选项
);
TransactionalStateOperation<String> op1 = new TransactionalStateOperation<>(
TransactionalStateOperation.OperationType.UPSERT, state1
);
TransactionalStateOperation<String> op2 = new TransactionalStateOperation<>(
TransactionalStateOperation.OperationType.UPSERT, state2
);
// 创建事务状态操作列表
List<TransactionalStateOperation<?>> ops = new ArrayList<>();
ops.add(op1);
ops.add(op2);
// 配置事务请求，设置状态存储
ExecuteStateTransactionRequest transactionRequest = new ExecuteStateTransactionRequest(DAPR_STORE_NAME);
transactionRequest.setOperations(ops);
// 执行状态事务
client.executeStateTransaction(transactionRequest).block();
System.out.println("State transaction executed.");
} catch (Exception e) {
e.printStackTrace();
}
}
}
通过将元数据项 "outbox.projection" 设置为 "true" 并确保 key 值匹配（key1）：
- 第一个操作被写入状态存储，不会向消息代理写入消息。
- 第二个操作值被发布到已配置的发布订阅主题。
在以下状态事务的 Go SDK 示例中，值 "2" 被保存到数据库，而值 "3" 被发布到最终用户主题。
ops := make([]*dapr.StateOperation, 0)
op1 := &dapr.StateOperation{
Type: dapr.StateOperationTypeUpsert,
Item: &dapr.SetStateItem{
Key: "key1",
Value: []byte("2"),
},
}
op2 := &dapr.StateOperation{
Type: dapr.StateOperationTypeUpsert,
Item: &dapr.SetStateItem{
Key: "key1",
Value: []byte("3"),
// 覆盖保存到数据库的数据负载
Metadata: map[string]string{
"outbox.projection": "true",
},
},
}
ops = append(ops, op1, op2)
meta := map[string]string{}
err := testClient.ExecuteStateTransaction(ctx, store, meta, ops)
通过将元数据项 "outbox.projection" 设置为 "true" 并确保 key 值匹配（key1）：
- 第一个操作被写入状态存储，不会向消息代理写入消息。
- 第二个操作值被发布到已配置的发布订阅主题。
你可以使用以下 HTTP 请求传递消息覆盖：
curl -X POST http://localhost:3500/v1.0/state/starwars/transaction \
-H "Content-Type: application/json" \
-d '{
"operations": [
{
"operation": "upsert",
"request": {
"key": "order1",
"value": {
"orderId": "7hf8374s",
"type": "book",
"name": "The name of the wind"
}
}
},
{
"operation": "upsert",
"request": {
"key": "order1",
"value": {
"orderId": "7hf8374s"
},
"metadata": {
"outbox.projection": "true"
},
"contentType": "application/json"
}
}
]
}'
通过将元数据项 "outbox.projection" 设置为 "true" 并确保 key 值匹配（key1）：
- 第一个操作被写入状态存储，不会向消息代理写入消息。
- 第二个操作值被发布到已配置的发布订阅主题。
覆盖 Dapr 生成的 CloudEvent 字段
你可以使用自定义 CloudEvent 元数据来覆盖已发布的 outbox 事件上的 Dapr 生成的 CloudEvent 字段。
async def execute_state_transaction():
async with DaprClient() as client:
# 定义状态操作
ops = []
op1 = {
'operation': 'upsert',
'request': {
'key': 'key1',
'value': b'2', # 将字符串转换为字节数组
'metadata': {
'cloudevent.id': 'unique-business-process-id',
'cloudevent.source': 'CustomersApp',
'cloudevent.type': 'CustomerCreated',
'cloudevent.subject': '123',
'my-custom-ce-field': 'abc'
}
}
}
ops.append(op1)
# 执行状态事务
store_name = 'your-state-store-name'
try:
await client.execute_state_transaction(store_name, ops)
print('State transaction executed.')
except Exception as e:
print('Error executing state transaction:', e)
# 运行异步函数
if __name__ == "__main__":
asyncio.run(execute_state_transaction())
const { DaprClient } = require('dapr-client');
async function executeStateTransaction() {
// 初始化 Dapr 客户端
const daprClient = new DaprClient();
// 定义状态操作
const ops = [];
const op1 = {
operationType: 'upsert',
request: {
key: 'key1',
value: Buffer.from('2'),
metadata: {
'id': 'unique-business-process-id',
'source': 'CustomersApp',
'type': 'CustomerCreated',
'subject': '123',
'my-custom-ce-field': 'abc'
}
}
};
ops.push(op1);
// 执行状态事务
const storeName = 'your-state-store-name';
const metadata = {};
}
executeStateTransaction();
public class StateOperationExample
{
public async Task ExecuteStateTransactionAsync()
{
var daprClient = new DaprClientBuilder().Build();
// 将值 "2" 定义为字符串并将其序列化为字节数组
var value = "2";
var valueBytes = JsonSerializer.SerializeToUtf8Bytes(value);
// 定义第一个状态操作以保存带有元数据的值 "2"
// 覆盖 Cloudevent 元数据
var metadata = new Dictionary<string, string>
{
{ "cloudevent.id", "unique-business-process-id" },
{ "cloudevent.source", "CustomersApp" },
{ "cloudevent.type", "CustomerCreated" },
{ "cloudevent.subject", "123" },
{ "my-custom-ce-field", "abc" }
};
var op1 = new StateTransactionRequest(
key: "key1",
value: valueBytes,
operationType: StateOperationType.Upsert,
metadata: metadata
);
// 创建状态操作列表
var ops = new List<StateTransactionRequest> { op1 };
// 执行状态事务
var storeName = "your-state-store-name";
await daprClient.ExecuteStateTransactionAsync(storeName, ops);
Console.WriteLine("State transaction executed.");
}
public static async Task Main(string[] args)
{
var example = new StateOperationExample();
await example.ExecuteStateTransactionAsync();
}
}
public class StateOperationExample {
public static void main(String[] args) {
executeStateTransaction();
}
public static void executeStateTransaction() {
// 构建 Dapr 客户端
try (DaprClient daprClient = new DaprClientBuilder().build()) {
// 覆盖 CloudEvent 元数据
Map<String, String> metadata = new HashMap<>();
metadata.put("cloudevent.id", "unique-business-process-id");
metadata.put("cloudevent.source", "CustomersApp");
metadata.put("cloudevent.type", "CustomerCreated");
metadata.put("cloudevent.subject", "123");
metadata.put("my-custom-ce-field", "abc");
State<String> state = new State<>(
"key1", // 定义键 "key1"
"value1", // 定义值 "value1"
null, // etag
metadata,
null // 并发与一致性选项
);
// 定义状态操作
List<TransactionalStateOperation<?>> ops = new ArrayList<>();
TransactionalStateOperation<String> op1 = new TransactionalStateOperation<>(
TransactionalStateOperation.OperationType.UPSERT,
state
);
ops.add(op1);
// 执行状态事务
String storeName = "your-state-store-name";
daprClient.executeStateTransaction(storeName, ops).block();
System.out.println("State transaction executed.");
} catch (Exception e) {
e.printStackTrace();
}
}
}
func main() {
// 创建 Dapr 客户端
client, err := dapr.NewClient()
if err != nil {
log.Fatalf("failed to create Dapr client: %v", err)
}
defer client.Close()
ctx := context.Background()
store := "your-state-store-name"
// 定义状态操作
ops := make([]*dapr.StateOperation, 0)
op1 := &dapr.StateOperation{
Type: dapr.StateOperationTypeUpsert,
Item: &dapr.SetStateItem{
Key: "key1",
Value: []byte("2"),
// 覆盖 Cloudevent 元数据
Metadata: map[string]string{
"cloudevent.id": "unique-business-process-id",
"cloudevent.source": "CustomersApp",
"cloudevent.type": "CustomerCreated",
"cloudevent.subject": "123",
"my-custom-ce-field": "abc",
},
},
}
ops = append(ops, op1)
// 事务的元数据（如果有）
meta := map[string]string{}
// 执行状态事务
err = client.ExecuteStateTransaction(ctx, store, meta, ops)
if err != nil {
log.Fatalf("failed to execute state transaction: %v", err)
}
log.Println("State transaction executed.")
}
curl -X POST http://localhost:3500/v1.0/state/starwars/transaction \
-H "Content-Type: application/json" \
-d '{
"operations": [
{
"operation": "upsert",
"request": {
"key": "key1",
"value": "2"
}
},
],
"metadata": {
"id": "unique-business-process-id",
"source": "CustomersApp",
"type": "CustomerCreated",
"subject": "123",
"my-custom-ce-field": "abc",
}
}'
注意
data CloudEvent 字段保留仅供 Dapr 使用，不可自定义。演示
后续步骤
6 - 操作指南：在应用程序之间共享状态
了解在不同应用程序之间共享状态的策略
Dapr 提供了多种在应用程序之间共享状态的方式。
不同的架构在共享状态方面可能有不同的需求。在一种场景中，你可能希望：
- 将所有状态封装在给定应用程序内
- 让 Dapr 为你管理访问权限
在另一种场景中，你可能需要两个应用程序同时操作相同的状态，获取和保存相同的键。
为启用状态共享，Dapr 支持以下键前缀策略：
| 键前缀 | 描述 |
|---|---|
appid | 默认策略，允许你仅通过具有指定 appid 的应用程序管理状态。所有状态键都将以 appid 为前缀，并限定在该应用程序范围内。 |
name | 使用状态存储组件的名称作为前缀。多个应用程序可以共享给定状态存储的相同状态。 |
namespace | 如果设置，此设置会将 appid 键加上配置的命名空间作为前缀，从而得到一个限定在给定命名空间范围内的键。这允许具有相同 appid 但在不同命名空间中的应用重用相同的状态存储。如果未配置命名空间，该设置会回退到 appid 策略。有关 Dapr 中命名空间的更多信息，请参阅操作指南：将组件限定到一个或多个应用程序 |
none | 不使用前缀。多个应用程序在不同状态存储之间共享状态。 |
指定状态前缀策略
要指定前缀策略，请在状态组件上添加名为 keyPrefix 的元数据键：
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
namespace: production
spec:
type: state.redis
version: v1
metadata:
- name: keyPrefix
value: <key-prefix-strategy>
示例
以下示例演示了使用每种支持的前缀策略进行状态检索的情况。
appid（默认）
在下面的示例中，应用程序 ID 为 myApp 的 Dapr 应用程序正在将状态保存到名为 redis 的状态存储中：
curl -X POST http://localhost:3500/v1.0/state/redis \
-H "Content-Type: application/json"
-d '[
{
"key": "darth",
"value": "nihilus"
}
]'
该键将保存为 myApp||darth。
namespace
在命名空间 production 中运行、应用程序 ID 为 myApp 的 Dapr 应用程序正在将状态保存到名为 redis 的状态存储中：
curl -X POST http://localhost:3500/v1.0/state/redis \
-H "Content-Type: application/json"
-d '[
{
"key": "darth",
"value": "nihilus"
}
]'
该键将保存为 production.myApp||darth。
name
在下面的示例中，应用程序 ID 为 myApp 的 Dapr 应用程序正在将状态保存到名为 redis 的状态存储中：
curl -X POST http://localhost:3500/v1.0/state/redis \
-H "Content-Type: application/json"
-d '[
{
"key": "darth",
"value": "nihilus"
}
]'
该键将保存为 redis||darth。
none
在下面的示例中，应用程序 ID 为 myApp 的 Dapr 应用程序正在将状态保存到名为 redis 的状态存储中：
curl -X POST http://localhost:3500/v1.0/state/redis \
-H "Content-Type: application/json"
-d '[
{
"key": "darth",
"value": "nihilus"
}
]'
该键将保存为 darth。
7 - 操作指南：加密应用状态
自动加密状态并管理密钥轮换
对应用状态进行静态加密，以在企业工作负载或受监管环境中提供更强的安全性。Dapr 基于 Galois/Counter Mode (GCM) 中的 AES 提供自动客户端加密，支持 128、192 和 256 位密钥。
除了自动加密外，Dapr 还支持主加密密钥和辅助加密密钥，使开发人员和运维团队更容易启用密钥轮换策略。所有 Dapr 状态存储都支持此功能。
加密密钥始终从机密中获取，不能在 metadata 部分以纯文本值形式提供。
启用自动加密
将以下 metadata 部分添加到任何 Dapr 支持的状态存储：
metadata:
- name: primaryEncryptionKey
secretKeyRef:
name: mysecret
key: mykey # key 是可选的。
例如，这是一个 Redis 加密状态存储的完整 YAML：
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: primaryEncryptionKey
secretKeyRef:
name: mysecret
key: mykey
现在您已经配置了一个 Dapr 状态存储，可以从名为 mysecret 的机密中获取加密密钥，该机密在名为 mykey 的键中包含实际的加密密钥。
实际的加密密钥必须是有效的十六进制编码的加密密钥。虽然支持 192 位和 256 位密钥，但建议您使用 128 位加密密钥。如果加密密钥无效，Dapr 会报错并退出。
例如，您可以使用以下命令生成随机的十六进制编码的 128 位（16 字节）密钥：
openssl rand 16 | hexdump -v -e '/1 "%02x"'
# 结果将类似于 "cb321007ad11a9d23f963bff600d58e0"
请注意，机密存储不必支持键。
密钥轮换
为了支持密钥轮换，Dapr 提供了一种指定辅助加密密钥的方法：
metadata:
- name: primaryEncryptionKey
secretKeyRef:
name: mysecret
key: mykey
- name: secondaryEncryptionKey
secretKeyRef:
name: mysecret2
key: mykey2
当 Dapr 启动时，它会获取包含 metadata 部分中列出的加密密钥的机密。Dapr 自动知道哪个状态项已使用哪个密钥加密，因为它将 secretKeyRef.name 字段附加到实际状态键的末尾。
要轮换密钥：
- 将
primaryEncryptionKey更改为指向包含您新密钥的机密。 - 将旧的主加密密钥移动到
secondaryEncryptionKey。
新数据将使用新密钥加密，而任何检索到的旧数据将使用辅助密钥解密。
对使用旧密钥加密的数据项的任何更新都将使用新密钥重新加密。
注意
当您轮换密钥时，除非您的应用程序再次写入，否则使用旧密钥加密的数据不会自动重新加密。如果您删除轮换的密钥（现在的辅助加密密钥），您将无法访问使用该密钥加密的数据。相关链接
8.1 - Azure Cosmos DB
使用 Azure Cosmos DB 作为状态存储
Dapr 在保存和检索状态时不会转换状态值。Dapr 要求所有状态存储实现遵守特定的键格式方案（参见[状态管理规范](https://docs.dapr.io/zh-hans/reference/api/state_api/）。您可以直接与底层存储交互来操作状态数据，例如：
- 查询状态。
- 创建聚合视图。
- 制作备份。
注意
Azure Cosmos DB 是一个支持多种 API 的多模态数据库。默认的 Dapr Cosmos DB 状态存储实现使用 Azure Cosmos DB SQL API。连接到 Azure Cosmos DB
要连接到您的 Cosmos DB 实例，您可以：
- 使用 Azure 管理门户上的数据资源管理器。
- 使用各种 SDK 和工具。
注意
当您为 Dapr 配置 Azure Cosmos DB 时，请指定要使用的确切数据库和集合。以下 Cosmos DB SQL API 示例假设您已经连接到正确的数据库和一个名为 “states” 的集合。按 App ID 列出键
要获取与应用程序 “myapp” 关联的所有状态键，请使用以下查询：
SELECT * FROM states WHERE CONTAINS(states.id, 'myapp||')
上述查询返回 id 包含 “myapp-” 的所有文档，这是状态键的前缀。
获取特定状态数据
要获取应用程序 “myapp” 中键为 “balance” 的状态数据，请使用以下查询：
SELECT * FROM states WHERE states.id = 'myapp||balance'
读取返回文档中的 value 字段。要获取状态版本/ETag，请使用以下命令：
SELECT states._etag FROM states WHERE states.id = 'myapp||balance'
读取 Actor 状态
要获取应用程序 ID 为 “mypets” 中 Actor 类型为 “cat” 且实例 ID 为 “leroy” 的 Actor 关联的所有状态键，请使用以下命令：
SELECT * FROM states WHERE CONTAINS(states.id, 'mypets||cat||leroy||')
而要获取特定的 Actor 状态（如 “food”），请使用以下命令：
SELECT * FROM states WHERE states.id = 'mypets||cat||leroy||food'
警告
不应手动更新或删除存储中的状态。所有写入和删除操作应通过 Dapr 运行时完成。唯一的例外： 通常需要删除状态存储中的 Actor 记录，一旦您知道这些记录已不再使用，以防止未使用的 Actor 实例堆积而可能永远无法再次加载。8.2 - Redis
使用 Redis 作为状态存储
Dapr 在保存和检索状态时不会转换状态值。Dapr 要求所有状态存储实现都遵循一定的键格式方案（参见状态管理规范）。您可以直接与底层存储交互来操作状态数据，例如：
- 查询状态。
- 创建聚合视图。
- 进行备份。
Dapr 提供了一个 Redis state store component 作为可插拔组件。
注意
以下示例使用 Redis CLI 针对使用默认 Dapr 状态存储实现的 Redis 存储进行操作。连接到 Redis
您可以使用官方的 redis-cli 或任何其他兼容 Redis 的工具连接到 Redis 状态存储，直接查询 Dapr 状态。如果您是在容器中运行 Redis，使用 redis-cli 的最简单方法是通过容器：
docker run --rm -it --link <Redis 容器的名称> redis redis-cli -h <Redis 容器的名称>
按应用 ID 列出键
要获取与应用 “myapp” 关联的所有状态键，请使用命令：
KEYS myapp*
上述命令返回现有键的列表，例如：
1) "myapp||balance"
2) "myapp||amount"
获取特定状态数据
Dapr 将状态值保存为哈希值。每个哈希值包含一个 “data” 字段，其中包含：
- 状态数据。
- 一个 “version” 字段，其中包含一个持续递增的版本，用作 ETag。
例如，要获取应用 “myapp” 的键 “balance” 对应的状态数据，请使用命令：
HGET myapp||balance data
要获取状态版本/ETag，请使用命令：
HGET myapp||balance version
读取 Actor 状态
要获取与应用 ID “mypets” 中 actor 类型为 “cat” 且实例 ID 为 “leroy” 的 actor 关联的所有状态键，请使用命令：
KEYS mypets||cat||leroy*
要获取特定的 actor 状态（如 “food”），请使用命令：
HGET mypets||cat||leroy||food value
警告
您不应手动更新或删除存储中的状态。所有写入和删除操作都应通过 Dapr 运行时完成。**唯一例外：**通常需要删除状态存储中的 actor 记录，一旦您知道这些 actor 不再使用，以防止可能永远不会再加载的未使用 actor 实例积累。8.3 - SQL Server
使用 SQL Server 作为后端状态存储
Dapr 在保存和检索状态时不会转换状态值。Dapr 要求所有状态存储实现都遵守一定的键格式方案（请参阅状态管理规范）。你可以直接与底层存储交互来操作状态数据，例如：
- 查询状态。
- 创建聚合视图。
- 进行备份。
连接到 SQL Server
连接 SQL Server 实例最简单的方式是使用：
- Azure Data Studio（Windows、macOS、Linux）
- SQL Server Management Studio（Windows）
注意
为 Dapr 配置 Azure SQL 数据库时，需要指定要使用的确切表名。以下 Azure SQL 示例假设你已经连接到正确的数据库，且表中有一个名为 “states” 的表。按 App ID 列出键
要获取与应用程序 “myapp” 关联的所有状态键，请使用以下查询：
SELECT * FROM states WHERE [Key] LIKE 'myapp||%'
上述查询返回所有 ID 包含 “myapp||” 的行，这是状态键的前缀。
获取特定状态数据
要获取应用程序 “myapp” 中键为 “balance” 的状态数据，请使用以下查询：
SELECT * FROM states WHERE [Key] = 'myapp||balance'
读取返回行的 Data 字段。要获取状态版本/ETag，请使用以下命令：
SELECT [RowVersion] FROM states WHERE [Key] = 'myapp||balance'
获取过滤后的状态数据
要获取 JSON 数据中 “color” 值等于 “blue” 的所有状态数据，请使用以下查询：
SELECT * FROM states WHERE JSON_VALUE([Data], '$.color') = 'blue'
读取 Actor 状态
要获取属于应用程序 ID 为 “mypets”、Actor 类型为 “cat”、实例 ID 为 “leroy” 的所有 Actor 状态键，请使用以下命令：
SELECT * FROM states WHERE [Key] LIKE 'mypets||cat||leroy||%'
要获取特定的 Actor 状态（如 “food”），请使用以下命令：
SELECT * FROM states WHERE [Key] = 'mypets||cat||leroy||food'
警告
你不应该手动更新或删除存储中的状态。所有写入和删除操作都应通过 Dapr 运行时完成。**唯一例外：**通常需要删除状态存储中的 Actor 记录，一旦你确定这些记录不再使用，以防止未使用的 Actor 实例堆积，而这些实例可能永远不会再次被加载。9 - 状态生存时间 (TTL)
使用 TTL 管理状态。
Dapr 支持为每个状态设置请求级的生存时间（TTL）。这意味着应用程序可以为每个存储的状态设置 TTL，过期后将无法检索这些状态。
对于支持的 状态存储，您只需在发布消息时设置 ttlInSeconds 元数据。其他 状态存储 将忽略此值。对于某些 状态存储，您可以按表/容器指定默认过期时间。
原生状态 TTL 支持
当 状态存储 组件原生支持状态 TTL 时，Dapr 直接转发 TTL 配置，不添加任何额外逻辑，保持可预测的行为。当组件以不同方式处理过期状态时，这非常有用。
未指定 TTL 时，将保留 状态存储 的默认行为。
显式持久化绕过全局定义的 TTL
持久化状态适用于所有允许为所有数据指定默认 TTL 的 状态存储，无论是：
- 通过 Dapr 组件设置全局 TTL 值，或
- 在 Dapr 外部创建 状态存储 并设置全局 TTL 值。
未指定特定 TTL 时，数据将在该全局 TTL 时间段后过期。这不由 Dapr 处理。
此外，所有 状态存储 还支持_显式_持久化数据的选项。这意味着您可以忽略默认数据库策略（可能在 Dapr 外部设置或通过 Dapr 组件设置）来无限期保留给定的数据库记录。您可以通过将 ttlInSeconds 设置为 -1 值来实现此目的。该值表示忽略任何设置的 TTL 值。
支持的组件
请参阅 状态存储 组件指南中的 TTL 列。
示例
您可以在 状态存储 set 请求的元数据中设置状态 TTL：
#dependencies
from dapr.clients import DaprClient
#code
DAPR_STORE_NAME = "statestore"
with DaprClient() as client:
client.save_state(DAPR_STORE_NAME, "order_1", str(orderId), state_metadata={
'ttlInSeconds': '120'
})
要启动 Dapr 边车 并运行上述示例应用程序，您需要运行类似于以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 -- python3 OrderProcessingService.py
// dependencies
using Dapr.Client;
// code
await client.SaveStateAsync(storeName, stateKeyName, state, metadata: new Dictionary<string, string>() {
{
"ttlInSeconds", "120"
}
});
要启动 Dapr 边车 并运行上述示例应用程序，您需要运行类似于以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 dotnet run
// dependencies
import (
dapr "github.com/dapr/go-sdk/client"
)
// code
md := map[string]string{"ttlInSeconds": "120"}
if err := client.SaveState(ctx, store, "key1", []byte("hello world"), md); err != nil {
panic(err)
}
要启动 Dapr 边车 并运行上述示例应用程序，您需要运行类似于以下的命令：
dapr run --app-id orderprocessing --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 go run .
curl -X POST -H "Content-Type: application/json" -d '[{ "key": "order_1", "value": "250", "metadata": { "ttlInSeconds": "120" } }]' http://localhost:3601/v1.0/state/statestore
Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '[{"key": "order_1", "value": "250", "metadata": {"ttlInSeconds": "120"}}]' -Uri 'http://localhost:3601/v1.0/state/statestore'
相关链接
- 查看状态 API 参考指南。
- 了解如何使用键值对持久化状态。
- 状态存储 组件列表。
- 阅读API 参考。