Akka.NET与.NETAspire配置Skill akka-net-aspire-configuration

配置 Akka.NET 与 .NET Aspire

何时使用此技能

• 设置新的 Akka.NET 项目与 .NET Aspire 编排
• 配置 Akka.Cluster 与集群引导和发现
• 集成 Akka.Persistence 与 SQL Server
• 设置 Akka.Management 进行集群管理
• 在本地开发中配置多副本演员系统
• 将 Akka.NET 应用程序部署到 Kubernetes 与 Aspire

相关技能

• akka-net-management - Akka.Management、集群引导和发现提供者（Kubernetes、Azure、Config）的深入研究
• microsoft-extensions-configuration - IValidateOptions 模式进行配置验证
• akka-net-best-practices - 为可测试的演员系统提供集群/本地模式抽象
• aspire-integration-testing - 使用真实基础设施测试 Aspire 应用程序

核心原则

1. 通过 Microsoft.Extensions.Configuration 配置 - 使用从 appsettings.json 绑定的强类型设置类（参见 microsoft-extensions-configuration 技能）
2. Akka.Hosting 用于 DI 集成 - 使用 Akka.Hosting 库实现与 ASP.NET Core 的无缝集成
3. Aspire 用于编排 - 让 Aspire 管理服务依赖项、网络和环境配置
4. 健康检查 - 始终为集群、持久性和就绪性配置健康检查
5. 分离关注点 - 保持演员定义、配置和 Aspire 编排在不同的层
6. 在启动时验证配置 - 使用 IValidateOptions<T> 和 .ValidateOnStart() 在配置错误时快速失败

项目结构

YourSolution/
├── src/
│   ├── YourApp.Actors/              # 演员定义和业务逻辑
│   │   ├── YourActor.cs
│   │   └── YourApp.Actors.csproj
│   ├── YourApp/                     # ASP.NET Core 网络应用程序
│   │   ├── Config/
│   │   │   ├── AkkaConfiguration.cs  # Akka 设置扩展方法
│   │   │   └── AkkaSettings.cs       # 配置模型
│   │   ├── Program.cs
│   │   ├── appsettings.json
│   │   └── YourApp.csproj
│   └── YourApp.AppHost/             # Aspire 编排
│       ├── Program.cs
│       ├── AkkaManagementExtensions.cs
│       └── YourApp.AppHost.csproj

所需的 NuGet 包

演员项目（YourApp.Actors.csproj）

<ItemGroup>
  <PackageReference Include="Akka.Cluster.Hosting" />
  <PackageReference Include="Akka.Streams" />
</ItemGroup>

网络应用程序（YourApp.csproj）

<ItemGroup>
  <PackageReference Include="Akka.Hosting" />
  <PackageReference Include="Akka.Cluster.Hosting" />
  <PackageReference Include="Akka.Persistence.Sql.Hosting" />
  <PackageReference Include="Akka.Management" />
  <PackageReference Include="Akka.Management.Cluster.Bootstrap" />
  <PackageReference Include="Akka.Discovery.KubernetesApi" />
  <PackageReference Include="Akka.Discovery.Azure" />
  <PackageReference Include="Akka.Discovery.Config.Hosting" />
  <PackageReference Include="Petabridge.Cmd.Host" />
  <PackageReference Include="Petabridge.Cmd.Cluster" />
</ItemGroup>

AppHost（YourApp.AppHost.csproj）

<Sdk Name="Aspire.AppHost.Sdk" Version="$(AspireVersion)" />

<ItemGroup>
  <PackageReference Include="Aspire.Hosting.AppHost" />
  <PackageReference Include="Aspire.Hosting.Azure.Storage" />
  <PackageReference Include="Aspire.Hosting.SqlServer" />
</ItemGroup>

配置模型（AkkaSettings.cs）

创建一个强类型配置类：

using System.Net;
using System.Security.Cryptography.X509Certificates;
using Akka.Cluster.Hosting;
using Akka.Remote.Hosting;
using Petabridge.Cmd.Host;

namespace YourApp.Config;

