C# и микросервисы: Распределённые системы на .NET

public async Task<Product> GetProductAsync(int id)
{
    using var client = new HttpClient();
    var response = await client.GetAsync($"http://product-service/products/{id}");
    
    if (!response.IsSuccessStatusCode)
        return null;
        
    var content = await response.Content.ReadAsStringAsync();
    return JsonConvert.DeserializeObject<Product>(content);
}

var channel = GrpcChannel.ForAddress("https://product-service:5001");
var client = new ProductService.ProductServiceClient(channel);
var response = await client.GetProductAsync(new GetProductRequest { Id = 42 });

public class OrderContext : DbContext
{
    public DbSet<Order> Orders { get; set; }
    
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<Order>().ToTable("Orders", schema: "ordering");
        // Конфигурация конкретно для этого микросервиса
    }
}

// REST взаимодействие
public class CatalogService
{
    private readonly HttpClient _httpClient;
    
    public CatalogService(HttpClient httpClient)
    {
        _httpClient = httpClient;
        // Настройка базового адреса, заголовков и т.д.
    }
    
    public async Task<Product> GetProductAsync(int id)
    {
        var response = await _httpClient.GetAsync($"/api/products/{id}");
        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<Product>();
    }
}

// Публикация события с использованием MassTransit
public class OrderProcessor
{
    private readonly IBus _bus;
    
    public OrderProcessor(IBus bus)
    {
        _bus = bus;
    }
    
    public async Task ProcessOrderAsync(Order order)
    {
        // Обработка заказа
        
        // Публикация события о том, что заказ обработан
        await _bus.Publish(new OrderProcessedEvent
        {
            OrderId = order.Id,
            CustomerId = order.CustomerId,
            TotalAmount = order.TotalAmount
        });
    }
}

// Конфигурация Ocelot в ASP.NET Core
public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddOcelot(Configuration);
    }
    
    public void Configure(IApplicationBuilder app)
    {
        app.UseOcelot().Wait();
    }
}

var policy = Policy
    .Handle<HttpRequestException>()
    .CircuitBreakerAsync(
        exceptionsAllowedBeforeBreaking: 2,
        durationOfBreak: TimeSpan.FromMinutes(1)
    );
    
// Использование
await policy.ExecuteAsync(async () => 
{
    var response = await httpClient.GetAsync("http://service-url");
    return await response.Content.ReadAsStringAsync();
});

// Пример команды в CQRS
public class CreateOrderCommand : IRequest<bool>
{
    public int CustomerId { get; set; }
    public List<OrderItem> Items { get; set; }
    public string ShippingAddress { get; set; }
}
 
// Обработчик команды
public class CreateOrderCommandHandler : IRequestHandler<CreateOrderCommand, bool>
{
    private readonly IOrderRepository _repository;
    
    public CreateOrderCommandHandler(IOrderRepository repository)
    {
        _repository = repository;
    }
    
    public async Task<bool> Handle(CreateOrderCommand command, CancellationToken token)
    {
        var order = new Order(command.CustomerId, command.ShippingAddress);
        order.AddItems(command.Items);
        
        return await _repository.SaveOrderAsync(order, token);
    }
}

public class ServiceDiscoveryMessageHandler : DelegatingHandler
{
    private readonly IServiceRegistry _registry;
    
    public ServiceDiscoveryMessageHandler(IServiceRegistry registry)
    {
        _registry = registry;
    }
    
    protected override async Task<HttpResponseMessage> SendAsync(
        HttpRequestMessage request, CancellationToken cancellationToken)
    {
        var serviceName = // извлекаем имя сервиса из URL
        var instances = await _registry.GetServiceInstancesAsync(serviceName);
        var instance = instances.RandomInstance(); // простейшая стратегия
        
        // Меняем URL на конкретный экземпляр
        request.RequestUri = new Uri($"http://{instance.Host}:{instance.Port}{request.RequestUri.PathAndQuery}");
        
        return await base.SendAsync(request, cancellationToken);
    }
}

public void ConfigureServices(IServiceCollection services)
{
    // Регистрируем сервисы с разными жизненными циклами
    services.AddSingleton<IServiceDiscovery, ConsulServiceDiscovery>();
    services.AddScoped<IOrderRepository, OrderRepository>();
    services.AddTransient<IPaymentProcessor, PaymentProcessor>();
    
    // Типичная конфигурация для микросервиса
    services.AddControllers();
    services.AddHealthChecks();
    services.AddSwaggerGen();
}

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();
 
// Определяем API эндпоинты
app.MapGet("/api/products/{id}", async (int id, IProductRepository repo) => 
    await repo.GetByIdAsync(id) is Product product
        ? Results.Ok(product)
        : Results.NotFound());
 
app.Run();

