[Java知识库]Dapr for dotnet | 状态管理-State management

什么是应用的 “有状态” 和 “无状态” ？

当用户登录时，将 session 或者 token 传给应用服务器管理，应用服务器里持有用户的上下文信息，并且给用户一个 cookie 值，记录对应的 session（或 用户 id 值，记录对应的 token）。然后下次请求，用户携带 cookie 值来，我们就能识别到对应 session，从而找到用户的信息。这时应用服务器是 “有状态” 的。

同样用户登陆时，我们将 session 或 token 存储在第三方的一些服务或者中间件上，比如存储在 redis 上。此时应用服务器不保存上下文信息，只负责对用户的每次请求进行处理，然后返回处理的结果即可，这时应用服务器是“无状态” 的。

应用的 “有状态” 和 “无状态” 的优缺点

有状态服务

【定义】服务端需要记录每次会话的客户端信息，从而识别客户端身份，根据用户身份信息进行请求的处理，并响应相应的处理结果。

【有状态服务的优点】

• 当状态是共享的跨调用时，开发是容易的；
• 不需要额外的持久存储；
• 通常，为低延时优化；

【有状态服务的缺点】

• 服务端保存大量数据，增加服务端压力；
• 伸缩扩展复杂，服务端保存用户状态，无法进行水平扩展；
• 客户端请求依赖服务端，多次请求必须访问同一个服务（或同一台服务器）；

无状态服务

【定义】客户端的每次请求必须具备自描述信息，服务端通过这些信息来识别客户端身份。服务端不保存任何客户端请求者信息。

【无状态服务的优点】

• 服务间数据不需要同步；
• 动态伸缩，快速扩容；
• 持久化存储灵活，热备冷备切换容易；
• 容易水平扩展，服务实现负载均衡；

【无状态服务的缺点】

• 依赖额外的持久化存储；
• 客户端请求不依赖服务端的信息，任何多次请求不需要必须访问到同一个服务（或同一台服务器）；
• 服务端的集群和状态对客户端透明，服务端可以任意的迁移和伸缩，减小服务端的存储压力；

状态化的判断指标

状态化的判断是指两个来自相同发起者的请求在服务器端是否具备上下文关系。

• 如果是状态化请求，那么服务器端一般都要保存请求的相关信息，每个请求可以默认地使用以前的请求信息。
• 而无状态的请求，服务器端的处理信息必须全部来自于请求所携带的信息以及可以被所有请求所使用的公共信息。

关于应用的 “有/无状态” 参考以下文章：

• 分布式系统中的“无状态”和“有状态”详解
• 构建可伸缩的有状态服务

Dapr 状态管理介绍

在分布式应用环境中，应用程序都是由独立进程的服务组成。 虽然每个服务都应是无状态的，但某些服务必须跟踪状态才能完成业务操作。

【案例场景】请考虑电子商务网站的购物篮服务。 如果服务无法跟踪状态，则客户可能因为离开网站丢失购物篮内容，从而导致供应商或公司的销售损失和顾客购物的不愉快体验。 对于这些情况，需要将状态持久保存在分布式状态存储中。 Dapr 状态管理构建基块简化了状态跟踪，并跨各种数据存储提供高级功能。

使用 Dapr 的状态管理，应用程序可以将数据作为 键/值对 存储在 受支持的状态存储 中并进行查询。这 使您能够构建有状态的、长时间运行的应用程序，这些应用程序可以保存和检索其状态，例如购物车、游戏、即时聊天工具的会话状态。

Dapr 状态管理工作模型

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

Dapr 状态管理具备的能力

在分布式应用程序中的有状态化跟踪，可能面临如下挑战：

• 应用程序可能需要不同类型的数据存储；
• 访问和更新数据可能需要不同的一致性级别；
• 多个用户可以同时更新数据，这需要解决并发冲突；
• 服务必须重试与数据存储交互时发生的任何短期、暂时性错误；

