Event-Driven архитектуры в C# - выбираем правильного брокера сообщений

[HttpPost]
public async Task<IActionResult> CreateOrder(OrderRequest request)
{
    // Проверка наличия товаров
    var inventoryResult = await _inventoryClient.CheckAvailability(request.Items);
    if (!inventoryResult.IsSuccess)
        return BadRequest("Товары недоступны");
    
    // Расчет доставки
    var shippingResult = await _shippingClient.CalculateShipping(request.Address, request.Items);
    if (!shippingResult.IsSuccess)
        return BadRequest("Невозможно рассчитать доставку");
    
    // Создание заказа
    var order = await _orderRepository.CreateOrder(request, shippingResult.Cost);
    
    // Резервирование товаров
    await _inventoryClient.ReserveItems(order.Id, request.Items);
    
    return Ok(order);
}

[HttpPost]
public async Task<IActionResult> InitiateOrder(OrderRequest request)
{
    var orderId = Guid.NewGuid();
    
    // Публикуем событие инициации заказа
    await _eventBus.PublishAsync(new OrderInitiatedEvent
    {
        OrderId = orderId,
        CustomerId = request.CustomerId,
        Items = request.Items,
        ShippingAddress = request.Address,
        Timestamp = DateTime.UtcNow
    });
    
    return Accepted(new { OrderId = orderId, Status = "Processing" });
}

public class OrderProcessingConsumer : IConsumer<OrderInitiatedEvent>
{
    public async Task Consume(ConsumeContext<OrderInitiatedEvent> context)
    {
        var message = context.Message;
        
        // Приоритизация VIP-клиентов
        if (await _customerRepository.IsVipCustomer(message.CustomerId))
        {
            // Устанавливаем высокий приоритет для дальнейшей обработки
            await context.Forward(new Uri("queue:vip-order-processing"));
            return;
        }
        
        // Стандартная обработка для обычных клиентов
        // ...
    }
}

public async Task Consume(ConsumeContext<ReserveInventoryCommand> context)
{
    var message = context.Message;
    
    try
    {
        var result = await _inventoryService.ReserveItems(message.OrderId, message.Items);
        
        if (result.Success)
        {
            await context.Publish(new InventoryReservedEvent
            {
                OrderId = message.OrderId,
                ReservationId = result.ReservationId
            });
        }
        else
        {
            // Публикуем событие о невозможности резервации
            await context.Publish(new InventoryReservationFailedEvent
            {
                OrderId = message.OrderId,
                Reason = result.FailureReason
            });
        }
    }
    catch (Exception ex)
    {
        // В случае исключения тоже публикуем событие об ошибке
        await context.Publish(new InventoryReservationFailedEvent
        {
            OrderId = message.OrderId,
            Reason = $"Exception: {ex.Message}"
        });
        
        // Перебрасываем исключение, чтобы сообщение не считалось обработанным
        throw;
    }
}

public async Task Consume(ConsumeContext<InventoryReservationFailedEvent> context)
{
    var message = context.Message;
    
    // Изменяем статус заказа
    await _orderRepository.UpdateOrderStatus(
        message.OrderId, 
        OrderStatus.Failed, 
        $"Не удалось зарезервировать товары: {message.Reason}"
    );
    
    // Уведомляем пользователя
    await _notificationService.NotifyOrderFailed(message.OrderId, message.Reason);
}

// Команда (операция записи)
public async Task<Guid> CreateOrder(CreateOrderCommand command)
{
    var orderId = Guid.NewGuid();
    
    // Сохраняем заказ в БД команд
    await _commandDbContext.Orders.AddAsync(new OrderEntity
    {
        Id = orderId,
        CustomerId = command.CustomerId,
        Status = OrderStatus.Pending,
        // другие поля
    });
    await _commandDbContext.SaveChangesAsync();
    
    // Публикуем событие
    await _eventBus.PublishAsync(new OrderCreatedEvent
    {
        OrderId = orderId,
        CustomerId = command.CustomerId,
        Items = command.Items,
        // другие поля
    });
    
    return orderId;
}
 