// Определение интерфейса сервиса в .proto файле
syntax = "proto3";
 
service ProductCatalog {
  rpc GetProduct (GetProductRequest) returns (ProductModel);
}
 
message GetProductRequest {
  int32 product_id = 1;
}
 
message ProductModel {
  int32 id = 1;
  string name = 2;
  double price = 3;
}
 
// Использование в клиентском коде
var channel = GrpcChannel.ForAddress("https://product-service");
var client = new ProductCatalog.ProductCatalogClient(channel);
var product = await client.GetProductAsync(new GetProductRequest { ProductId = 42 });

// Публикация сообщения через MassTransit и RabbitMQ
public class OrderProcessingService
{
    private readonly IBus _bus;
    
    public OrderProcessingService(IBus bus)
    {
        _bus = bus;
    }
    
    public async Task ProcessOrderAsync(Order order)
    {
        // Обработка заказа
        await _bus.Publish(new OrderProcessedEvent
        {
            OrderId = order.Id,
            ProcessedAt = DateTime.UtcNow
        });
    }
}
 
// Настройка MassTransit с RabbitMQ
services.AddMassTransit(x =>
{
    x.AddConsumer<OrderProcessedConsumer>();
    
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");
        cfg.ReceiveEndpoint("order-processing", e =>
        {
            e.ConfigureConsumer<OrderProcessedConsumer>(context);
        });
    });
});

// Отправка сообщения в Azure Service Bus
var client = new ServiceBusClient("connection-string");
var sender = client.CreateSender("queue-name");
var message = new ServiceBusMessage(Encoding.UTF8.GetBytes(jsonMessage))
{
    MessageId = Guid.NewGuid().ToString(),
    ContentType = "application/json"
};
await sender.SendMessageAsync(message);

var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
using var producer = new ProducerBuilder<Null, string>(config).Build();
var result = await producer.ProduceAsync("topic-name", new Message<Null, string> { Value = "test message" });

// ocelot.json
{
  "Routes": [
    {
      "DownstreamPathTemplate": "/api/products/{id}",
      "DownstreamScheme": "https",
      "DownstreamHostAndPorts": [
        {
          "Host": "product-service",
          "Port": 5001
        }
      ],
      "UpstreamPathTemplate": "/products/{id}",
      "UpstreamHttpMethod": [ "Get" ]
    }
  ],
  "GlobalConfiguration": {
    "BaseUrl": "https://api.mycompany.com"
  }
}

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddReverseProxy()
    .LoadFromConfig(builder.Configuration.GetSection("ReverseProxy"));
    
var app = builder.Build();
app.MapReverseProxy();
app.Run();

FROM mcr.microsoft.com/dotnet/aspnet:6.0 AS base
WORKDIR /app
EXPOSE 80
 
FROM mcr.microsoft.com/dotnet/sdk:6.0 AS build
WORKDIR /src
COPY ["MyService.csproj", "./"]
RUN dotnet restore "MyService.csproj"
COPY . .
RUN dotnet build "MyService.csproj" -c Release -o /app/build
 
FROM build AS publish
RUN dotnet publish "MyService.csproj" -c Release -o /app/publish
 
FROM base AS final
WORKDIR /app
COPY --from=publish /app/publish .
ENTRYPOINT ["dotnet", "MyService.dll"]

// Startup.cs
public void ConfigureServices(IServiceCollection services)
{
    services.AddHealthChecks()
        .AddCheck("db_connection", () => 
        {
            // Проверка подключения к базе данных
            return HealthCheckResult.Healthy();
        })
        .AddCheck<ExternalServiceHealthCheck>("external_api");
}
 
public void Configure(IApplicationBuilder app)
{
    app.UseHealthChecks("/health/live", new HealthCheckOptions
    {
        Predicate = _ => true
    });
}

public async Task ProcessPaymentAsync(PaymentMessage message)
{
    // Проверяем, не обрабатывали ли мы уже этот платеж
    var processingId = $"payment-{message.PaymentId}";
    var lockTaken = await _distributedLock.TryAcquireAsync(processingId, TimeSpan.FromMinutes(5));
    
    if (!lockTaken)
    {
        _logger.LogWarning("Payment {0} is being processed by another instance", message.PaymentId);
        return; // Выходим, т.к. другой экземпляр уже обрабатывает платеж
    }
    
    try
    {
        // Проверяем, был ли уже обработан этот платеж
        if (await _paymentRepository.ExistsAsync(message.PaymentId))
        {
            _logger.LogInfo("Payment {0} already processed, skipping", message.PaymentId);
            return;
        }
        
        // Основная бизнес-логика обработки платежа
        await ProcessPaymentInternalAsync(message);
    }
    finally
    {
        await _distributedLock.ReleaseAsync(processingId);
    }
}

