Devlooped.CloudActors 0.4.0

一个有观点的、简化的和统一的云原生actors库，支持与Microsoft Orleans集成。

概述

与Orleans原生提供的（并鼓励使用的）RPC编程风格不同，Cloud Actors提供了一种统一API来访问actors的基于消息传递的编程风格：执行和查询。

这些统一操作接收一个消息（即命令或查询），并可选择返回一个结果。消费者始终使用相同的API在actors上调用操作，actor id和消息的组合足够用于将消息路由到正确的actor。

actor可以当作普通CLR对象实现，无需继承任何基类或实现任何接口。Orleans中grain及其激活的“管道”完全对开发人员隐藏。

功能

与依赖dynamic分发的实现相比，这种实现高度依赖源生成器来提供强类型消息路由，同时保留一个灵活的机制供实现者使用。

此外，这个库使grain完全透明于开发者。他们甚至不需要依赖Orleans。换句话说：开发者将其业务逻辑作为一个普通的CLR对象（POCO）编写。

库的中心抽象是actor总线

public interface IActorBus
{
    Task ExecuteAsync(string id, IActorCommand command);
    Task<TResult> ExecuteAsync<TResult>(string id, IActorCommand<TResult> command);
    Task<TResult> QueryAsync<TResult>(string id, IActorQuery<TResult> query);
}

actor会接收消息进行处理，这些通常是以普通记录形式的消息，例如

[GenerateSerializer]
public partial record Deposit(decimal Amount) : IActorCommand;  // 👈 marker interface for void commands

[GenerateSerializer]
public partial record Withdraw(decimal Amount) : IActorCommand;

[GenerateSerializer]
public partial record Close(CloseReason Reason = CloseReason.Customer) : IActorCommand<decimal>;         // 👈 marker interface for value-returning commands

public enum CloseReason
{
    Customer,
    Fraud,
    Other
}

[GenerateSerializer]
public partial record GetBalance() : IActorQuery<decimal>;      // 👈 marker interface for queries (a.k.a. readonly methods)

我们可以看出，与常规Orleans参数区别开的是actor消息需要实现IActorCommand或IActorQuery接口。您可以查看库支持的这三种消息类型

• IActorCommand - 发送给actor以处理的消息，但不返回结果。
• IActorCommand<TResult> - 发送给actor以处理的消息，并返回结果。
• IActorQuery<TResult> - 发送给actor以处理的消息，并返回结果。它与前面的类型不同，因为它是一个只读操作，这意味着它不会改变actor的状态。这导致了在grain上调用只读方法。

actor本身只需包含[Actor]属性就被识别为actor

[Actor]
public class Account    // 👈 no need to inherit or implement anything by default
{
    public Account(string id) => Id = id;       // 👈 no need for parameterless constructor

    public string Id { get; }
    public decimal Balance { get; private set; }
    public bool IsClosed { get; private set; }
    public CloseReason Reason { get; private set; }

    //public void Execute(Deposit command)      // 👈 methods can be overloads of message types
    //{
    //    // validate command
    //    // decrease balance
    //}

    // Showcases that operations can have a name that's not Execute
    public Task DepositAsync(Deposit command)   // 👈 but can also use any name you like
    {
        // validate command
        Balance +-= command.Amount;
        return Task.CompletedTask;
    }

    // Showcases that operations don't have to be async
    public void Execute(Withdraw command)       // 👈 methods can be sync too
    {
        // validate command
        Balance -= command.Amount;
    }

    // Showcases value-returning operation with custom name.
    // In this case, closing the account returns the final balance.
    // As above, this can be async or not.
    public decimal Close(Close command)
    {
        var balance = Balance;
        Balance = 0;
        IsClosed = true;
        Reason = command.Reason;
        return balance;
    }

    // Showcases a query that doesn't change state
    public decimal Query(GetBalance _) => Balance;  // 👈 becomes [ReadOnly] grain operation
}

注意：具有私有setter的属性在从存储中读取最新状态时无需任何其他属性即可正确反序列化。一个源生成器提供相应的构造函数以供反序列化使用

在托管方面，提供了一个名为AddCloudActors的扩展方法来注册自动生成的grain并将调用路由到actor

var builder = WebApplication.CreateSlimBuilder(args);

builder.Host.UseOrleans(silo =>
{
    silo.UseLocalhostClustering();
    // 👇 registers generated grains, actor bus and activation features
    silo.AddCloudActors(); 
});

如何工作