// Обработчик события для обновления БД для чтения
public async Task HandleAsync(OrderCreatedEvent @event)
{
    await _readDbContext.OrderReadModels.AddAsync(new OrderReadModel
    {
        Id = @event.OrderId,
        CustomerId = @event.CustomerId,
        Status = OrderStatus.Pending,
        // преобразование других полей
    });
    await _readDbContext.SaveChangesAsync();
}
 
// Запрос (операция чтения)
public async Task<OrderReadModel> GetOrderById(Guid orderId)
{
    return await _readDbContext.OrderReadModels
        .FirstOrDefaultAsync(o => o.Id == orderId);
}

public async Task CreateOrder(OrderDto orderDto)
{
    using var transaction = await _dbContext.Database.BeginTransactionAsync();
    
    try
    {
        // Создаем заказ
        var order = new Order
        {
            Id = Guid.NewGuid(),
            CustomerId = orderDto.CustomerId,
            Items = orderDto.Items.Select(i => new OrderItem { /* ... */ }).ToList(),
            Status = OrderStatus.Created,
            CreatedAt = DateTime.UtcNow
        };
        
        _dbContext.Orders.Add(order);
        
        // Создаем запись в outbox
        var orderCreatedEvent = new OrderCreatedEvent
        {
            OrderId = order.Id,
            CustomerId = order.CustomerId,
            Items = order.Items.Select(i => new OrderItemDto { /* ... */ }).ToList(),
            CreatedAt = order.CreatedAt
        };
        
        _dbContext.OutboxMessages.Add(new OutboxMessage
        {
            Id = Guid.NewGuid(),
            MessageType = nameof(OrderCreatedEvent),
            Content = JsonSerializer.Serialize(orderCreatedEvent),
            CreatedAt = DateTime.UtcNow,
            ProcessedAt = null
        });
        
        await _dbContext.SaveChangesAsync();
        await transaction.CommitAsync();
    }
    catch
    {
        await transaction.RollbackAsync();
        throw;
    }
}

public class OutboxProcessor : BackgroundService
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<OutboxProcessor> _logger;
    private readonly IMessageBroker _messageBroker;
    
    public OutboxProcessor(
        IServiceScopeFactory scopeFactory, 
        ILogger<OutboxProcessor> logger,
        IMessageBroker messageBroker)
    {
        _scopeFactory = scopeFactory;
        _logger = logger;
        _messageBroker = messageBroker;
    }
    
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ProcessOutboxMessages(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка при обработке сообщений из outbox");
            }
            
            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
        }
    }
    
    private async Task ProcessOutboxMessages(CancellationToken stoppingToken)
    {
        using var scope = _scopeFactory.CreateScope();
        var dbContext = scope.ServiceProvider.GetRequiredService<AppDbContext>();
        
        // Получаем пакет необработанных сообщений
        var messages = await dbContext.OutboxMessages
            .Where(m => m.ProcessedAt == null)
            .OrderBy(m => m.CreatedAt)
            .Take(20)
            .ToListAsync(stoppingToken);
        
        foreach (var message in messages)
        {
            try
            {
                // Отправляем сообщение в брокер
                await _messageBroker.PublishAsync(
                    message.MessageType, 
                    message.Content, 
                    stoppingToken);
                
                // Помечаем как обработанное
                message.ProcessedAt = DateTime.UtcNow;
                await dbContext.SaveChangesAsync(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка при обработке сообщения {MessageId}", message.Id);
                break; // Прерываем обработку пакета при ошибке
            }
        }
    }
}

public class RedisDistributedLock : IDistributedLock
{
    private readonly IConnectionMultiplexer _redis;
    private readonly string _lockKey;
    private readonly string _lockValue;
    private readonly TimeSpan _lockTimeout;
 
    public RedisDistributedLock(IConnectionMultiplexer redis, string resourceId)
    {
        _redis = redis;
        _lockKey = $"lock:{resourceId}";
        _lockValue = Guid.NewGuid().ToString();
        _lockTimeout = TimeSpan.FromSeconds(30);
    }
 
    public async Task<bool> AcquireLockAsync(CancellationToken token = default)
    {
        var db = _redis.GetDatabase();
        return await db.StringSetAsync(
            _lockKey,
            _lockValue,
            _lockTimeout,
            When.NotExists);
    }
 