// Типичная структура приложения:
// Catalog.API/
// Ordering.API/
// Basket.API/
// Identity.API/
// ApiGateways/
// BuildingBlocks/

// Program.cs для современного .NET 6+
var builder = WebApplication.CreateBuilder(args);
 
// Регистрация зависимостей
builder.Services.AddControllers();
builder.Services.AddDbContext<OrderContext>(options =>
    options.UseSqlServer(builder.Configuration.GetConnectionString("OrderingDb")));
builder.Services.AddScoped<IOrderRepository, OrderRepository>();
builder.Services.AddScoped<IOrderService, OrderService>();
builder.Services.AddHealthChecks()
    .AddSqlServer(builder.Configuration.GetConnectionString("OrderingDb"));
 
// Настройка OpenTelemetry для трассировки
builder.Services.AddOpenTelemetryTracing(b => {
    b.AddAspNetCoreInstrumentation()
     .AddSqlClientInstrumentation()
     .AddHttpClientInstrumentation()
     .AddJaegerExporter();
});
 
var app = builder.Build();
 
// Конфигурация middleware
app.UseRouting();
app.UseAuthentication();
app.UseAuthorization();
app.MapControllers();
app.MapHealthChecks("/health");
 
app.Run();

// Настройка HttpClient с политиками отказоустойчивости
services.AddHttpClient<ICatalogService, CatalogService>()
    .AddPolicyHandler(GetRetryPolicy())
    .AddPolicyHandler(GetCircuitBreakerPolicy());
 
// Политика повторных попыток
static IAsyncPolicy<HttpResponseMessage> GetRetryPolicy()
{
    return HttpPolicyExtensions
        .HandleTransientHttpError()
        .OrResult(msg => msg.StatusCode == System.Net.HttpStatusCode.NotFound)
        .WaitAndRetryAsync(3, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)));
}
 
// Политика размыкателя цепи
static IAsyncPolicy<HttpResponseMessage> GetCircuitBreakerPolicy()
{
    return HttpPolicyExtensions
        .HandleTransientHttpError()
        .CircuitBreakerAsync(5, TimeSpan.FromMinutes(1));
}

// Настройка OpenTelemetry в ASP.NET Core
services.AddOpenTelemetryTracing(builder =>
{
    builder
        .SetResourceBuilder(ResourceBuilder.CreateDefault()
            .AddService("OrderingService"))
        .AddAspNetCoreInstrumentation()
        .AddHttpClientInstrumentation()
        .AddEntityFrameworkCoreInstrumentation()
        .AddJaegerExporter(options =>
        {
            options.AgentHost = "jaeger";
            options.AgentPort = 6831;
        });
});
 
// Использование в коде для создания пользовательских спанов
public async Task<OrderDto> ProcessOrderAsync(OrderDto orderDto)
{
    using var activity = Activity.StartActivity("ProcessOrder", ActivityKind.Internal);
    activity?.SetTag("orderId", orderDto.Id);
    
    // Бизнес-логика обработки заказа
    
    activity?.SetStatus(ActivityStatusCode.Ok);
    return orderDto;
}

// Определение саги
public class OrderSaga : MassTransitStateMachine<OrderState>
{
    public OrderSaga()
    {
        InstanceState(x => x.CurrentState);
        
        Event(() => OrderSubmitted, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => PaymentProcessed, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => StockReserved, x => x.CorrelateById(m => m.Message.OrderId));
        Event(() => OrderShipped, x => x.CorrelateById(m => m.Message.OrderId));
        
        // Определение состояний и переходов
        Initially(
            When(OrderSubmitted)
                .Then(context => {
                    context.Instance.OrderId = context.Data.OrderId;
                    context.Instance.CustomerId = context.Data.CustomerId;
                })
                .TransitionTo(Submitted)
                .Publish(context => new ProcessPaymentCommand { OrderId = context.Instance.OrderId })
        );
        
        During(Submitted,
            When(PaymentProcessed)
                .TransitionTo(PaymentCompleted)
                .Publish(context => new ReserveStockCommand { OrderId = context.Instance.OrderId })
        );
        
        // Обработка отказов и компенсирующие действия
        During(Submitted,
            When(PaymentFailed)
                .TransitionTo(Faulted)
                .Publish(context => new CancelOrderCommand { OrderId = context.Instance.OrderId })
        );
    }
    
    // Состояния
    public State Submitted { get; private set; }
    public State PaymentCompleted { get; private set; }
    public State StockReserved { get; private set; }
    public State Completed { get; private set; }
    public State Faulted { get; private set; }
    
