指南:创建一个有状态的服务

对可伸缩的副本使用状态管理

在这篇文章中,你将了解到如何创建一个可以水平扩展的有状态服务,使用选入并发和一致性模型。 使用状态管理API可以使开发人员摆脱繁琐的状态协调、冲突解决和故障处理。

建立一个状态存储

状态存储组件代表Dapr用来与数据库进行通信的资源。 在本指南中,我们将使用默认的 Redis 状态存储。

使用 Dapr CLI

当你在自托管模式下运行dapr init时,Dapr会创建一个默认的Redisstatestore.yaml并在你的本地机器上运行一个Redis状态存储,它位于:

  • 在Windows上,在 %UserProfile%\.dapr\components\statestore.yaml
  • 在Linux/MacOS上,在~/.dapr/components/statestore.yaml

通过 statestore.yaml 组件,您可以轻松更换底层组件,而无需更改应用程序代码。

查看状态存储列表

Kubernetes

如何在Kubernetes上设置不同的状态存储

强一致性和最终一致性

使用强一致性,Dapr 确保底层状态存储:

  • 数据写入所有副本后返回响应。
  • 在写入或删除状态之前,从法定人数那里接收到一个 ACK。

对于 get 请求,Dapr 确保存储在副本之间始终返回最新的数据。 除非在对状态 API 的请求中另有指定,否则默认为最终一致性。

下面的例子说明了如何使用强一致性保存、获取和删除状态。 该示例是用 Python 编写的,但适用于任何编程语言。

保存状态

  1. import requests
  2. import json
  3. store_name = "redis-store" # name of the state store as specified in state store component yaml file
  4. dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
  5. stateReq = '[{ "key": "k1", "value": "Some Data", "options": { "consistency": "strong" }}]'
  6. response = requests.post(dapr_state_url, json=stateReq)

获取状态

  1. import requests
  2. import json
  3. store_name = "redis-store" # name of the state store as specified in state store component yaml file
  4. dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
  5. response = requests.get(dapr_state_url + "/key1", headers={"consistency":"strong"})
  6. print(response.headers['ETag'])

删除状态

  1. import requests
  2. import json
  3. store_name = "redis-store" # name of the state store as specified in state store component yaml file
  4. dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
  5. response = requests.delete(dapr_state_url + "/key1", headers={"consistency":"strong"})

如果未指定concurrency选项,则默认为最后写入并发模式。

First-write-wins 和 Last-write-wins

Dapr 允许开发人员在处理数据存储时选择两种常见的并发模式:

  • First-write-wins:在有多个应用程序实例,同时向同一个键进行写入的情况下,First-Write-Wins 策略非常有用。
  • Last-write-wins: Dapr 的默认模式。

Dapr 使用版本号来确定一个特定的键是否已经更新。 你可以:

  1. 读取密钥数据时保留版本号。
  2. 在更新时(例如写入和删除操作)使用版本号。

如果版本信息在检索版本号后发生了变化,就会抛出一个错误,需要您执行另一个读取操作以获取最新的版本信息和状态。

Dapr利用 ETags 来确定状态的版本号。 ETags从状态请求中以ETag头返回。 使用 ETags,您的应用程序知道自上次检查以来资源已经被更新,因为在 ETag 不匹配时出现错误。

以下示例显示如何:

  • 获取一个 ETag。
  • 使用 ETag 保存状态。
  • 删除状态。

该示例是用 Python 编写的,但适用于任何编程语言。

  1. import requests
  2. import json
  3. store_name = "redis-store" # name of the state store as specified in state store component yaml file
  4. dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
  5. response = requests.get(dapr_state_url + "/key1", headers={"concurrency":"first-write"})
  6. etag = response.headers['ETag']
  7. newState = '[{ "key": "k1", "value": "New Data", "etag": {}, "options": { "concurrency": "first-write" }}]'.format(etag)
  8. requests.post(dapr_state_url, json=newState)
  9. response = requests.delete(dapr_state_url + "/key1", headers={"If-Match": "{}".format(etag)})

处理版本不匹配引起的失败

在下面的例子中,您将看到在版本发生变化时如何重试保存状态操作:

  1. import requests
  2. import json
  3. # This method saves the state and returns false if failed to save state
  4. def save_state(data):
  5. try:
  6. store_name = "redis-store" # name of the state store as specified in state store component yaml file
  7. dapr_state_url = "http://localhost:3500/v1.0/state/{}".format(store_name)
  8. response = requests.post(dapr_state_url, json=data)
  9. if response.status_code == 200:
  10. return True
  11. except:
  12. return False
  13. return False
  14. # This method gets the state and returns the response, with the ETag in the header -->
  15. def get_state(key):
  16. response = requests.get("http://localhost:3500/v1.0/state/<state_store_name>/{}".format(key), headers={"concurrency":"first-write"})
  17. return response
  18. # Exit when save state is successful. success will be False if there's an ETag mismatch -->
  19. success = False
  20. while success != True:
  21. response = get_state("key1")
  22. etag = response.headers['ETag']
  23. newState = '[{ "key": "key1", "value": "New Data", "etag": {}, "options": { "concurrency": "first-write" }}]'.format(etag)
  24. success = save_state(newState)