public class AkkaSettings
{
    public string ActorSystemName { get; set; } = "YourSystem";

    public bool LogConfigOnStart { get; set; } = false;

    public RemoteOptions RemoteOptions { get; set; } = new()
    {
        PublicHostName = Dns.GetHostName(),
        HostName = "0.0.0.0",
        Port = 8081
    };

    public ClusterOptions ClusterOptions { get; set; } = new()
    {
        SeedNodes = [$"akka.tcp://YourSystem@{Dns.GetHostName()}:8081"],
        Roles = ["your-role"]
    };

    public ShardOptions ShardOptions { get; set; } = new();

    public AkkaManagementOptions? AkkaManagementOptions { get; set; }

    public PetabridgeCmdOptions PbmOptions { get; set; } = new()
    {
        Host = "0.0.0.0",
        Port = 9110
    };

    public TlsSettings? TlsSettings { get; set; }
}

public class TlsSettings
{
    public bool Enabled { get; set; } = false;
    public string? CertificatePath { get; set; }
    public string? CertificatePassword { get; set; }
    public bool ValidateCertificates { get; set; } = true;

    public X509Certificate2? LoadCertificate()
    {
        if (string.IsNullOrWhiteSpace(CertificatePath))
            return null;

        if (!File.Exists(CertificatePath))
            throw new FileNotFoundException($"Certificate file not found at: {CertificatePath}");

        return !string.IsNullOrWhiteSpace(CertificatePassword)
            ? X509CertificateLoader.LoadPkcs12FromFile(CertificatePath, CertificatePassword)
            : X509CertificateLoader.LoadCertificateFromFile(CertificatePath);
    }
}

public class AkkaManagementOptions
{
    public bool Enabled { get; set; }
    public string? Hostname { get; set; }
    public int Port { get; set; } = 8558;
    public string ServiceName { get; set; } = "your-service";
    public string PortName { get; set; } = "management";
    public int RequiredContactPointsNr { get; set; } = 1;
    public bool FilterOnFallbackPort { get; set; } = true;
    public DiscoveryMethod DiscoveryMethod { get; set; } = DiscoveryMethod.Config;
}

public enum DiscoveryMethod
{
    Config,
    Kubernetes,
    AzureTableStorage,
    AwsEcsTagBased,
    AwsEc2TagBased
}

Akka 配置扩展方法（AkkaConfiguration.cs）

using Akka.Cluster.Hosting;
using Akka.Discovery.Azure;
using Akka.Discovery.Config.Hosting;
using Akka.Discovery.KubernetesApi;
using Akka.Hosting;
using Akka.Management;
using Akka.Management.Cluster.Bootstrap;
using Akka.Persistence.Sql.Config;
using Akka.Persistence.Sql.Hosting;
using Akka.Remote.Hosting;
using LinqToDB;

namespace YourApp.Config;

public static class AkkaConfiguration
{
    public static IServiceCollection ConfigureAkka(
        this IServiceCollection services,
        IConfiguration configuration,
        Action<AkkaConfigurationBuilder, IServiceProvider> additionalConfig)
    {
        var akkaSettings = BindAkkaSettings(services, configuration);

        var connectionString = configuration.GetConnectionString("DefaultConnection");
        if (connectionString is null)
            throw new Exception("DefaultConnection ConnectionString is missing");

        const string roleName = "your-role";

        services.AddAkka(akkaSettings.ActorSystemName, (builder, provider) =>
        {
            builder.ConfigureNetwork(provider)
                .WithAkkaClusterReadinessCheck()
                .WithActorSystemLivenessCheck()
                .WithSqlPersistence(
                    connectionString: connectionString,
                    providerName: ProviderName.SqlServer2022,
                    databaseMapping: DatabaseMapping.SqlServer,
                    tagStorageMode: TagMode.TagTable,
                    deleteCompatibilityMode: true,
                    useWriterUuidColumn: true,
                    autoInitialize: true,
                    journalBuilder: journalBuilder =>
                    {
                        journalBuilder.WithHealthCheck(name: "Akka.Persistence.Sql.Journal[default]");
                    },
                    snapshotBuilder: snapshotBuilder =>
                    {
                        snapshotBuilder.WithHealthCheck(name: "Akka.Persistence.Sql.SnapshotStore[default]");
                    });

            // Add your actors here
            // Example: builder.WithActors((system, registry) => { ... });

            additionalConfig(builder, provider);
        });

        return services;
    }