Dapr 状态管理构建基块解决了这些难题。 它简化了跟踪状态，没有依赖关系或第三方存储 SDK 学习曲线。

• 可配置的状态存储行为，设置 并发控制（ETag） 和 数据一致性（最终一致、强一致性） 选项；
• 执行 CRUD，包括 批量更新操作，多个事务操作；
• 查询和筛选 键/值对（key/value） 数据；
• 可插拔状态存储，Dapr 数据存储被建模为组件，可以在不修改你的服务代码的情况下进行替换；

数据一致性（CAP）原则

CAP 定理是一组适用于存储状态的分布式系统的原则，包含了以下三个属性。

该定理指出，分布式数据系统将在一致性、可用性和分区容错之间做出权衡。而且，任何数据存储只能保证三个属性中的两个（三者不可兼得）：

• 【一致性（C ）】 群集中的每个节点都会使用最新数据进行响应（即使系统必须阻止请求），直到所有副本都更新。 如果你向“一致性系统”查询当前正在更新的项，直到所有副本都成功更新，才会获得响应。 不过，你将始终收到最新的数据。
• 【可用性（A）】每个节点都会返回即时响应，即使该响应不是最新数据。 如果你向“可用系统”查询正在更新的项，将获得服务此时可以提供的最佳答案。
• 【分区容错（P）】保证系统继续运行，即使复制的数据节点发生故障或者与其他复制的数据节点断开连接。

分布式应用程序必须处理 P 属性。 当服务与网络调用相互通信时，会发生网络中断 P。 因此，分布式应用程序必须是 AP 或 CP。

Dapr 状态管理的一致性模式

当使用 强一致性 时，Dapr 会等待所有副本（或指定的quorums）确认后才会确认写入请求。

§ CP 应用程序选择 一致性，而不选择可用性 。 Dapr 通过其 强一致性 策略支持此选择。 在此方案中，状态存储将在完成写入请求前必须同步更新所有副本（或者，在某些情况下，达到仲裁）。 读取操作将跨副本一致地返回最新数据。

当使用 最终一致性 时，Dapr 将在基本数据存储接受写入请求后立即返回，即使这是单个副本。

§ AP 应用程序选择 可用性 ，而不选择一致性 。 Dapr 通过 最终一致性 策略支持此选择。 请考虑基础数据存储（例如 Azure CosmosDB），它在多个副本上存储冗余数据。 借助最终一致性，状态存储会将更新写入副本，并完成客户端的写入请求。 之后，存储将异步更新其他副本。 读取请求可以从任何副本返回数据（包括尚未收到最新更新的副本）。

查看更多 Dapr 状态存储组件支持的完整度

• Reference / Component specs / State stores

注意：由 Dapr 状态存储组件完成附加到该操作的一致性提示。 并非所有数据存储都支持这两种一致性级别。 如果未设置一致性提示，则默认行为是最终一致性。

并发性（Concurrency）

在了解 Dapr 的状态管理前，我们先了解下并发控制的方案有哪些？

数据库中的并发控制

在数据库中，【并发控制】是指在多个用户/进程/线程同时对数据库进行操作时，如何保证事务的一致性和隔离性并保障并发程度的最大化。

并发冲突场景分析

• 读-读（read-read）：不存在任何问题；
• 读-写（read-write）：有隔离性问题，可能遇到脏读，幻读，不可重复读；
• 写-写（write-write）：有数据更新丢失，脏写问题；

名词解释

并发冲突的解决方案

1.乐观的并发控制（乐观锁）

• OCC
• TS

是一种用来解决【写-写】冲突的无锁并发控制，认为事务间争用没有那么多，所以先进行修改，在提交事务前，检查一下事务开始后，有没有新提交改变，如果没有就提交，如果有就放弃并重试。
乐观并发控制类似自选锁。乐观并发控制适用于低数据争用，写冲突比较少的环境。

2.悲观的并发控制（悲观锁）