    public async Task ReleaseLockAsync()
    {
        var db = _redis.GetDatabase();
        
        // Скрипт Lua для безопасного освобождения блокировки
        // Проверяем, что значение не изменилось (блокировка наша)
        var script = @"
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end";
        
        await db.ScriptEvaluateAsync(
            script,
            new RedisKey[] { _lockKey },
            new RedisValue[] { _lockValue });
    }
}

public async Task<BookingResult> BookSeatAsync(string seatId, Guid userId)
{
    using var lock = new RedisDistributedLock(_redis, $"seat:{seatId}");
    
    if (!await lock.AcquireLockAsync())
    {
        return BookingResult.Failed("Место временно недоступно для бронирования");
    }
    
    try
    {
        // Проверяем доступность места
        var seat = await _seatRepository.GetByIdAsync(seatId);
        if (seat.IsBooked)
        {
            return BookingResult.Failed("Место уже забронировано");
        }
        
        // Выполняем бронирование
        seat.IsBooked = true;
        seat.BookedBy = userId;
        seat.BookedAt = DateTime.UtcNow;
        
        await _seatRepository.UpdateAsync(seat);
        
        // Публикуем событие о бронировании
        await _eventBus.PublishAsync(new SeatBookedEvent
        {
            SeatId = seatId,
            UserId = userId,
            BookedAt = seat.BookedAt
        });
        
        return BookingResult.Success(seat);
    }
    finally
    {
        await lock.ReleaseLockAsync();
    }
}

public class OrderCreatedEvent
{
  public Guid OrderId { get; set; }
  public Guid CustomerId { get; set; }
  public string CustomerName { get; set; }
  public string CustomerEmail { get; set; }
  public decimal TotalAmount { get; set; }
  public List<OrderItem> Items { get; set; }
  public DateTime CreatedAt { get; set; }
}
 
public class OrderItem
{
  public Guid ProductId { get; set; }
  public string ProductName { get; set; }
  public int Quantity { get; set; }
  public decimal UnitPrice { get; set; }
}

[ProtoContract]
public class OrderCreatedEvent
{
  [ProtoMember(1)]
  public Guid OrderId { get; set; }
  
  [ProtoMember(2)]
  public Guid CustomerId { get; set; }
  
  [ProtoMember(3)]
  public string CustomerName { get; set; }
  
  [ProtoMember(4)]
  public string CustomerEmail { get; set; }
  
  [ProtoMember(5)]
  public decimal TotalAmount { get; set; }
  
  [ProtoMember(6)]
  public List<OrderItem> Items { get; set; }
  
  [ProtoMember(7)]
  public DateTime CreatedAt { get; set; }
}
 
[ProtoContract]
public class OrderItem
{
  [ProtoMember(1)]
  public Guid ProductId { get; set; }
  
  [ProtoMember(2)]
  public string ProductName { get; set; }
  
  [ProtoMember(3)]
  public int Quantity { get; set; }
  
  [ProtoMember(4)]
  public decimal UnitPrice { get; set; }
}

// Сериализация
using var memoryStream = new MemoryStream();
Serializer.Serialize(memoryStream, orderCreatedEvent);
byte[] serializedData = memoryStream.ToArray();
 
// Десериализация
using var inputStream = new MemoryStream(serializedData);
var deserializedEvent = Serializer.Deserialize<OrderCreatedEvent>(inputStream);

[MessagePackObject]
public class OrderCreatedEvent
{
  [Key(0)]
  public Guid OrderId { get; set; }
  
  [Key(1)]
  public Guid CustomerId { get; set; }
  
  [Key(2)]
  public string CustomerName { get; set; }
  
  // и так далее
}
 
// Сериализация
byte[] serializedData = MessagePackSerializer.Serialize(orderCreatedEvent);
 
// Десериализация
var deserializedEvent = MessagePackSerializer.Deserialize<OrderCreatedEvent>(serializedData);

// Сериализация в JSON с помощью System.Text.Json
var jsonOptions = new JsonSerializerOptions 
{ 
    PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
    WriteIndented = false  // важно для производительности!
};
string jsonData = JsonSerializer.Serialize(orderEvent, jsonOptions);
 
// Десериализация
var deserializedOrder = JsonSerializer.Deserialize<OrderCreatedEvent>(jsonData, jsonOptions);