    public static AkkaSettings BindAkkaSettings(IServiceCollection services, IConfiguration configuration)
    {
        var akkaSettings = new AkkaSettings();
        configuration.GetSection(nameof(AkkaSettings)).Bind(akkaSettings);
        services.AddSingleton(akkaSettings);
        return akkaSettings;
    }

    public static AkkaConfigurationBuilder ConfigureNetwork(
        this AkkaConfigurationBuilder builder,
        IServiceProvider serviceProvider)
    {
        var settings = serviceProvider.GetRequiredService<AkkaSettings>();
        var configuration = serviceProvider.GetRequiredService<IConfiguration>();

        // Apply TLS configuration if enabled
        if (settings.TlsSettings is { Enabled: true })
        {
            ConfigureRemoteOptionsWithTls(settings);
        }

        builder.WithRemoting(settings.RemoteOptions);

        if (settings.AkkaManagementOptions is { Enabled: true })
        {
            // Clear seed nodes when using Akka.Management
            var clusterOptions = settings.ClusterOptions;
            clusterOptions.SeedNodes = [];

            builder
                .WithClustering(clusterOptions)
                .WithAkkaManagement(setup =>
                {
                    setup.Http.HostName = settings.AkkaManagementOptions.Hostname?.ToLower();
                    setup.Http.Port = settings.AkkaManagementOptions.Port;
                    setup.Http.BindHostName = "0.0.0.0";
                    setup.Http.BindPort = settings.AkkaManagementOptions.Port;
                })
                .WithClusterBootstrap(options =>
                {
                    options.ContactPointDiscovery.ServiceName = settings.AkkaManagementOptions.ServiceName;
                    options.ContactPointDiscovery.PortName = settings.AkkaManagementOptions.PortName;
                    options.ContactPointDiscovery.RequiredContactPointsNr =
                        settings.AkkaManagementOptions.RequiredContactPointsNr;
                    options.ContactPointDiscovery.ContactWithAllContactPoints = true;
                    options.ContactPointDiscovery.StableMargin = TimeSpan.FromSeconds(5);
                    options.ContactPoint.FilterOnFallbackPort =
                        settings.AkkaManagementOptions.FilterOnFallbackPort;
                }, autoStart: true);

            ConfigureDiscovery(builder, settings, configuration);
        }
        else
        {
            builder.WithClustering(settings.ClusterOptions);
        }

        return builder;
    }

    private static void ConfigureDiscovery(
        AkkaConfigurationBuilder builder,
        AkkaSettings settings,
        IConfiguration configuration)
    {
        switch (settings.AkkaManagementOptions!.DiscoveryMethod)
        {
            case DiscoveryMethod.Kubernetes:
                builder.WithKubernetesDiscovery();
                break;

            case DiscoveryMethod.AzureTableStorage:
                var connectionString = configuration.GetConnectionString("AkkaManagementAzure");
                if (connectionString is null)
                    throw new Exception("AkkaManagement table storage connection string [AkkaManagementAzure] is missing");

                builder
                    .WithAzureDiscovery(options =>
                    {
                        options.ServiceName = settings.AkkaManagementOptions.ServiceName;
                        options.ConnectionString = connectionString;
                        options.HostName = settings.RemoteOptions.PublicHostName?.ToLower() ?? "localhost";
                        options.Port = settings.AkkaManagementOptions.Port;
                    })
                    .AddHocon(AzureDiscovery.DefaultConfiguration(), HoconAddMode.Append);
                break;

            case DiscoveryMethod.Config:
                builder.WithConfigDiscovery(options =>
                {
                    options.Services.Add(new Service
                    {
                        Name = settings.AkkaManagementOptions.ServiceName,
                        Endpoints =
                        [
                            $"{settings.AkkaManagementOptions.Hostname}:{settings.AkkaManagementOptions.Port}"
                        ]
                    });
                });
                break;

            default:
                throw new ArgumentOutOfRangeException();
        }
    }