• 2PL

基于锁（lock）的并发控制，这种方式开销比较高，而且无法避免死锁问题。

3.多版本并发控制（MVCC）

是一种用来解决【读-写】冲突的无锁并发控制，也就是为事务分配单向增长的时间戳：

• 为每个修改保存一个版本，版本与事务时间戳关联，
• 读操作只读该事务开始前的数据库的快照。

这样在读操作不用阻塞写操作，写操作不用阻塞读操作的同时，避免了脏读和不可重复读，但不能解决【写-写】冲突。

Dapr 状态管理的并发控制（ETag）

Dapr 支持使用 ETag 的乐观并发控制 / 乐观锁（OCC，Optimistic Concurrency Control）。

• 🐹 当一个发送请求操作状态时，Dapr 会给返回的状态附加一个ETag 属性。
• 🐹 当用户代码试图更新或删除一个状态时，它应该通过更新的请求体或删除的 If-Match 头附加的 ETag 属性。

🦀🦀🦀 只有当提供的 ETag 属性与状态存储中的 ETag 属性匹配时，写操作才能成功。

Dapr 之所以选择 OCC，是因为在不少应用中，数据更新冲突都是很少的，因为客户端是按业务上下文自然分割的，可以对不同的数据进行操作。 然而，如果你的应用选择使用 ETag，请求可能会因为不匹配的 ETag 而被拒绝。 建议您在使用 ETag 时，使用重试策略来补偿这种冲突。

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

【ETag 两种模式的区别】

• 🦂first-write-wins（最先写赢）：应用程序在写入请求时附带 ETag 属性；
• 🦂last-write-wins（最后写赢）：应用程序在写入请求时省略 ETag 属性；

first-write-wins 在您有多个应用程序实例的情况下很有用，所有实例都同时写入同一个键。Dapr 状态管理的默认模式是 last-write-wins。

注意：对于原生不支持 ETag 的存储引擎，要求相应的 Dapr 状态存储实现能够模拟ETag，并在处理状态时遵循 Dapr 状态管理 API 规范。 由于 Dapr 状态存储实现在技术上是底层数据存储引擎的客户端，所以这种模拟应该直接使用存储引擎提供的并发控制机制。

Actor 状态

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

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

Dapr 状态管理的批量操作（bulk 或 multi）

Dapr 支持两种类型的批量操作 - bulk 或 multi。 您可以将几个相同类型的请求分组成批量(或批次)。 Dapr 将请求作为单个请求批量提交给基础数据存储。 换句话说，批量（bulk）操作不是事务性的。 另一方面，您可以将不同类型的请求分组为多（multi）操作，作为原子事务处理。

此功能仅对支持 ACID 事务的数据存储可用。 在撰写本文时，这些存储包括 Redis、MongoDB、PostgreSQL、SQL Server和Azure CosmosDB。

存储组件查看：supported-state-stores

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

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

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

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

了解更多，请查看：

• How-To: Build a stateful service / 如何构建一个有状态服务
• How-To: Share state between applications / 如何在应用程序之间共享状态

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

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

应用程序状态通常需要静态加密，以在企业工作负载或受监管环境中提供更强的安全性。 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}");

例如，要显式设置 持久化状态 （忽略为键设置的任何 TTL），请将 ttlInSeconds 值指定为 -1。

Dapr 状态管理 PAI

Dapr Component（组件）yaml 配置文件格式：

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: <NAME>
  namespace: <NAMESPACE>
spec:
  type: state.<TYPE>
  version: v1
  metadata:
  - name: <KEY>
    value: <VALUE>
  - name: <KEY>
    value: <VALUE>

yaml 配置文件说明

• metadata.name 是状态存储的名称；
• spec/metadata 部分是一个开放键/值对（key/value）元数据，它允许绑定定义连接属性；

