手把手教你学Dapr - 7. Actors

2021年11月21日 阅读数:12
这篇文章主要向大家介绍手把手教你学Dapr - 7. Actors,主要内容包括基础应用、实用技巧、原理机制等方面,希望对大家有所帮助。

介绍

Actor模式将Actor描述为最低级别的“计算单元”。换句话说,您在一个独立的单元(称为actor)中编写代码,该单元接收消息并一次处理一个消息,没有任何并发或线程。git

再换句话说,根据ActorId划分独立计算单元后,相同的ActorId重入要排队,能够理解为lock(ActorId)github

:这里有个反例,就是重入性的引入,这个概念目前仍是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惟一标识。app

actor_background_game_example.png

生命周期

Dapr Actors是虚拟的,这意味着他们的生命周期与他们的内存表现无关。所以,它们不须要显式建立或销毁。Dapr Actors运行时在第一次收到对该Actor ID 的请求时会自动激活该Actor。若是一个Actor在一段时间内没有被使用,Dapr Actors运行时就会对内存中的对象进行垃圾回收。若是稍后须要从新激活,它还将保持对参与者存在的了解。若是稍后须要从新激活,它还将保持对 Actor 的一切原有数据。框架

调用 Actor 方法和提醒会重置空闲时间,例如提醒触发将使Actor保持活跃。不管Actor是活跃仍是不活跃,Actor提醒都会触发,若是为不活跃的Actor触发,它将首先激活演员。Actor 计时器不会重置空闲时间,所以计时器触发不会使 Actor 保持活动状态。计时器仅在Actor处于活动状态时触发。dom

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服务的新实例的建立和销毁而动态变化。这以下图所示:

actors_background_placement_service_registration.png

当客户端调用具备特定ID的Actor(例如,Actor ID 123)时,客户端的 Dapr 实例会Hash Actor类型和 ID,并使用该信息调用能够为特定Actor ID的请求提供服务的相应Dapr实例。所以,始终为任何给定的Actor ID 调用相同的分区(或服务实例)。这以下图所示:

actors_background_id_hashing_calling.png

这简化了一些选择,但也带来了一些考虑:

  • 默认状况下,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调用并向调用者抛出异常以中断可能的死锁状况。

actors_background_communication.png

重入性(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)执行这些方法和回调的时间线示例。

actors_background_concurrency.png

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();

运行Assignment.Client

使用Dapr CLI来启动,先使用命令行工具跳转到目录 dapr-study-room\Assignment07\Assignment.Client,而后执行下面命令

dotnet run

本章源码

Assignment07

https://github.com/doddgu/dapr-study-room

咱们正在行动,新的框架、新的生态

咱们的目标是自由的易用的可塑性强的功能丰富的健壮的

因此咱们借鉴Building blocks的设计理念,正在作一个新的框架MASA Framework,它有哪些特色呢?

  • 原生支持Dapr,且容许将Dapr替换成传统通讯方式
  • 架构不限,单体应用、SOA、微服务都支持
  • 支持.Net原生框架,下降学习负担,除特定领域必须引入的概念,坚持不造新轮子
  • 丰富的生态支持,除了框架之外还有组件库、权限中心、配置中心、故障排查中心、报警中心等一系列产品
  • 核心代码库的单元测试覆盖率90%+
  • 开源、免费、社区驱动
  • 还有什么?咱们在等你,一块儿来讨论

通过几个月的生产项目实践,已完成POC,目前正在把以前的积累重构到新的开源项目中

目前源码已开始同步到Github(文档站点在规划中,会慢慢完善起来):

MASA.BuildingBlocks

MASA.Contrib

MASA.Utils

MASA.EShop

BlazorComponent

MASA.Blazor

QQ群:7424099

微信群:加技术运营微信(MasaStackTechOps),备注来意,邀请进群

masa_stack_tech_ops.png