库使用源生成器来生成grain类。通过在项目中将EmitCompilerGeneratedFiles属性设置为true和检查obj文件夹，可以轻松地检查生成的代码。

对于上面的actor，生成的grain看起来像这样

public partial class AccountGrain : Grain, IActorGrain
{
    readonly IPersistentState<Account> storage; // 👈 uses recommended injected state approach

    // 👇 use [Actor("stateName", "storageName")] on actor to customize this
    public AccountGrain([PersistentState] IPersistentState<Account> storage) 
        => this.storage = storage;

    [ReadOnly]
    public Task<TResult> QueryAsync<TResult>(IActorQuery<TResult> command)
    {
        switch (command)
        {
            case Tests.GetBalance query:
                return Task.FromResult((TResult)(object)storage.State.Query(query));
            default:
                throw new NotSupportedException();
        }
    }

    public async Task<TResult> ExecuteAsync<TResult>(IActorCommand<TResult> command)
    {
        switch (command)
        {
            case Tests.Close cmd:
                var result = await storage.State.CloseAsync(cmd);
                try
                {
                    await storage.WriteStateAsync();
                }
                catch 
                {
                    await storage.ReadStateAsync(); // 👈 rollback state on failure
                    throw;
                }
                return (TResult)(object)result;
            default:
                throw new NotSupportedException();
        }
    }

    public async Task ExecuteAsync(IActorCommand command)
    {
        switch (command)
        {
            case Tests.Deposit cmd:
                await storage.State.DepositAsync(cmd);
                try
                {
                    await storage.WriteStateAsync();
                }
                catch 
                {
                    await storage.ReadStateAsync();
                    throw;
                }
                break;
            case Tests.Withdraw cmd:
                storage.State.Execute(cmd);
                try
                {
                    await storage.WriteStateAsync();
                }
                catch 
                {
                    await storage.ReadStateAsync();
                    throw;
                }
                break;
            case Tests.Close cmd:
                await storage.State.CloseAsync(cmd);
                try
                {
                    await storage.WriteStateAsync();
                }
                catch 
                {
                    await storage.ReadStateAsync();
                    throw;
                }
                break;
            default:
                throw new NotSupportedException();
        }
    }
}

注意，grain是一个部分类，因此您可以向其中添加自己的方法。生成的代码还使用你在actor类中用于处理传入消息的方法名（和重载），因此它不施加任何特定的命名约定。

由于grain元数据/注册是由源生成器生成的，而源生成器不能依赖于其他生成的代码，因此类型元数据将不会自动可用，即使我们生成了从Grain继承的类型，这通常是足够的。因此，单独的生成器生成了AddCloudActors扩展方法，这些方法正确地注册了这些类型与Orleans相关。生成的扩展方法如下（如上所示）（配置Orleans时所示的使用方法）

namespace Orleans.Runtime
{
    public static class CloudActorsExtensions
    {
        public static ISiloBuilder AddCloudActors(this ISiloBuilder builder)
        {
            builder.Configure<GrainTypeOptions>(options => 
            {
                // 👇 registers each generated grain type
                options.Classes.Add(typeof(Tests.AccountGrain));
            });

            builder.ConfigureServices(services =>
            {
                // 👇 registers IActorBus and actor activation features
                services.AddCloudActors();
            });

            return builder;
        }
    }
}

最后，为了提高对IActorBus接口的消费者可发现性，将生成扩展方法的重载，以便以非泛型重载的形式公开可用的actor消息，如以下示例所示

状态反序列化

上面的Account类只提供了一个接受账户标识符的构造函数。然而，在执行各种操作后，其状态将通过私有属性setter发生改变，而这些setter默认情况下对反序列化器不可用。.NET 7+添加了对属性的JsonInclude属性的支持进行设置，但需要添加到所有此类属性中这一点并不直观。

JSON.NET 等同的属性是 JsonProperty，同样存在相同的缺点。

为了帮助用户避免误入成功陷阱，该库自动为演员类生成一个带有 [JsonConstructor] 注解的构造函数，该构造函数将用于反序列化状态。在上面的 Account 示例中，生成的构造函数如下所示

partial class Account
{
    [EditorBrowsable(EditorBrowsableState.Never)]
    [JsonConstructor]
    public Account(string id, System.Decimal balance, System.Boolean isClosed, Tests.CloseReason reason) 
        : this(id) 
    {
        this.Balance = balance;
        this.IsClosed = isClosed;
        this.Reason = reason;
    }
}