从0.4.0版本开始，添加了对多个状态存储的支持。与以前的版本相比，这是一个突破性的变化，因为更改了状态API以支持这一新场景。
了解更多详细信息，请参考：https://github.com/dapr/dapr/blob/master/docs/decision_records/api/API-008-multi-state-store-api-design.md

Dapr 状态管理 API 的 CURD 操作

关于上面操作行为的详细信息（url参数，请求头/请求体，响应状态码/响应体和示例），请查看

• Reference / Dapr API / State management API

Dapr 状态管理项目实践（.NET）

我们继续在之前创建的【FrontEnd】项目创建 DaprStateManagementController ，并用以演示 Dapr 状态管理的 API 操作。

使用的存储组件是默认的 redis，对应的 yaml 配置如下：

# 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: ""
  - name: actorStateStore
    value: "true"

• windows 环境组件的 yaml 文件路径 ：C:\Users<username>.dapr\components
• self-hosted 模式默认初始化了 statestore.yaml

代码 demo 演示：

using Dapr;
using Dapr.Client;
using Microsoft.AspNetCore.Mvc;
using System.Text;

namespace FrontEnd.Controllers
{
    [Route("api/[controller]")]
    [ApiController]
    public class DaprStateManagementController : ControllerBase
    {
        private readonly ILogger<DaprStateManagementController> _logger;
        private readonly DaprClient _daprClient;
        public DaprStateManagementController(ILogger<DaprStateManagementController> logger, DaprClient daprClient)
        {
            _logger = logger;
            _daprClient = daprClient;
        }

        const string STATE_STORE = "statestore";
        const string KEY_NAME = "guid";

        // 强一致性保存一个值
        [HttpPost]
        public async Task<ActionResult> PostAsync()
        {
            _logger.LogInformation("Enter the PostAsync method.");
            await _daprClient.SaveStateAsync<string>(STATE_STORE, KEY_NAME, Guid.NewGuid().ToString(), new StateOptions() { Consistency = ConsistencyMode.Strong });
            return Ok("done");
        }

        // 通过ETag防止并发冲突，保存一个值(默认最后写入获胜)
        [HttpPost("withetag")]
        public async Task<ActionResult> PostWithETagAsync()
        {
            _logger.LogInformation("Enter the PostWithETagAsync method.");
            var (value, etag) = await _daprClient.GetStateAndETagAsync<string>(STATE_STORE, KEY_NAME);
            await _daprClient.TrySaveStateAsync<string>(STATE_STORE, KEY_NAME, Guid.NewGuid().ToString(), etag, new StateOptions() { Concurrency = ConcurrencyMode.FirstWrite });
            return Ok("done");
        }

        // 获取一个值
        [HttpGet]
        public async Task<ActionResult> GetAsync()
        {
            _logger.LogInformation("Enter the GetAsync method.");
            var result = await _daprClient.GetStateAsync<string>(STATE_STORE, KEY_NAME);
            return Ok(result);
        }
        // 获取一个值和ETag
        [HttpGet("withetag")]
        public async Task<ActionResult> GetWithETagAsync()
        {
            _logger.LogInformation("Enter the GetWithETagAsync method.");
            var (value, etag) = await _daprClient.GetStateAndETagAsync<string>(STATE_STORE, KEY_NAME);
            return Ok($"value is {value}, etag is {etag}");
        }

        // 删除一个值
        [HttpDelete]
        public async Task<ActionResult> DeleteAsync()
        {
            _logger.LogInformation("Enter the DeleteAsync method.");
            await _daprClient.DeleteStateAsync(STATE_STORE, KEY_NAME);
            return Ok("done");
        }
        // 通过ETag防止并发冲突，删除一个值
        [HttpDelete("withetag")]
        public async Task<ActionResult> DeleteWithETagAsync()
        {
            _logger.LogInformation("Enter the DeleteWithETagAsync method.");
            var (value, etag) = await _daprClient.GetStateAndETagAsync<string>(STATE_STORE, KEY_NAME);
            var success = await _daprClient.TryDeleteStateAsync(STATE_STORE, KEY_NAME, etag);
            return Ok($"value is {value}, etag is {etag}");
        }