    // События
    public Event<OrderSubmittedEvent> OrderSubmitted { get; private set; }
    public Event<PaymentProcessedEvent> PaymentProcessed { get; private set; }
    public Event<PaymentFailedEvent> PaymentFailed { get; private set; }
    public Event<StockReservedEvent> StockReserved { get; private set; }
    public Event<OrderShippedEvent> OrderShipped { get; private set; }
}

[Fact]
public async Task GetOrderById_ShouldReturnOrder_WhenOrderExists()
{
    // Arrange
    var mockRepo = new Mock<IOrderRepository>();
    mockRepo.Setup(r => r.GetByIdAsync(1))
            .ReturnsAsync(new Order { Id = 1, Status = OrderStatus.Pending });
    
    var controller = new OrdersController(mockRepo.Object);
    
    // Act
    var result = await controller.GetOrderAsync(1);
    
    // Assert
    var okResult = Assert.IsType<OkObjectResult>(result.Result);
    var order = Assert.IsType<Order>(okResult.Value);
    Assert.Equal(1, order.Id);
}

public class OrderApiIntegrationTests : IClassFixture<WebApplicationFactory<Startup>>
{
    private readonly WebApplicationFactory<Startup> _factory;
    
    [Fact]
    public async Task CreateOrder_ReturnsCreatedResponse()
    {
        // Запускаем API с тестовой базой данных
        var client = _factory.CreateClient();
        
        // Отправляем реальный HTTP-запрос
        var response = await client.PostAsJsonAsync("/api/orders", 
            new CreateOrderRequest { /*...*/ });
        
        Assert.Equal(HttpStatusCode.Created, response.StatusCode);
    }
}

[FunctionName("ProcessOrder")]
public static async Task<IActionResult> Run(
    [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequest req,
    [CosmosDB("orders", "items", ConnectionStringSetting = "CosmosDbConnection")] IAsyncCollector<Order> orderCollector,
    ILogger log)
{
    string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
    var order = JsonConvert.DeserializeObject<Order>(requestBody);
    
    await orderCollector.AddAsync(order);
    
    return new OkObjectResult($"Order {order.Id} processed successfully");
}

// Это уходит в прошлое благодаря Service Mesh
services.AddAuthentication("Bearer")
    .AddJwtBearer(...);
    
services.AddAuthorization(...);
 
// Вместо этого настраиваем сервис-меш единожды на уровне кластера

// Настройка GraphQL-сервера
services.AddGraphQLServer()
    .AddQueryType<Query>()
    .AddMutationType<Mutation>()
    .AddSubscriptionType<Subscription>()
    .AddDataLoader<ProductByIdDataLoader>()
    .AddFiltering()
    .AddSorting();

// Команда для создания платежа
public class CreatePaymentCommand : IRequest<Guid>
{
    public decimal Amount { get; set; }
    public string Currency { get; set; }
    public Guid SenderId { get; set; }
    public Guid RecipientId { get; set; }
}
 
// Обработчик команды, публикующий событие
public class CreatePaymentCommandHandler : IRequestHandler<CreatePaymentCommand, Guid>
{
    private readonly IEventStore _eventStore;
    
    public CreatePaymentCommandHandler(IEventStore eventStore)
    {
        _eventStore = eventStore;
    }
    
    public async Task<Guid> Handle(CreatePaymentCommand command, CancellationToken token)
    {
        var paymentId = Guid.NewGuid();
        
        // Создаем событие
        var paymentCreatedEvent = new PaymentCreatedEvent(
            paymentId,
            command.Amount,
            command.Currency,
            command.SenderId,
            command.RecipientId,
            DateTime.UtcNow);
            
        // Сохраняем событие в хранилище
        await _eventStore.AppendEventAsync("payment", paymentId, paymentCreatedEvent);
        
        return paymentId;
    }
}

public class OrderService
{
    private readonly OrderDbContext _dbContext;
    
    public async Task CreateOrderAsync(Order order)
    {
        using var transaction = await _dbContext.Database.BeginTransactionAsync();
        
        try
        {
            // Добавляем заказ в БД
            _dbContext.Orders.Add(order);
            
            // Добавляем событие в таблицу исходящих сообщений в той же БД
            _dbContext.OutboxMessages.Add(new OutboxMessage
            {
                Id = Guid.NewGuid(),
                Type = "OrderCreated",
                Content = JsonSerializer.Serialize(new OrderCreatedEvent
                {
                    OrderId = order.Id,
                    CustomerId = order.CustomerId,
                    TotalAmount = order.TotalAmount,
                    CreatedAt = DateTime.UtcNow
                }),
                Status = OutboxMessageStatus.Ready,
                CreatedAt = DateTime.UtcNow
            });
            
            await _dbContext.SaveChangesAsync();
            await transaction.CommitAsync();
        }
        catch (Exception)
        {
            await transaction.RollbackAsync();
            throw;
        }
    }
}