akka-hosting-actor-patterns

Akka.Hosting Actor 模式

何时使用此技能

适用于以下场景：

• 构建表示领域对象（用户、订单、发票等）的实体 Actor
• 需要 Actor 同时支持单元测试（不使用集群）和生产环境（使用集群分片）
• 使用 akka-reminders 设置计划任务
• 使用 Akka.Hosting 扩展方法注册 Actor
• 创建可复用的 Actor 配置模式

核心原则

1. 执行模式抽象 - 同一份 Actor 代码可在本地（测试）或集群（生产）运行
2. 本地使用 GenericChildPerEntityParent - 无需集群开销即可模拟分片语义
3. 使用消息提取器进行路由 - 复用 Akka.Cluster.Sharding 的 IMessageExtractor 接口
4. Akka.Hosting 扩展方法 - 流畅且易于组合的配置
5. 使用 ITimeProvider 提高可测试性 - 使用 ActorSystem.Scheduler，而不是 DateTime.Now

执行模式

定义一个枚举来控制 Actor 行为：

/// <summary>
/// Determines how Akka.NET should be configured
/// </summary>
public enum AkkaExecutionMode
{
    /// <summary>
    /// Pure local actor system - no remoting, no clustering.
    /// Use GenericChildPerEntityParent instead of ShardRegion.
    /// Ideal for unit tests and simple scenarios.
    /// </summary>
    LocalTest,

    /// <summary>
    /// Full clustering with ShardRegion.
    /// Use for integration testing and production.
    /// </summary>
    Clustered
}

GenericChildPerEntityParent

一个轻量级父 Actor，可将消息路由到子实体，模拟集群分片语义，无需使用集群：

using Akka.Actor;
using Akka.Cluster.Sharding;

/// <summary>
/// A generic "child per entity" parent actor.
/// </summary>
/// <remarks>
/// Reuses Akka.Cluster.Sharding's IMessageExtractor for consistent routing.
/// Ideal for unit tests where clustering overhead is unnecessary.
/// </remarks>
public sealed class GenericChildPerEntityParent : ReceiveActor
{
    public static Props CreateProps(
        IMessageExtractor extractor,
        Func<string, Props> propsFactory)
    {
        return Props.Create(() =>
            new GenericChildPerEntityParent(extractor, propsFactory));
    }

    private readonly IMessageExtractor _extractor;
    private readonly Func<string, Props> _propsFactory;

    public GenericChildPerEntityParent(
        IMessageExtractor extractor,
        Func<string, Props> propsFactory)
    {
        _extractor = extractor;
        _propsFactory = propsFactory;

        ReceiveAny(message =>
        {
            var entityId = _extractor.EntityId(message);
            if (entityId is null) return;

            // Get existing child or create new one
            Context.Child(entityId)
                .GetOrElse(() => Context.ActorOf(_propsFactory(entityId), entityId))
                .Forward(_extractor.EntityMessage(message));
        });
    }
}

消息提取器

创建实现 Akka.Cluster.Sharding 中 IMessageExtractor 的提取器：

using Akka.Cluster.Sharding;

/// <summary>
/// Routes messages to entity actors based on a strongly-typed ID.
/// </summary>
public sealed class OrderMessageExtractor : HashCodeMessageExtractor
{
    public const int DefaultShardCount = 40;

    public OrderMessageExtractor(int maxNumberOfShards = DefaultShardCount)
        : base(maxNumberOfShards)
    {
    }

    public override string? EntityId(object message)
    {
        return message switch
        {
            IWithOrderId msg => msg.OrderId.Value.ToString(),
            _ => null
        };
    }
}

// Define an interface for messages that target a specific entity
public interface IWithOrderId
{
    OrderId OrderId { get; }
}

// Use strongly-typed IDs
public readonly record struct OrderId(Guid Value)
{
    public static OrderId New() => new(Guid.NewGuid());
    public override string ToString() => Value.ToString();
}

Akka.Hosting 扩展方法

创建用于抽象执行模式的扩展方法：

using Akka.Cluster.Hosting;
using Akka.Cluster.Sharding;
using Akka.Hosting;

