手把手教你學Dapr - 7. Actors-有解無憂

介紹

Actor模式將Actor描述為最低級別的“計算單元”，換句話說，您在一個獨立的單元（稱為actor）中撰寫代碼，該單元接收訊息并一次處理一個訊息，沒有任何并發或執行緒，

再換句話說，根據ActorId劃分獨立計算單元后，相同的ActorId重入要排隊，可以理解為lock(ActorId)

注：這里有個反例，就是重入性的引入，這個概念目前還是Preview，它允許同一個鏈內可以重復進入，判斷的標準不止是ActorId這么簡單，即自己調自己是被允許的，這個默認是關閉的，需要手動開啟，即默認不允許自己調自己，

當您的代碼處理一條訊息時，它可以向其他參與者發送一潭訓多條訊息，或者創建新的參與者，底層運行時管理每個參與者運行的方式、時間和地點，并在參與者之間路由訊息，

大量的Actor可以同時執行，Actor彼此獨立執行，

Dapr 包含一個運行時，它專門實作了 Virtual Actor 模式，通過 Dapr 的實作，您可以根據 Actor 模型撰寫 Dapr Actor，而 Dapr 利用底層平臺提供的可擴展性和可靠性保證，

什么時候用Actors

Actor 設計模式非常適合許多分布式系統問題和場景，但您首先應該考慮的是該模式的約束，一般來說，如果出現以下情況，請考慮使用Actors模式來為您的問題或場景建模：

您的問題空間涉及大量（數千個或更多）小的、獨立且孤立的狀態和邏輯單元，
您希望使用不需要與外部組件進行大量互動的單執行緒物件，包括跨一組Actors查詢狀態，
您的 Actor 實體不會通過發出 I/O 操作來阻塞具有不可預測延遲的呼叫者，

Dapr Actor

每個Actor都被定義為Actor型別的實體，就像物件是類的實體一樣，例如，可能有一個執行計算器功能的Actor型別，并且可能有許多該型別的Actor分布在集群的各個節點上，每個這樣的Actor都由一個Acotr ID唯一標識，

生命周期

Dapr Actors是虛擬的，這意味著他們的生命周期與他們的記憶體表現無關，因此，它們不需要顯式創建或銷毀，Dapr Actors運行時在第一次收到對該Actor ID 的請求時會自動激活該Actor，如果一個Actor在一段時間內沒有被使用，Dapr Actors運行時就會對記憶體中的物件進行垃圾回收，如果稍后需要重新激活，它還將保持對參與者存在的了解，如果稍后需要重新激活，它還將保持對 Actor 的一切原有資料，

呼叫 Actor 方法和提醒會重置空閑時間，例如提醒觸發將使Actor保持活躍，無論Actor是活躍還是不活躍，Actor提醒都會觸發，如果為不活躍的Actor觸發，它將首先激活演員，Actor 計時器不會重置空閑時間，因此計時器觸發不會使 Actor 保持活動狀態，計時器僅在Actor處于活動狀態時觸發，

Reminders 和 Timers 最大的區別就是Reminders會保持Actor的活動狀態，而Timers不會

Dapr 運行時用來查看Actor是否可以被垃圾回收的空閑超時和掃描間隔是可配置的，當 Dapr 運行時呼叫 Actor 服務以獲取支持的 Actor 型別時，可以傳遞此資訊，

由于Virtual Actor模型的存在，這種Virtual Actor生命周期抽象帶來了一些注意事項，事實上，Dapr Actors實作有時會偏離這個模型，

第一次將訊息發送到Actor ID時，Actor被自動激活(導致構建Actor物件)，經過一段時間后，Actor物件將被垃圾回收，被回收后再次使用Actor ID將導致構造一個新的Actor物件， Actor 的狀態比物件的生命周期長，因為狀態存盤在 Dapr 運行時配置的狀態管理組件中，

注：Actor被垃圾回收之前，Actor物件是會復用的，這里會導致一個問題，在.Net Actor類中，建構式在Actor存活期間只會被呼叫一次，

分發和故障轉移

為了提供可擴展性和可靠性，Actor 實體分布在整個集群中，Dapr 根據需要自動將它們從故障節點遷移到健康節點，

Actors 分布在 Actor 服務的實體中，而這些實體分布在集群中的節點之間，對于給定的Actor型別，每個服務實體都包含一組Actor，

Dapr安置服務(Placement Service)