构造函数使用 [JsonContructor] 注解并不意味着状态必须以 JSON 格式序列化。这取决于存储供应商如何调用该构造函数并传递合适的值。如果它确实使用 System.Text.Json 进行序列化，则构造函数将自动使用。

事件源

对于这些演员的消息传递风格编程的一种自然扩展是完全事件源。该库为此提供了一个名为 IEventSourced 的接口。

public interface IEventSourced
{
    IReadOnlyList<object> Events { get; }
    void AcceptEvents();
    void LoadEvents(IEnumerable<object> history);
}

基于 Streamstone 的样本谷物存储将使用从流中找到的事件调用 LoadEvents，并在谷物保存后调用 AcceptEvents 以清理事件列表。

乐观并发通过将流版本暴露给 IGranState.ETag 并在持久化时解析它来实现，以确保一致性。

用户可以按他们认为合适的方式实现此接口，但如果没有实现该接口而是继承了它，则库提供了一个默认实现。生成的实现为演员的方法提供 Raise<T>(@event) 方法以提升事件，并调用提供的 Apply(@event) 方法将事件应用到状态。生成器假定这 conventions，使用每个 Apply 方法上的 actor 的单个参数作为路由事件（无论是触发还是从存储中加载）的switch。

例如，如果上面的 Account 演员被转换为事件源演员，它将看起来像这样

[Actor]
public partial class Account : IEventSourced  // 👈 interface is *not* implemented by user!
{
    public Account(string id) => Id = id;

    public string Id { get; }
    public decimal Balance { get; private set; }
    public bool IsClosed { get; private set; }

    public void Execute(Deposit command)
    {
        if (IsClosed)
            throw new InvalidOperationException("Account is closed");

        // 👇 Raise<T> is generated when IEventSourced is inherited
        Raise(new Deposited(command.Amount));
    }

    public void Execute(Withdraw command)
    {
        if (IsClosed)
            throw new InvalidOperationException("Account is closed");
        if (command.Amount > Balance)
            throw new InvalidOperationException("Insufficient funds.");

        Raise(new Withdrawn(command.Amount));
    }

    public decimal Execute(Close command)
    {
        if (IsClosed)
            throw new InvalidOperationException("Account is closed");

        var balance = Balance;
        Raise(new Closed(Balance, command.Reason));
        return balance;
    }

    public decimal Query(GetBalance _) => Balance;

    // 👇 generated generic Apply dispatches to each based on event type

    void Apply(Deposited @event) => Balance += @event.Amount;

    void Apply(Withdrawn @event) => Balance -= @event.Amount;

    void Apply(Closed @event)
    {
        Balance = 0;
        IsClosed = true;
        Reason = @event.Reason;
    }
}

注意接口在演员中没有任何实现。生成器提供的实现如下所示

partial class Account
{
    List<object>? events;

    IReadOnlyList<object> IEventSourced.Events => events ??= new List<object>();

    void IEventSourced.AcceptEvents() => events?.Clear();

    void IEventSourced.LoadEvents(IEnumerable<object> history)
    {
        foreach (var @event in history)
        {
            Apply(@event);
        }
    }

    /// <summary>
    /// Applies an event. Invoked automatically when raising or loading events. 
    /// Do not invoke directly.
    /// </summary>
    void Apply(object @event)
    {
        switch (@event)
        {
            case Tests.Deposited e:
                Apply(e);
                break;
            case Tests.Withdrawn e:
                Apply(e);
                break;
            case Tests.Closed e:
                Apply(e);
                break;
            default:
                throw new NotSupportedException();
        }
    }

    /// <summary>
    /// Raises and applies a new event of the specified type.
    /// See <see cref="Raise{T}(T)"/>.
    /// </summary>
    void Raise<T>() where T : notnull, new() => Raise(new T());

    /// <summary>
    /// Raises and applies an event.
    /// </summary>
    void Raise<T>(T @event) where T : notnull
    {
        Apply(@event);
        (events ??= new List<object>()).Add(@event);
    }
}

注意这里也没有动态分发 💯。

此项目的一个重要特征是，如果它可以假定源生成器将在其消费中发挥作用，则库的设计及其实现细节将会有很大差异。在这种情况下，我放置了生成器后，许多设计决策与最初有所不同，结果是许多方面的简化，主库和接口项目中减少了基本类型，随着用户选择加入某些特性，还增加了增量行为。

赞助商

了解更多关于 GitHub Sponsors 的信息