// Пример бенчмарка производительности с BenchmarkDotNet
[Benchmark]
public byte[] SerializeWithProtobuf()
{
    using var stream = new MemoryStream();
    Serializer.Serialize(stream, orderEvent);
    return stream.ToArray();
}
 
[Benchmark]
public string SerializeWithJson()
{
    return JsonSerializer.Serialize(orderEvent);
}

// Пример схемы Avro
var schema = "{\"type\":\"record\",\"name\":\"OrderEvent\",\"fields\":[" +
             "{\"name\":\"OrderId\",\"type\":\"string\"}," +
             "{\"name\":\"CustomerId\",\"type\":\"string\"}," +
             "{\"name\":\"Amount\",\"type\":\"double\"}]}";
 
// Сериализация
using var memStream = new MemoryStream();
var avroWriter = new SpecificWriter<GenericRecord>(Schema.Parse(schema));
var encoder = new BinaryEncoder(memStream);
var record = new GenericRecord(Schema.Parse(schema));
record.Add("OrderId", orderEvent.OrderId.ToString());
record.Add("CustomerId", orderEvent.CustomerId.ToString());
record.Add("Amount", (double)orderEvent.Amount);
avroWriter.Write(record, encoder);

 Partition 0: [msg0] → [msg1] → [msg2] → [msg3] → ...
 Partition 1: [msg0] → [msg1] → [msg2] → ...
 Partition 2: [msg0] → [msg1] → ...

using Confluent.Kafka;
 
var config = new ProducerConfig 
{ 
    BootstrapServers = "localhost:9092",
    // Важные настройки для обеспечения надежности
    EnableIdempotence = true,  // Предотвращает дубликаты
    Acks = Acks.All,           // Ждать подтверждения от всех реплик
    MessageSendMaxRetries = 5   // Количество повторных попыток
};
 
using var producer = new ProducerBuilder<string, string>(config).Build();
 
// Публикация с указанием ключа (влияет на выбор партиции)
await producer.ProduceAsync("my-topic", new Message<string, string> 
{ 
    Key = "user-123",  // Сообщения с одним ключом попадут в одну партицию
    Value = JsonSerializer.Serialize(new { Id = 123, Name = "Иван" }) 
});

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "my-consumer-group",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false  // Ручное управление коммитами для надежности
};
 
using var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
consumer.Subscribe("my-topic");
 
while (true)
{
    try 
    {
        var consumeResult = consumer.Consume(TimeSpan.FromMilliseconds(100));
        if (consumeResult != null)
        {
            // Обработка сообщения
            Console.WriteLine($"Получено: {consumeResult.Message.Value}");
            
            // Коммит смещения ПОСЛЕ успешной обработки
            consumer.Commit(consumeResult);
        }
    }
    catch (ConsumeException e)
    {
        Console.WriteLine($"Ошибка: {e.Error.Reason}");
    }
}

// Потребление с обработкой исключений и повторами
public async Task ConsumeMessages(CancellationToken cancellationToken)
{
    consumer.Subscribe(topicName);
    
    while (!cancellationToken.IsCancellationRequested)
    {
        try
        {
            var consumeResult = consumer.Consume(cancellationToken);
            
            // Обработка сообщения с повторами при временных ошибках
            await ProcessMessageWithRetry(consumeResult.Message, 3);
            
            // Коммит ТОЛЬКО после успешной обработки
            consumer.Commit(consumeResult);
        }
        catch (OperationCanceledException)
        {
            break;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка при обработке сообщения");
            await Task.Delay(1000, cancellationToken);  // Пауза перед повтором
        }
    }
}

// Настройка продюсера с Avro-сериализатором и Schema Registry
var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081" };
 
using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
using var producer = new ProducerBuilder<string, User>(config)
   .SetValueSerializer(new AvroSerializer<User>(schemaRegistry))
   .Build();
 
// Модель с атрибутами Avro
public class User
{
   public string Name { get; set; }
   public int Age { get; set; }
   // Новое поле, добавленное в версии 2 схемы
   [AvroDefault(null)]  // Делает поле опциональным для обратной совместимости
   public string Email { get; set; }
}
 
// Публикация сообщения
await producer.ProduceAsync("users-topic", new Message<string, User> 
{ 
   Key = "user-123", 
   Value = new User { Name = "Иван", Age = 30, Email = "ivan@example.com" } 
});