public static class OrderActorHostingExtensions
{
    /// <summary>
    /// Adds OrderActor with support for both local and clustered modes.
    /// </summary>
    public static AkkaConfigurationBuilder WithOrderActor(
        this AkkaConfigurationBuilder builder,
        AkkaExecutionMode executionMode = AkkaExecutionMode.Clustered,
        string? clusterRole = null)
    {
        if (executionMode == AkkaExecutionMode.LocalTest)
        {
            // Non-clustered mode: Use GenericChildPerEntityParent
            builder.WithActors((system, registry, resolver) =>
            {
                var parent = system.ActorOf(
                    GenericChildPerEntityParent.CreateProps(
                        new OrderMessageExtractor(),
                        entityId => resolver.Props<OrderActor>(entityId)),
                    "orders");

                registry.Register<OrderActor>(parent);
            });
        }
        else
        {
            // Clustered mode: Use ShardRegion
            builder.WithShardRegion<OrderActor>(
                "orders",
                (system, registry, resolver) =>
                    entityId => resolver.Props<OrderActor>(entityId),
                new OrderMessageExtractor(),
                new ShardOptions
                {
                    StateStoreMode = StateStoreMode.DData,
                    Role = clusterRole
                });
        }

        return builder;
    }
}

组合多个 Actor

创建一个便捷方法，用于注册所有领域 Actor：

public static class DomainActorHostingExtensions
{
    /// <summary>
    /// Adds all order domain actors with sharding support.
    /// </summary>
    public static AkkaConfigurationBuilder WithOrderDomainActors(
        this AkkaConfigurationBuilder builder,
        AkkaExecutionMode executionMode = AkkaExecutionMode.Clustered,
        string? clusterRole = null)
    {
        return builder
            .WithOrderActor(executionMode, clusterRole)
            .WithPaymentActor(executionMode, clusterRole)
            .WithShipmentActor(executionMode, clusterRole)
            .WithNotificationActor(); // Singleton, no sharding needed
    }
}

使用 ITimeProvider 进行调度

将 ActorSystem 的 Scheduler 注册为 ITimeProvider，以便测试基于时间的逻辑：

public static class SharedAkkaHostingExtensions
{
    public static IServiceCollection AddAkkaWithTimeProvider(
        this IServiceCollection services,
        Action<AkkaConfigurationBuilder, IServiceProvider> configure)
    {
        // Register ITimeProvider using the ActorSystem's Scheduler
        services.AddSingleton<ITimeProvider>(sp =>
            sp.GetRequiredService<ActorSystem>().Scheduler);

        return services.ConfigureAkka((builder, sp) =>
        {
            configure(builder, sp);
        });
    }
}

// In your actor, inject ITimeProvider
public class SubscriptionActor : ReceiveActor
{
    private readonly ITimeProvider _timeProvider;

    public SubscriptionActor(ITimeProvider timeProvider)
    {
        _timeProvider = timeProvider;

        // Use _timeProvider.GetUtcNow() instead of DateTime.UtcNow
        // This allows tests to control time
    }
}

Akka.Reminders 集成

对于可在重启后继续运行的持久计划任务，请使用 akka-reminders：

using Akka.Reminders;
using Akka.Reminders.Sql;
using Akka.Reminders.Sql.Configuration;
using Akka.Reminders.Storage;

public static class ReminderHostingExtensions
{
    /// <summary>
    /// Configures akka-reminders with PostgreSQL storage.
    /// </summary>
    public static AkkaConfigurationBuilder WithPostgresReminders(
        this AkkaConfigurationBuilder builder,
        string connectionString,
        string schemaName = "reminders",
        string tableName = "scheduled_reminders",
        bool autoInitialize = true)
    {
        return builder.WithLocalReminders(reminders => reminders
            .WithResolver(sys => new GenericChildPerEntityResolver(sys))
            .WithStorage(system =>
            {
                var settings = SqlReminderStorageSettings.CreatePostgreSql(
                    connectionString,
                    schemaName,
                    tableName,
                    autoInitialize);
                return new SqlReminderStorage(settings, system);
            })
            .WithSettings(new ReminderSettings
            {
                MaxSlippage = TimeSpan.FromSeconds(30),
                MaxDeliveryAttempts = 3,
                RetryBackoffBase = TimeSpan.FromSeconds(10)
            }));
    }

    /// <summary>
    /// Configures akka-reminders with in-memory storage for testing.
    /// </summary>
    public static AkkaConfigurationBuilder WithInMemoryReminders(
        this AkkaConfigurationBuilder builder)
    {
        return builder.WithLocalReminders(reminders => reminders
            .WithResolver(sys => new GenericChildPerEntityResolver(sys))
            .WithStorage(system => new InMemoryReminderStorage())
            .WithSettings(new ReminderSettings
            {
                MaxSlippage = TimeSpan.FromSeconds(1),
                MaxDeliveryAttempts = 3,
                RetryBackoffBase = TimeSpan.FromMilliseconds(100)
            }));
    }
}

用于 Child-Per-Entity 的自定义 Reminder Resolver

将 Reminder 回调路由到 GenericChildPerEntityParent Actor：

using Akka.Actor;
using Akka.Hosting;
using Akka.Reminders;

/// <summary>
/// Resolves reminder targets to GenericChildPerEntityParent actors.
/// </summary>
public sealed class GenericChildPerEntityResolver : IReminderActorResolver
{
    private readonly ActorSystem _system;