Dapr Actor 運行時為您管理分發方案和密鑰范圍設定，這是由Actor Placement 服務完成的，創建服務的新實體時，相應的 Dapr 運行時會注冊它可以創建的Actor型別，并且安置服務會計算給定Actor型別的所有實體的磁區，每個Actor型別的磁區資訊表被更新并存盤在環境中運行的每個Dapr實體中，并且可以隨著Actor服務的新實體的創建和銷毀而動態變化，這如下圖所示：

當客戶端呼叫具有特定ID的Actor(例如，Actor ID 123)時，客戶端的 Dapr 實體會Hash Actor型別和 ID，并使用該資訊呼叫可以為特定Actor ID的請求提供服務的相應Dapr實體，因此，始終為任何給定的Actor ID 呼叫相同的磁區（或服務實體），這如下圖所示：

這簡化了一些選擇，但也帶來了一些考慮:

默認情況下，Actor 隨機放置到 pod 中，從而實作均勻分布，
因為Actor是隨機放置的，應該可以預料到Actor操作總是需要網路通信，包括方法呼叫資料的序列化和反序列化，產生延遲和開銷，

注：Dapr Actor 放置服務僅用于 Actor 放置，因此如果您的服務不使用 Dapr Actors，則不需要，放置服務可以在所有托管環境中運行，包括自托管和 Kubernetes，

Actor通訊

您可以通過HTTP/gRPC呼叫Actor，當然也可以用SDK，

POST/GET/PUT/DELETE http://localhost:3500/v1.0/actors/<actorType>/<actorId>/<method/state/timers/reminders>

并發

Dapr Actor 運行時為訪問 Actor 方法提供了一個簡單的回合制（turn-basesd）的訪問模型，這意味著在任何時候，Actor 物件的代碼中都不能有超過一個執行緒處于活動狀態，

單個Actor實體一次不能處理多個請求，如果預期要處理并發請求，Actor 實體可能會導致吞吐量瓶頸，

單個Actor實體指每個Actor ID對應的Actor物件，單個Actor不并發就沒有問題

如果兩個 Actor 之間存在回圈請求，而同時向其中一個 Actor 發出外部請求，則 Actor 之間可能會陷入僵局，Dapr Actor運行時自動超時Actor呼叫并向呼叫者拋出例外以中斷可能的死鎖情況，

重入性（Preview）

作為對 dapr 中基礎 Actor 的增強，現在重入性為預覽功能，感興趣的小伙伴可以到看官方檔案，

回合制訪問（Turn-based access）

一個回合包括一個Actor方法的完整執行以回應來自其他Actor或客戶端的請求，或者一個計時器/提醒回呼的完整執行，即使這些方法和回呼是異步的，Dapr Actor運行時也不會將它們交叉，一個回合必須完全完成后，才允許進行新的回合，換句話說，當前正在執行的Actor方法或計時器/提醒回呼必須完全完成，才能允許對方法或回呼的新呼叫，

Dapr Actor運行時通過在回合開始時獲取每個Actor的鎖并在回合結束時釋放鎖來實作基于回合的并發性，因此，基于回合的并發是在每個Actor的基礎上執行的，而不是跨Actor，Actor 方法和計時器/提醒回呼可以代表不同的 Actor 同時執行，

以下示例說明了上述概念，考慮實作兩個異步方法（例如 Method1 和 Method2）、計時器和提醒的Actor 型別，下圖顯示了代表屬于此Actor型別的兩個Actors（ActorId1 和 ActorId2）執行這些方法和回呼的時間線示例，

Actor狀態管理

Actor可以使用狀態管理功能可靠地保存狀態，您可以通過 HTTP/gRPC 端點與 Dapr 互動以進行狀態管理，

要使用 actor，您的狀態存盤必須支持事務，這意味著您的狀態存盤組件必須實作 TransactionalStore 介面，只有一個狀態存盤組件可以用作所有參與者的狀態存盤，

事務支持串列：https://docs.dapr.io/reference/components-reference/supported-state-stores/

注：建議學習的時候都用Redis，官方所有的示例也都是基于Redis，比較容易上手，且Dapr init默認集成

Actor計時器和提醒

Actor可以通過注冊計時器或提醒來安排自己的定期作業，

計時器和提醒的功能非常相似，主要區別在于，Dapr Actor 運行時在停用后不保留有關計時器的任何資訊，而使用 Dapr Actor 狀態提供程式保留有關提醒的資訊，

定時器和提醒的調度配置是相同的，總結如下：

DueTime 是一個可選引數，用于設定第一次呼叫回呼之前的時間或時間間隔，如果省略 DueTime，則在定時器/提醒注冊后立即呼叫回呼，