    private static void ConfigureRemoteOptionsWithTls(AkkaSettings settings)
    {
        var tlsSettings = settings.TlsSettings!;
        var remoteOptions = settings.RemoteOptions;

        var certificate = tlsSettings.LoadCertificate();
        if (certificate is null)
            throw new InvalidOperationException("TLS is enabled but no certificate could be loaded");

        remoteOptions.EnableSsl = true;
        remoteOptions.Ssl = new SslOptions
        {
            X509Certificate = certificate,
            SuppressValidation = !tlsSettings.ValidateCertificates
        };

        // Update seed nodes to use akka.ssl.tcp:// protocol
        if (settings.ClusterOptions.SeedNodes?.Length > 0)
        {
            settings.ClusterOptions.SeedNodes = settings.ClusterOptions.SeedNodes
                .Select(node => node.Replace("akka.tcp://", "akka.ssl.tcp://"))
                .ToArray();
        }
    }
}

Program.cs 集成

using YourApp.Config;
using Petabridge.Cmd.Host;
using Petabridge.Cmd.Cluster;

var builder = WebApplication.CreateBuilder(args);

// Add services
builder.Services.AddRazorPages(); // or whatever your app needs

// Configure Akka.NET
builder.Services.ConfigureAkka(builder.Configuration,
    (configurationBuilder, provider) =>
    {
        var options = provider.GetRequiredService<AkkaSettings>();

        // Add Petabridge.Cmd for cluster management
        configurationBuilder.AddPetabridgeCmd(
            options: options.PbmOptions,
            hostConfiguration: cmd =>
            {
                cmd.RegisterCommandPalette(ClusterCommands.Instance);
            });
    });

var app = builder.Build();

// Configure middleware
app.MapRazorPages();
app.Run();

Aspire AppHost 配置（Program.cs）

using System.Net.Sockets;

var builder = DistributedApplication.CreateBuilder(args);

var config = builder.Configuration.GetSection("YourApp")
    .Get<YourAppConfiguration>() ?? new YourAppConfiguration();

var saPassword = builder.AddParameter(
    "sql-sa-password",
    () => "YourStrong!Passw0rd",
    secret: true);

var sqlServer = builder.AddSqlServer("sql", saPassword);

if (config.UseVolumes)
{
    sqlServer.WithDataVolume();
}

var db = sqlServer.AddDatabase("YourDb");

var app = builder.AddProject<Projects.YourApp>("yourapp")
    .WithReplicas(config.Replicas)
    .WithReference(db, "DefaultConnection")
    .ConfigureAkkaManagementForApp(config);

builder.Build().Run();

public class YourAppConfiguration
{
    public int Replicas { get; set; } = 1;
    public bool UseVolumes { get; set; } = false;
    public bool UseAkkaManagement { get; set; } = false;
}

Aspire Akka.Management 扩展（AkkaManagementExtensions.cs）

using System.Net.Sockets;
using Aspire.Hosting.Azure;

namespace YourApp.AppHost;