    public GenericChildPerEntityResolver(ActorSystem system)
    {
        _system = system;
    }

    public IActorRef ResolveActorRef(ReminderEntry entry)
    {
        var registry = ActorRegistry.For(_system);

        return entry.Key switch
        {
            var k when k.StartsWith("order-") =>
                registry.Get<OrderActor>(),
            var k when k.StartsWith("subscription-") =>
                registry.Get<SubscriptionActor>(),
            _ => throw new InvalidOperationException(
                $"Unknown reminder key format: {entry.Key}")
        };
    }
}

单例 Actor（不分片）

对于只应有一个实例的 Actor：

public static AkkaConfigurationBuilder WithEmailSenderActor(
    this AkkaConfigurationBuilder builder)
{
    return builder.WithActors((system, registry, resolver) =>
    {
        var actor = system.ActorOf(
            resolver.Props<EmailSenderActor>(),
            "email-sender");
        registry.Register<EmailSenderActor>(actor);
    });
}

用于注册表的标记类型

当需要引用已注册为父 Actor 的 Actor 时：

/// <summary>
/// Marker type for ActorRegistry to retrieve the order manager
/// (GenericChildPerEntityParent for OrderActors).
/// </summary>
public sealed class OrderManagerActor;

// Usage in extension method
registry.Register<OrderManagerActor>(parent);

// Usage in controller/service
public class OrderService
{
    private readonly IActorRef _orderManager;

    public OrderService(IRequiredActor<OrderManagerActor> orderManager)
    {
        _orderManager = orderManager.ActorRef;
    }

    public async Task<OrderResponse> CreateOrder(CreateOrderCommand cmd)
    {
        return await _orderManager.Ask<OrderResponse>(cmd);
    }
}

Actor 中的 DI 作用域管理

Actor 不会自动创建 DI 作用域。 与 ASP.NET 控制器（每个 HTTP 请求都会创建一个作用域）不同，Actor 的生命周期很长。如果需要使用作用域服务（例如 DbContext），请注入 IServiceProvider 并手动创建作用域。

模式：每条消息一个作用域

public sealed class OrderProcessingActor : ReceiveActor
{
    private readonly IServiceProvider _serviceProvider;
    private readonly IActorRef _notificationActor;

    public OrderProcessingActor(
        IServiceProvider serviceProvider,
        IRequiredActor<NotificationActor> notificationActor)
    {
        _serviceProvider = serviceProvider;
        _notificationActor = notificationActor.ActorRef;

        ReceiveAsync<ProcessOrder>(HandleProcessOrder);
    }

    private async Task HandleProcessOrder(ProcessOrder msg)
    {
        // Create scope for this message - disposed after processing
        using var scope = _serviceProvider.CreateScope();

        // Resolve scoped services within the scope
        var orderRepository = scope.ServiceProvider.GetRequiredService<IOrderRepository>();
        var paymentService = scope.ServiceProvider.GetRequiredService<IPaymentService>();
        var emailComposer = scope.ServiceProvider.GetRequiredService<IOrderEmailComposer>();

        // Do work with scoped services
        var order = await orderRepository.GetByIdAsync(msg.OrderId);
        var payment = await paymentService.ProcessAsync(order);

        // DbContext changes committed when scope disposes
    }
}

使用此模式的原因

优势	说明
每条消息使用全新的 DbContext	避免消息之间残留过时的实体跟踪
正确释放资源	每条消息处理完毕后释放数据库连接
隔离性	一条消息的错误不会破坏另一条消息的状态
可测试	可以在测试中注入模拟的 IServiceProvider

单例服务 - 直接注入

对于无状态且线程安全的服务，可直接注入（无需作用域）：

public sealed class NotificationActor : ReceiveActor
{
    private readonly IEmailLinkGenerator _linkGenerator;  // Singleton - OK!
    private readonly IMjmlTemplateRenderer _renderer;     // Singleton - OK!

    public NotificationActor(
        IEmailLinkGenerator linkGenerator,
        IMjmlTemplateRenderer renderer)
    {
        _linkGenerator = linkGenerator;
        _renderer = renderer;

        Receive<SendWelcomeEmail>(Handle);
    }
}

常见错误：直接注入作用域服务

// BAD: Scoped service injected into long-lived actor
public sealed class BadActor : ReceiveActor
{
    private readonly IOrderRepository _repo;  // Scoped! DbContext lives forever!

    public BadActor(IOrderRepository repo)  // Captured at actor creation
    {
        _repo = repo;  // This DbContext will become stale
    }
}

// GOOD: Inject IServiceProvider, create scope per message
public sealed class GoodActor : ReceiveActor
{
    private readonly IServiceProvider _sp;