支持的格式：

RFC3339 日期格式，例如2020-10-02T15:00:00Z
time.Duration 格式，例如2h30m
ISO 8601 持續時間格式，例如PT2H30M

period 是一個可選引數，用于設定兩次連續回呼呼叫之間的時間間隔，當以 ISO 8601-1 持續時間格式指定時，您還可以配置重復次數以限制回呼呼叫的總數，如果省略 period，則回呼將僅被呼叫一次，

支持的格式：

time.Duration 格式，例如2h30m
ISO 8601 持續時間格式，例如PT2H30M, R5/PT1M30S

ttl 是一個可選引數，用于設定計時器/提醒到期和洗掉的時間或時間間隔，如果省略 ttl，則不應用任何限制，

支持的格式：

RFC3339 日期格式，例如2020-10-02T15:00:00Z
time.Duration 格式，例如2h30m
ISO 8601 持續時間格式，例如PT2H30M

當您同時指定周期內的重復次數和 ttl 時，計時器/提醒將在滿足任一條件時停止，

Actor 運行時配置

actorIdleTimeout - 停用空閑 actor 之前的超時時間，每個 actorScanInterval 間隔都會檢查超時，默認值：60 分鐘
actorScanInterval - 指定掃描演員以停用空閑Actor的頻率的持續時間，閑置時間超過 actor_idle_timeout 的 Actor 將被停用，默認值：30 秒
drainOngoingCallTimeout - 在耗盡Rebalanced的Actor的程序中的持續時間，這指定了當前活動 Actor 方法完成的超時時間，如果當前沒有 Actor 方法呼叫，則忽略此項，默認值：60 秒
drainRebalancedActors - 如果為 true，Dapr 將等待 drainOngoingCallTimeout 持續時間以允許當前角色呼叫完成，然后再嘗試停用角色，默認值：true

drainRebalancedActors與上面的drainOngoingCallTimeout需搭配使用
reentrancy - (ActorReentrancyConfig) - 配置角色的重入行為，如果未提供，則禁用可重入，默認值：disabled, 0
remindersStoragePartitions - 配置Actor提醒的磁區數，如果未提供，則所有提醒都將保存為Actor狀態存盤中的單個記錄，默認值：0

// In Startup.cs
public void ConfigureServices(IServiceCollection services)
{
    // Register actor runtime with DI
    services.AddActors(options =>
    {
        // Register actor types and configure actor settings
        options.Actors.RegisterActor<MyActor>();

        // Configure default settings
        options.ActorIdleTimeout = TimeSpan.FromMinutes(60);
        options.ActorScanInterval = TimeSpan.FromSeconds(30);
        options.DrainOngoingCallTimeout = TimeSpan.FromSeconds(60);
        options.DrainRebalancedActors = true;
        options.RemindersStoragePartitions = 7;
        // reentrancy not implemented in the .NET SDK at this time
    });

    // Register additional services for use with actors
    services.AddSingleton<BankService>();
}

磁區提醒（Preview）

在 sidecar 重新啟動后，Actor 提醒會保留并繼續觸發，在 Dapr 運行時版本 1.3 之前，提醒被保存在 actor 狀態存盤中的單個記錄上，

此為Preview功能，感興趣可以看官方檔案

.Net呼叫Dapr的Actor

與以往不同，Actor示例會多創建一個共享類別庫用于存放Server和Client共用的部分

創建Assignment.Shared

創建類別庫專案，并添加Dapr.ActorsNuGet包參考，最后添加以下幾個類：

AccountBalance.cs

namespace Assignment.Shared;
public class AccountBalance
{
    public string AccountId { get; set; } = default!;

    public decimal Balance { get; set; }
}

IBankActor.cs

注：這個是Actor介面，IActor是Dapr SDK提供的

using Dapr.Actors;

namespace Assignment.Shared;
public interface IBankActor : IActor
{
    Task<AccountBalance> GetAccountBalance();

    Task Withdraw(WithdrawRequest withdraw);
}

OverdraftException.cs

namespace Assignment.Shared;
public class OverdraftException : Exception
{
    public OverdraftException(decimal balance, decimal amount)
        : base($"Your current balance is {balance:c} - that's not enough to withdraw {amount:c}.")
    {
    }
}

WithdrawRequest.cs

namespace Assignment.Shared;
public class WithdrawRequest
{
    public decimal Amount { get; set; }
}

創建Assignment.Server