        // 根据FromState获取一个值，健值name从路由模板获取
        [HttpGet("fromState/{name}")]
        public async Task<ActionResult> GetFromBindingAsync([FromState(STATE_STORE, KEY_NAME)] StateEntry<string> state)
        {
            _logger.LogInformation("Enter the GetFromBindingAsync method.");
            var myVal = await Task.FromResult(state.Value);
            return Ok(myVal);
        }

        // 根据FromState获取并修改值，健值name从路由模板获取
        [HttpPut("fromState/{name}")]
        public async Task<ActionResult> PutFromBindingAsync([FromState(STATE_STORE, KEY_NAME)] StateEntry<string> state)
        {
            _logger.LogInformation("Enter the PostWithBindingAsync method.");
            state.Value = Guid.NewGuid().ToString();
            return Ok(await state.TrySaveAsync());
        }

        // 通过事务保存和检索多个状态
        [HttpPost("list")]
        public async Task<ActionResult> PostListAsync() 
        {
            var list = new List<StateTransactionRequest>()
            {
                new StateTransactionRequest("test1", Encoding.UTF8.GetBytes("value1"), StateOperationType.Upsert),
                new StateTransactionRequest("test2", Encoding.UTF8.GetBytes("value2"), StateOperationType.Upsert),
            };
            await _daprClient.ExecuteStateTransactionAsync(STATE_STORE, list);

            var datas = await _daprClient.GetBulkStateAsync(STATE_STORE, list.Select(r => r.Key).ToList(), 0);
            return Ok($"Got items: {string.Join(",", datas.Select(d => $"{d.Key}={d.Value}"))}");
        }

        // 获取多个个值
        [HttpGet("list")]
        public async Task<ActionResult> GetListAsync()
        {
            _logger.LogInformation("Enter the GetListAsync method.");
            var result = await _daprClient.GetBulkStateAsync(STATE_STORE, new List<string> { KEY_NAME }, 10);
            return Ok(result);
        }

        // 删除多个个值
        [HttpDelete("list")]
        public async Task<ActionResult> DeleteListAsync()
        {
            _logger.LogInformation("Enter the DeleteListAsync method.");
            var data = await _daprClient.GetBulkStateAsync(STATE_STORE, new List<string> { KEY_NAME }, 10);
            var removeList = new List<BulkDeleteStateItem>();
            foreach (var item in data)
            {
                removeList.Add(new BulkDeleteStateItem(item.Key, item.ETag));
            }
            await _daprClient.DeleteBulkStateAsync(STATE_STORE, removeList);
            return Ok("done");
        }

    }
}

总结

在【一致性】中，dapr能够满足 最终一致性 和 强一致性，这两个特性的区别在于对 可用性 的影响程序。如果是最终一致性，那么可用性能够在某种程度上得到保证，而如果是强一致性，那么可能系统会出现较长时间的未响应，从而影响到了可用性（Availability，CAP中的A）。

在【并发性】中，dapr 支持 first-write-wins 和 last-write-wins 两种策略，first-write-wins 的实现主要依靠携带 ETag 所包含的数据版本信息的乐观并发处理/乐观锁（OCC），而如果没有携带 ETag 的情况就是一个 last-write-wins 的并发策略，在这种情况下，数据总是能够被更新，最后更新的数据会覆盖掉所有更新。

参考 Dapr 状态管理

• Dapr 状态管理构建基块
• 状态管理概览 （状态管理构建块概览）
• 指南：如何保存和获取状态 （使用键值对来持久化状态）
• 指南：如何创建一个有状态的服务 （对可伸缩的副本使用状态管理）
• 指南：如何在应用程序之间共享状态 （以不同的策略在不同的应用程序之间共享状态）
• 使用后端状态存储 （用于使用特定后端状态存储的指南）