public static class AkkaManagementExtensions
{
    public static IResourceBuilder<ProjectResource> ConfigureAkkaManagementForApp(
        this IResourceBuilder<ProjectResource> appBuilder,
        YourAppConfiguration config)
    {
        if (!config.UseAkkaManagement) return appBuilder;

        var builder = appBuilder.ApplicationBuilder;

        // Setup Azure Table Storage for discovery
        var azureStorage = builder.AddAzureStorage("storage")
            .RunAsEmulator();

        var tableStorage = azureStorage.AddTables("akka-discovery");

        appBuilder.WaitFor(tableStorage)
            .WithReference(tableStorage, "AkkaManagementAzure");

        // Setup network endpoint ports
        appBuilder
            .WithEndpoint(name: "remote", protocol: ProtocolType.Tcp,
                env: "AkkaSettings__RemoteOptions__Port")
            .WithEndpoint(name: "management", protocol: ProtocolType.Tcp,
                env: "AkkaSettings__AkkaManagementOptions__Port")
            .WithEndpoint(name: "pbm", protocol: ProtocolType.Tcp,
                env: "AkkaSettings__PbmOptions__Port");

        // Configure Akka.Management settings via environment variables
        appBuilder
            .WithEnvironment("AkkaSettings__RemoteOptions__PublicHostName", "localhost")
            .WithEnvironment("AkkaSettings__AkkaManagementOptions__Enabled", "true")
            .WithEnvironment("AkkaSettings__AkkaManagementOptions__Hostname", "localhost")
            .WithEnvironment("AkkaSettings__AkkaManagementOptions__DiscoveryMethod", "AzureTableStorage")
            .WithEnvironment("AkkaSettings__AkkaManagementOptions__RequiredContactPointsNr",
                config.Replicas.ToString())
            .WithEnvironment("AkkaSettings__AkkaManagementOptions__FilterOnFallbackPort", "false");

        return appBuilder;
    }
}

appsettings.json 配置

{
  "ConnectionStrings": {
    "DefaultConnection": "Server=localhost;Database=YourDb;User Id=sa;Password=YourStrong!Passw0rd;"
  },
  "AkkaSettings": {
    "ActorSystemName": "YourSystem",
    "LogConfigOnStart": false,
    "RemoteOptions": {
      "PublicHostName": null,
      "HostName": "0.0.0.0",
      "Port": 8081
    },
    "ClusterOptions": {
      "Roles": ["your-role"],
      "SeedNodes": []
    },
    "PbmOptions": {
      "Host": "0.0.0.0",
      "Port": 9110
    }
  }
}

常见模式

模式 1：带依赖注入的演员注册

// 在你的演员项目中
public static class ActorRegistration
{
    public static AkkaConfigurationBuilder AddYourActor(
        this AkkaConfigurationBuilder builder,
        string roleName)
    {
        builder.WithActors((system, registry, resolver) =>
        {
            var props = resolver.Props<YourActor>();
            var actor = system.ActorOf(props, "your-actor");
            registry.Register<YourActor>(actor);
        });

        return builder;
    }
}

// 在 AkkaConfiguration.cs
builder
    .ConfigureNetwork(provider)
    .WithSqlPersistence(...)
    .AddYourActor(roleName);  // 注册你的演员

模式 2：集群分片设置

builder.WithShardRegion<YourEntityActor>(
    typeName: "your-entity",
    entityPropsFactory: (_, _, resolver) => resolver.Props<YourEntityActor>(),
    extractEntityId: ExtractEntityId,
    extractShardId: ExtractShardId,
    shardOptions: new ShardOptions
    {
        Role = "your-role",
        StateStoreMode = StateStoreMode.Persistence
    });

private static string ExtractEntityId(object message)
{
    return message switch
    {
        IEntityMessage msg => msg.EntityId,
        _ => null
    };
}

private static string ExtractShardId(object message)
{
    return message switch
    {
        IEntityMessage msg => (msg.EntityId.GetHashCode() % 10).ToString(),
        _ => null
    };
}

模式 3：健康检查

始终在 Program.cs 中配置健康检查：

builder.Services.AddHealthChecks()
    .AddCheck("self", () => HealthCheckResult.Healthy("Application is running"),
        tags: new[] { "liveness" });

// Akka 健康检查由以下自动添加：
// - .WithAkkaClusterReadinessCheck()
// - .WithActorSystemLivenessCheck()
// - journalBuilder.WithHealthCheck()
// - snapshotBuilder.WithHealthCheck()

常见问题和解决方案

问题 1：集群节点无法相互发现

症状： 节点在集群状态中保持为 “Unreachable”

解决方案：