var consumerConfig = new ConsumerConfig
{
   BootstrapServers = "localhost:9092",
   GroupId = "schema-test-group",
   AutoOffsetReset = AutoOffsetReset.Earliest
};
 
using var consumer = new ConsumerBuilder<string, User>(consumerConfig)
   .SetValueDeserializer(new AvroDeserializer<User>(schemaRegistry))
   .Build();
 
consumer.Subscribe("users-topic");
 
while (true)
{
   var result = consumer.Consume(TimeSpan.FromSeconds(1));
   if (result != null)
   {
       var user = result.Message.Value;
       Console.WriteLine($"Получен пользователь: {user.Name}, возраст: {user.Age}, email: {user.Email ?? "не указан"}");
   }
}

// Подключение к RabbitMQ
var factory = new ConnectionFactory() { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
 
// Объявление обменника и очереди
channel.ExchangeDeclare(exchange: "orders", type: ExchangeType.Topic);
channel.QueueDeclare(queue: "new-orders", durable: true, exclusive: false);
channel.QueueBind(queue: "new-orders", exchange: "orders", routingKey: "order.created.*");
 
// Публикация сообщения
var orderCreatedEvent = new OrderCreatedEvent { OrderId = Guid.NewGuid(), /* ... */ };
var messageBody = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(orderCreatedEvent));
 
channel.BasicPublish(
    exchange: "orders",
    routingKey: "order.created.website",
    basicProperties: null,
    body: messageBody);

// Подключение и объявление очереди (лучше дублировать эту логику и у потребителя)
var factory = new ConnectionFactory() { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
 
channel.QueueDeclare(queue: "new-orders", durable: true, exclusive: false);
 
// Настройка потребления сообщений
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    var orderEvent = JsonSerializer.Deserialize<OrderCreatedEvent>(message);
    
    // Обработка сообщения
    Console.WriteLine($"Получен новый заказ: {orderEvent.OrderId}");
    
    // Подтверждение обработки
    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
 
// Начало потребления с предпочтительным count = 1 (не более 1 сообщения одновременно)
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
channel.BasicConsume(queue: "new-orders", autoAck: false, consumer: consumer);

public class Program
{
    public static async Task Main()
    {
        var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            cfg.Host("localhost", "/", h => {
                h.Username("guest");
                h.Password("guest");
            });
            
            cfg.ReceiveEndpoint("order-service", e =>
            {
                e.Consumer<OrderCreatedConsumer>();
            });
        });
        
        await busControl.StartAsync();
        
        // Отправка сообщения
        var endpoint = await busControl.GetSendEndpoint(new Uri("queue:order-service"));
        await endpoint.Send<IOrderCreated>(new
        {
            OrderId = Guid.NewGuid(),
            CustomerId = "customer-123",
            Amount = 99.95m,
            CreatedAt = DateTime.UtcNow
        });
        
        Console.ReadLine();
        
        await busControl.StopAsync();
    }
}
 
public class OrderCreatedConsumer : IConsumer<IOrderCreated>
{
    public async Task Consume(ConsumeContext<IOrderCreated> context)
    {
        var message = context.Message;
        Console.WriteLine($"Обработка заказа: {message.OrderId}");
        
        // Бизнес-логика здесь
        
        // Публикация следующего события в цепочке
        await context.Publish<IOrderProcessingStarted>(new
        {
            OrderId = message.OrderId,
            StartedAt = DateTime.UtcNow
        });
    }
}

// Получение с явным подтверждением
channel.BasicConsume(queue: "important-queue",
    autoAck: false, // Отключаем автоподтверждение
    consumer: consumer);
 
// В обработчике после успешной обработки
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

// Создание долговечной очереди
channel.QueueDeclare(queue: "critical-data",
    durable: true, // Очередь переживет перезапуск
    exclusive: false,
    autoDelete: false);
 
// Публикация с пометкой сообщения как persistent
var properties = channel.CreateBasicProperties();
properties.Persistent = true; // Сообщение будет сохранено на диск
channel.BasicPublish(exchange: "", routingKey: "critical-data",
    basicProperties: properties,
    body: messageBody);