    public GoodActor(IServiceProvider sp)
    {
        _sp = sp;
        ReceiveAsync<ProcessOrder>(async msg =>
        {
            using var scope = _sp.CreateScope();
            var repo = scope.ServiceProvider.GetRequiredService<IOrderRepository>();
            // Fresh DbContext for this message
        });
    }
}

有关 DI 生命周期和作用域管理的更多信息，请参阅 microsoft-extensions/dependency-injection 技能。

---

集群分片配置

RememberEntities：几乎总是设为 False

RememberEntities 用于控制分片区域是否记住并自动重启所有曾创建的实体。此选项几乎总是应设为 false。

builder.WithShardRegion<OrderActor>(
    "orders",
    (system, registry, resolver) => entityId => resolver.Props<OrderActor>(entityId),
    new OrderMessageExtractor(),
    new ShardOptions
    {
        StateStoreMode = StateStoreMode.DData,
        RememberEntities = false,  // DEFAULT - almost always correct
        Role = clusterRole
    });

RememberEntities = true 会导致的问题：

问题	说明
内存无限增长	曾创建的每个实体都会被记住并不断重启
集群启动缓慢	集群启动时必须重启数千甚至数百万个实体
过时实体复活	过期会话、已发送的邮件、旧订单都会被重启
无法停用实体	空闲实体会无限期占用内存（停用功能会被禁用）

何时使用各项设置

实体类型	RememberEntities	原因
UserSessionActor	false	会话会过期，登录时创建
DraftActor	false	草稿会被发送或丢弃，生命周期短暂
EmailSenderActor	false	即发即弃操作
OrderActor	false	订单会完成，并持续创建新订单
ShoppingCartActor	false	购物车会过期，弃置购物车很常见
TenantActor	maybe true	租户集合固定，始终需要
AccountActor	maybe true	账户集合有界，生命周期长

经验法则： 仅在以下情况下使用 RememberEntities = true：

1. 实体集合有界（已知上限）
2. 生命周期长且应始终可用的领域实体
3. 记住实体的成本低于延迟创建的成本

使用 WithShardRegion 的标记类型

使用 WithShardRegion<T> 时，泛型参数 T 充当 ActorRegistry 的标记类型。请使用专用标记类型（而非 Actor 类本身），以便一致地访问注册表：

/// <summary>
/// Marker type for ActorRegistry. Use this to retrieve the OrderActor shard region.
/// </summary>
public sealed class OrderActorRegion;

// Registration - use marker type as generic parameter
builder.WithShardRegion<OrderActorRegion>(
    "orders",
    (system, registry, resolver) => entityId => resolver.Props<OrderActor>(entityId),
    new OrderMessageExtractor(),
    new ShardOptions { StateStoreMode = StateStoreMode.DData });

// Retrieval - same marker type
var orderRegion = ActorRegistry.Get<OrderActorRegion>();
orderRegion.Tell(new CreateOrder(orderId, amount));

为何使用标记类型？

• WithShardRegion<T> 会自动将分片区域注册为类型 T
• 直接使用 Actor 类可能造成混淆（注册表返回的是区域，而非 Actor）
• 标记类型能明确表达意图，并在 LocalTest 和 Clustered 模式下保持一致

避免重复调用注册表

WithShardRegion<T> 会自动将分片区域注册到 ActorRegistry。请勿再次调用 registry.Register<T>()：

// BAD - redundant registration
builder.WithShardRegion<OrderActorRegion>("orders", ...)
    .WithActors((system, registry, resolver) =>
    {
        var region = registry.Get<OrderActorRegion>();
        registry.Register<OrderActorRegion>(region);  // UNNECESSARY!
    });

// GOOD - WithShardRegion already registers
builder.WithShardRegion<OrderActorRegion>("orders", ...);
// That's it - OrderActorRegion is now in the registry

---

最佳实践

1. 始终支持两种执行模式 - 无需修改代码即可轻松测试
2. 使用强类型 ID - 使用 OrderId，而非 string 或 Guid
3. 基于接口的消息路由 - 使用 IWithOrderId 进行类型安全的提取
4. 注册父级，而非子级 - 对于每实体一个子 Actor 的模式，在 ActorRegistry 中注册父级
5. 使用标记类型以明确意图 - 使用空标记类进行注册表查找
6. 组合优于继承 - 链式调用扩展方法，不要创建深层继承层次
7. 使用 ITimeProvider 进行调度 - 切勿在 Actor 中直接使用 DateTime.Now
8. 使用 akka-reminders 确保持久性 - 用于必须在重启后继续运行的计划任务
9. 默认使用 RememberEntities = false - 仅对有界且生命周期长的实体设为 true