1. 验证 RequiredContactPointsNr 是否与副本数量匹配
2. 检查所有节点是否在 AkkaManagementOptions 中使用相同的 ServiceName
3. 确保 Azure Table Storage 连接字符串正确
4. 验证防火墙/网络允许远程和管理端口上的 TCP

问题 2：持久性初始化失败

症状： 应用程序启动时出现 SQL 连接错误

解决方案：

1. 确保 SQL Server 正在运行（检查 Aspire 仪表板）
2. 验证连接字符串是否正确配置
3. 在 WithSqlPersistence 中设置 autoInitialize: true
4. 检查数据库是否存在且可访问

问题 3：开发中的脑裂

症状： 形成多个独立集群而不是一个统一集群

解决方案：

1. 在本地开发中使用 FilterOnFallbackPort = false
2. 确保所有副本使用相同的发现配置
3. 设置 ContactWithAllContactPoints = true
4. 为较慢的开发机器增加 StableMargin

测试 Akka.NET 演员

有关使用 Akka.Hosting.TestKit 进行综合 Akka.NET 测试模式的更多信息，请参阅 akka-net-testing-patterns 技能。

该技能涵盖：

• 使用 Akka.Hosting.TestKit 和依赖注入进行现代测试
• 测试 Probe 模式，用于验证演员互动
• 测试持久性演员和事件溯源
• 使用 AkkaExecutionMode.LocalTest 进行本地集群分片测试
• 基于场景的集成测试
• 最佳实践和反模式

快速示例：测试 Akka + Aspire 集成

当测试带有 Akka.NET 演员的 Aspire 应用程序时，结合使用 aspire-integration-testing 模式和 akka-net-testing-patterns：

// 使用 Aspire 的 DistributedApplicationTestingBuilder 进行基础设施测试
// 使用 Akka.Hosting.TestKit 进行演员测试
public class AkkaAspireIntegrationTests : IAsyncLifetime
{
    private DistributedApplication? _app;

    public async Task InitializeAsync()
    {
        var appHost = await DistributedApplicationTestingBuilder
            .CreateAsync<Projects.YourApp_AppHost>();

        _app = await appHost.BuildAsync();
        await _app.StartAsync();
    }

    [Fact]
    public async Task ActorSystem_WithRealDatabase_ShouldPersistEvents()
    {
        // 从 Aspire 获取 SQL 连接字符串
        var dbResource = _app!.GetResource("yourdb");
        var connectionString = await dbResource.GetConnectionStringAsync();

        // 创建 HttpClient 测试演员端点
        var httpClient = _app.CreateHttpClient("yourapp");

        // 通过 HTTP API 测试演员行为
        var response = await httpClient.PostAsJsonAsync("/orders", new
        {
            OrderId = "ORDER-001",
            Amount = 100.00m
        });

        response.Should().BeSuccessStatusCode();

        // 验证数据是否已持久化到真实数据库
        await using var connection = new SqlConnection(connectionString);
        await connection.OpenAsync();

        var events = await connection.QueryAsync<string>(
            "SELECT EventType FROM EventJournal WHERE PersistenceId = 'order-ORDER-001'");

        events.Should().Contain("OrderCreated");
    }

    public async Task DisposeAsync()
    {
        if (_app is not null)
            await _app.DisposeAsync();
    }
}

对于单元测试单个演员，使用 akka-net-testing-patterns 与内存中持久性（不需要 Aspire）。

最佳实践总结

1. 始终使用健康检查 - 为所有组件配置就绪和存活检查
2. 从配置中绑定设置 - 永远不要硬编码主机名、端口或连接字符串
3. 对于多节点使用 Akka.Management - 不要为 >1 副本的集群使用静态种子节点
4. 在生产中配置 TLS - 在生产环境中始终使用 TLS
5. 将演员逻辑与配置分离 - 保持演员纯净，配置在扩展方法中
6. 使用 Petabridge.Cmd - 对于调试和管理集群至关重要
7. 用多个副本进行测试 - 始终使用 Replicas > 1 进行测试，以捕获集群问题
8. 监控持久性健康 - 为日志和快照存储配置健康检查