// Объявление "мертвой очереди" и обменника
channel.ExchangeDeclare("dlx", ExchangeType.Direct);
channel.QueueDeclare("failed-messages", durable: true, exclusive: false, autoDelete: false);
channel.QueueBind("failed-messages", "dlx", "orders");
 
// Объявление основной очереди с настройкой DLX
var arguments = new Dictionary<string, object>
{
    {"x-dead-letter-exchange", "dlx"},
    {"x-dead-letter-routing-key", "orders"}
};
channel.QueueDeclare("orders", durable: true, exclusive: false, autoDelete: false, arguments: arguments);
 
// В потребителе при ошибке обработки
channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: false); // Отправит в DLX

// На стороне продюсера
var messageId = Guid.NewGuid().ToString();
var properties = channel.CreateBasicProperties();
properties.MessageId = messageId;
 
// На стороне потребителя
var messageId = ea.BasicProperties.MessageId;
if (_processedMessageIds.Contains(messageId))
{
    // Сообщение уже обрабатывалось, пропускаем
    channel.BasicAck(ea.DeliveryTag, false);
    return;
}
 
// Обработка сообщения
// ...
 
// Сохраняем ID обработанного сообщения
_processedMessageIds.Add(messageId);

public async Task Consume(ConsumeContext<OrderCreated> context)
{
    var orderId = context.Message.OrderId;
    
    // Проверяем, существует ли уже заказ
    if (await _orderRepository.ExistsAsync(orderId))
    {
        _logger.LogInformation($"Заказ {orderId} уже существует, пропускаем обработку");
        return;
    }
    
    // Создаем заказ
    var order = new Order { Id = orderId, /* другие поля */ };
    await _orderRepository.CreateAsync(order);
}

// Создание приоритетной очереди с 10 уровнями приоритета
var arguments = new Dictionary<string, object>
{
    {"x-max-priority", 10}
};
channel.QueueDeclare("priority-queue", durable: true, exclusive: false, autoDelete: false, arguments: arguments);
 
// Отправка сообщения с высоким приоритетом
var properties = channel.CreateBasicProperties();
properties.Priority = 9; // Высокий приоритет
channel.BasicPublish("", "priority-queue", properties, messageBody);

// Создание парковочной очереди с DLX
var parkingArguments = new Dictionary<string, object>
{
    {"x-dead-letter-exchange", ""},
    {"x-dead-letter-routing-key", "actual-queue"},
    {"x-message-ttl", 60000} // 60 секунд задержки
};
channel.QueueDeclare("parking-queue", durable: true, exclusive: false, autoDelete: false, arguments: parkingArguments);
channel.QueueDeclare("actual-queue", durable: true, exclusive: false, autoDelete: false);
 
// Отправка отложенного сообщения
channel.BasicPublish("", "parking-queue", null, messageBody);

// Ограничиваем количество неподтвержденных сообщений до 10
channel.BasicQos(0, 10, false);

// Проверка длины очереди перед публикацией
var queueInfo = channel.QueueDeclarePassive("my-queue");
var messageCount = queueInfo.MessageCount;
 
if (messageCount > MaxQueueLength)
{
    // Очередь переполнена, замедляем публикацию или останавливаем её
    await Task.Delay(BackoffTime);
}
 
channel.BasicPublish("", "my-queue", null, messageBody);

// Установка соединения
using var connection = new ConnectionFactory().CreateConnection("nats://localhost:4222");
 
// Публикация сообщения
string message = JsonSerializer.Serialize(new { OrderId = "12345", Status = "Created" });
connection.Publish("orders.created", Encoding.UTF8.GetBytes(message));
 
// Подписка на сообщения
var subscription = connection.SubscribeAsync("orders.*", (sender, args) =>
{
    string receivedMessage = Encoding.UTF8.GetString(args.Message.Data);
    Console.WriteLine($"Получено сообщение: {receivedMessage} в {args.Message.Subject}");
});

// Подписка на все события, связанные с заказами конкретного клиента
connection.SubscribeAsync("customer.123.>", HandleCustomerEvents);
 
// Подписка на все события создания заказов независимо от региона
connection.SubscribeAsync("orders.created.*", HandleOrderCreation);
 
// Подписка на все события в системе (осторожно с этим!)
connection.SubscribeAsync(">", HandleAllEvents);