創建類別庫專案，并添加Dapr.Actors.AspNetCoreNuGet包參考和Assignment.Shared專案參考，最后修改程式埠為5000，

注：Server與Shared和Client的NuGet包不一樣，Server是集成了服務端的一些功能

修改program.cs

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSingleton<BankService>();
builder.Services.AddActors(options =>
{
    options.Actors.RegisterActor<DemoActor>();
});

var app = builder.Build();

app.UseRouting();

app.UseEndpoints(endpoints =>
{
    endpoints.MapActorsHandlers();
});

app.Run();

添加BankService.cs

using Assignment.Shared;

namespace Assignment.Server;
public class BankService
{
    // Allow overdraft of up to 50 (of whatever currency).
    private readonly decimal OverdraftThreshold = -50m;

    public decimal Withdraw(decimal balance, decimal amount)
    {
        // Imagine putting some complex auditing logic here in addition to the basics.

        var updated = balance - amount;
        if (updated < OverdraftThreshold)
        {
            throw new OverdraftException(balance, amount);
        }

        return updated;
    }
}

添加BankActor.cs

using Assignment.Shared;
using Dapr.Actors.Runtime;
using System;

namespace Assignment.Server;
public class BankActor : Actor, IBankActor, IRemindable // IRemindable is not required
{
    private readonly BankService bank;

    public BankActor(ActorHost host, BankService bank)
        : base(host)
    {
        // BankService is provided by dependency injection.
        // See Program.cs
        this.bank = bank;
    }

    public async Task<AccountBalance> GetAccountBalance()
    {
        var starting = new AccountBalance()
        {
            AccountId = this.Id.GetId(),
            Balance = 10m, // Start new accounts with 100, we're pretty generous.
        };

        var balance = await StateManager.GetOrAddStateAsync("balance", starting);
        return balance;
    }

    public async Task Withdraw(WithdrawRequest withdraw)
    {
        var starting = new AccountBalance()
        {
            AccountId = this.Id.GetId(),
            Balance = 10m, // Start new accounts with 100, we're pretty generous.
        };

        var balance = await StateManager.GetOrAddStateAsync("balance", starting)!;

        if (balance.Balance <= 0)
        {
            // Simulated reminder deposit
            if (Random.Shared.Next(100) > 90)
            {
                await RegisterReminderAsync("Deposit", null, TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(-1));
            }
        }

        // Throws Overdraft exception if the account doesn't have enough money.
        var updated = this.bank.Withdraw(balance.Balance, withdraw.Amount);

        balance.Balance = updated;
        await StateManager.SetStateAsync("balance", balance);
    }

    public async Task ReceiveReminderAsync(string reminderName, byte[] state, TimeSpan dueTime, TimeSpan period)
    {
        if (reminderName == "Deposit")
        {
            var balance = await StateManager.GetStateAsync<AccountBalance>("balance")!;

            if (balance.Balance <= 0)
            {
                balance.Balance += 60; // 50(Overdraft Threshold) + 10 = 60
                Console.WriteLine("Deposit: 10");
            }
            else
            {
                Console.WriteLine("Deposit: ignore");
            }
        }
    }
}

運行Assignment.Server

使用Dapr CLI來啟動，先使用命令列工具跳轉到目錄 dapr-study-room\Assignment07\Assignment.Server，然后執行下面命令

dapr run --app-id testactor --app-port 5000 --dapr-http-port 3500 --dapr-grpc-port 50001 dotnet run

創建Assignment.Client

創建控制臺專案，并添加Dapr.ActorsNuGet包參考和Assignment.Shared專案參考，

修改Program.cs

using Assignment.Shared;
using Dapr.Actors;
using Dapr.Actors.Client;

Console.WriteLine("Creating a Bank Actor");
var bank = ActorProxy.Create<IBankActor>(ActorId.CreateRandom(), "BankActor");
Parallel.ForEach(Enumerable.Range(1, 10), async i =>
{
    while (true)
    {
        var balance = await bank.GetAccountBalance();
        Console.WriteLine($"[Worker-{i}] Balance for account '{balance.AccountId}' is '{balance.Balance:c}'.");

        Console.WriteLine($"[Worker-{i}] Withdrawing '{1m:c}'...");
        try
        {
            await bank.Withdraw(new WithdrawRequest() { Amount = 1m });
        }
        catch (ActorMethodInvocationException ex)
        {
            Console.WriteLine("[Worker-{i}] Overdraft: " + ex.Message);
        }

        Task.Delay(1000).Wait();
    }
});

Console.ReadKey();