metrics.cpu.server1
metrics.memory.server1
metrics.disk.server1
metrics.cpu.server2
...

// Подключение к NATS Streaming
var cf = new StanConnectionFactory();
using var connection = cf.CreateConnection("test-cluster", "client-123");
 
// Публикация с подтверждением
string guid = connection.Publish("orders.created", Encoding.UTF8.GetBytes(message), (sender, args) =>
{
    Console.WriteLine($"Подтверждение публикации получено: {args.Guid}");
});
 
// Подписка с указанием, с какого момента начать получение сообщений
var opts = StanSubscriptionOptions.GetDefaultOptions();
opts.StartWithLastReceived(); // Начать с последнего полученного
var subscription = connection.Subscribe("orders.*", opts, (sender, args) =>
{
    string receivedMessage = Encoding.UTF8.GetString(args.Message.Data);
    Console.WriteLine($"Получено: {receivedMessage}");
});

// Настройка JetStream
var js = connection.CreateJetStreamContext();
 
// Создание потока (stream)
var streamConfig = StreamConfig.Builder()
    .WithName("ORDERS")
    .WithSubjects("orders.*")
    .WithStorage(StorageType.File)
    .WithRetention(RetentionPolicy.Limits)
    .Build();
 
js.AddStream(streamConfig);
 
// Публикация с подтверждением через JetStream
var ack = await js.PublishAsync("orders.created", Encoding.UTF8.GetBytes(message));
Console.WriteLine($"Сообщение опубликовано, sequence: {ack.Sequence}");
 
// Подписка с доставкой сообщений с самого начала потока
var pushOpts = PushSubscribeOptions.Builder().WithDurable("my-durable").Build();
var sub = await js.PushSubscribeAsync("orders.*", (sender, args) =>
{
    string receivedMessage = Encoding.UTF8.GetString(args.Message.Data);
    Console.WriteLine($"Получено: {receivedMessage}");
    args.Message.Ack(); // Подтверждение обработки
}, pushOpts);

// Создание Key-Value хранилища
var js = connection.CreateJetStreamContext();
var kv = await js.CreateKeyValueAsync(KeyValueConfig.Builder()
    .WithName("CONFIG")
    .WithHistory(5) // Хранить историю изменений
    .Build());
 
// Запись значения
await kv.PutAsync("app.settings.timeout", Encoding.UTF8.GetBytes("30000"));
 
// Чтение значения
var entry = await kv.GetAsync("app.settings.timeout");
var timeout = Encoding.UTF8.GetString(entry.Value);
Console.WriteLine($"Текущий таймаут: {timeout}");
 
// Отслеживание изменений
var watcher = await kv.WatchAllAsync();
await foreach (var update in watcher.Enumerate())
{
    Console.WriteLine($"Ключ {update.Key} был {update.Operation}, значение: {Encoding.UTF8.GetString(update.Value)}");
}

// Создание Object Store
var js = connection.CreateJetStreamContext();
var os = await js.CreateObjectStoreAsync(ObjectStoreConfig.Builder()
    .WithName("FILES")
    .WithMaxBytes(1024 * 1024 * 100) // 100 MB лимит
    .Build());
 
// Загрузка файла
using var fileStream = File.OpenRead("report.pdf");
await os.PutAsync("reports/q2/final.pdf", fileStream);
 
// Получение информации о файле
var info = await os.GetInfoAsync("reports/q2/final.pdf");
Console.WriteLine($"Файл {info.Name}, размер: {info.Size} байт, изменен: {info.ModTime}");
 
// Скачивание файла
using var downloadStream = File.Create("downloaded.pdf");
await os.GetAsync("reports/q2/final.pdf", downloadStream);

var options = ConnectionFactory.GetDefaultOptions();
options.Url = "nats://192.168.1.1:4222,nats://192.168.1.2:4222,nats://192.168.1.3:4222";
options.AllowReconnect = true;
options.MaxReconnect = -1; // Бесконечные попытки реконнекта
 
using var connection = new ConnectionFactory().CreateConnection(options);

var streamConfig = StreamConfig.Builder()
    .WithName("CRITICAL_DATA")
    .WithSubjects("critical.*")
    .WithReplicas(3) // Минимум 3 реплики для надежности
    .Build();
 
js.AddStream(streamConfig);