Интеграция Hangfire с RabbitMQ в проектах C#.NET

BackgroundJob.Enqueue(() => SomeMethodThatMightFail());

// Пример создания запланированной задачи
RecurringJob.AddOrUpdate(
    "daily-report-generation",
    () => _messageSender.SendMessage(new ReportGenerationMessage { ReportType = "Daily" }),
    Cron.Daily);

// Пример идемпотентного обработчика
public async Task ProcessMessage(OrderMessage message)
{
    // Проверяем, обрабатывали ли мы уже это сообщение
    var messageId = message.Id;
    if (await _processedMessageRepository.Exists(messageId))
    {
        _logger.Info($"Сообщение {messageId} уже обработано, пропускаем");
        return;
    }
    
    try
    {
        // Основная логика обработки
        await ProcessOrder(message.OrderData);
        
        // Запоминаем, что обработали это сообщение
        await _processedMessageRepository.Add(messageId);
    }
    catch (Exception ex)
    {
        _logger.Error($"Ошибка обработки сообщения {messageId}: {ex.Message}");
        throw; // Позволяем RabbitMQ знать, что обработка не удалась
    }
}

// В момент планирования задачи
var correlationId = Guid.NewGuid().ToString();
BackgroundJob.Enqueue(() => _sender.SendWithContext(payload, correlationId));
 
// В обработчике сообщения
public void Process(MessagePayload payload, string correlationId)
{
    using (_logger.BeginScope(new Dictionary<string, object> { ["CorrelationId"] = correlationId }))
    {
        _logger.LogInformation("Начало обработки сообщения");
        // Обработка
        _logger.LogInformation("Сообщение успешно обработано");
    }
}

// Транзакционная отправка сообщений
using (var transaction = _dbContext.Database.BeginTransaction())
{
    try
    {
        // Сохраняем основную сущность
        _dbContext.Orders.Add(order);
        
        // Записываем сообщения в исходящий ящик
        _dbContext.OutboxMessages.Add(new OutboxMessage 
        { 
            Queue = "billing", 
            Payload = JsonConvert.SerializeObject(new BillingMessage { OrderId = order.Id }) 
        });
        _dbContext.OutboxMessages.Add(new OutboxMessage 
        { 
            Queue = "shipping", 
            Payload = JsonConvert.SerializeObject(new ShippingMessage { OrderId = order.Id }) 
        });
        
        _dbContext.SaveChanges();
        transaction.Commit();
        
        // Планируем задачу обработки исходящего ящика
        BackgroundJob.Enqueue(() => _outboxProcessor.ProcessPendingMessages());
    }
    catch (Exception)
    {
        transaction.Rollback();
        throw;
    }
}

// Для быстрых операций
BackgroundJob.Enqueue(() => _messageBroker.SendToQueue("fast-operations", 
    new PaymentConfirmationMessage { PaymentId = paymentId }));
 
// Для длительных операций
BackgroundJob.Enqueue(() => _messageBroker.SendToQueue("long-running", 
    new ReportGenerationMessage { ReportId = reportId }));

public class RabbitMQCircuitBreaker
{
    private readonly IConnectionFactory _connectionFactory;
    private readonly CircuitState _circuitState = new CircuitState();
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
    
    public async Task<bool> PublishMessage(string queue, object message)
    {
        if (_circuitState.IsOpen)
        {
            if (DateTime.UtcNow - _circuitState.LastFailure < _circuitState.ResetTimeout)
            {
                _logger.Warning($"Circuit breaker open, skipping message to {queue}");
                return false;
            }
            
            // Пробуем восстановить соединение
            await _semaphore.WaitAsync();
            try
            {
                _circuitState.State = CircuitBreakerState.HalfOpen;
            }
            finally
            {
                _semaphore.Release();
            }
        }
        
        try
        {
            using (var connection = _connectionFactory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                // Отправка сообщения в RabbitMQ
                // ...
                
                if (_circuitState.State == CircuitBreakerState.HalfOpen)
                {
                    await _semaphore.WaitAsync();
                    try
                    {
                        _circuitState.State = CircuitBreakerState.Closed;
                    }
                    finally
                    {
                        _semaphore.Release();
                    }
                }
                
                return true;
            }
        }
        catch (Exception ex)
        {
            await _semaphore.WaitAsync();
            try
            {
                _circuitState.LastFailure = DateTime.UtcNow;
                _circuitState.State = CircuitBreakerState.Open;
                _circuitState.FailureCount++;
                
                // Увеличиваем время восстановления при повторных сбоях
                if (_circuitState.FailureCount > 3)
                {
                    _circuitState.ResetTimeout = TimeSpan.FromMinutes(5);
                }
            }
            finally
            {
                _semaphore.Release();
            }
            
            _logger.Error($"Failed to publish message to {queue}: {ex.Message}");
            return false;
        }
    }
}

public async Task SendMessage(object payload, string queue)
{
    var stopwatch = Stopwatch.StartNew();
    
    try
    {
        using (var connection = _connectionFactory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // Публикация сообщения
            // ...
            
            stopwatch.Stop();
            _metrics.RecordPublishTime(queue, stopwatch.ElapsedMilliseconds);
            _metrics.IncrementPublishedCount(queue);
        }
    }
    catch (Exception ex)
    {
        stopwatch.Stop();
        _metrics.IncrementFailedPublishCount(queue);
        throw;
    }
}

public async Task SendWithRetries(string queue, object message)
{
    int attempt = 0;
    int maxAttempts = 5;
    TimeSpan delay = TimeSpan.FromSeconds(1);
    
    while (attempt < maxAttempts)
    {
        try
        {
            await SendToRabbitMQ(queue, message);
            return; // Успешно отправлено
        }
        catch (Exception ex) when (IsTransientException(ex))
        {
            attempt++;
            
            if (attempt >= maxAttempts)
            {
                _logger.Error($"Failed to send message after {maxAttempts} attempts: {ex.Message}");
                throw;
            }
            
            _logger.Warning($"Attempt {attempt} failed, retrying after {delay.TotalSeconds}s...");
            await Task.Delay(delay);
            
            // Увеличиваем задержку экспоненциально
            delay = TimeSpan.FromSeconds(Math.Min(60, Math.Pow(2, attempt)));
        }
    }
}

public class RabbitMQService
{
    private readonly IConnectionFactory _connectionFactory;
    private readonly ILogger<RabbitMQService> _logger;
 
    public RabbitMQService(IConfiguration config, ILogger<RabbitMQService> logger)
    {
        _logger = logger;
        _connectionFactory = new ConnectionFactory
        {
            HostName = config["RabbitMQ:Host"] ?? "localhost",
            Port = int.Parse(config["RabbitMQ:Port"] ?? "5672"),
            UserName = config["RabbitMQ:Username"] ?? "guest",
            Password = config["RabbitMQ:Password"] ?? "guest",
            VirtualHost = config["RabbitMQ:VirtualHost"] ?? "/"
        };
    }
 
    public void PublishMessage<T>(string queue, T message) where T : class
    {
        try
        {
            using var connection = _connectionFactory.CreateConnection();
            using var channel = connection.CreateModel();
            
            channel.QueueDeclare(
                queue: queue,
                durable: true,  // Очередь сохранится после перезагрузки брокера
                exclusive: false, // Очередь могут использовать разные соединения
                autoDelete: false, // Очередь не будет удалена автоматически
                arguments: null);
 
            var messageBody = JsonSerializer.SerializeToUtf8Bytes(message);
            
            channel.BasicPublish(
                exchange: "",
                routingKey: queue,
                basicProperties: CreatePersistentProperties(channel),
                body: messageBody);
            
            _logger.LogInformation($"Сообщение отправлено в очередь {queue}");
        }
        catch (Exception ex)
        {
            _logger.LogError($"Ошибка при отправке сообщения: {ex.Message}");
            throw;
        }
    }
 
    private IBasicProperties CreatePersistentProperties(IModel channel)
    {
        var properties = channel.CreateBasicProperties();
        properties.Persistent = true; // Сообщения сохранятся при перезагрузке брокера
        return properties;
    }
}

builder.Services.AddHangfire(config => config
    .SetDataCompatibilityLevel(CompatibilityLevel.Version_170)
    .UseSimpleAssemblyNameTypeSerializer()
    .UseRecommendedSerializerSettings()
    .UseSqlServerStorage(builder.Configuration.GetConnectionString("HangfireConnection")));
 
builder.Services.AddHangfireServer();

app.UseHangfireDashboard();

public class JobScheduler
{
    private readonly IBackgroundJobClient _backgroundJobs;
    private readonly RabbitMQService _rabbitMQService;
 
    public JobScheduler(IBackgroundJobClient backgroundJobs, RabbitMQService rabbitMQService)
    {
        _backgroundJobs = backgroundJobs;
        _rabbitMQService = rabbitMQService;
    }
 
    public string ScheduleMessageDelivery<T>(string queueName, T message, DateTimeOffset executionTime) where T : class
    {
        // Планируем задачу с помощью Hangfire
        return _backgroundJobs.Schedule(
            () => _rabbitMQService.PublishMessage(queueName, message),
            executionTime - DateTimeOffset.Now);
    }
 
    public void RegisterRecurringTask<T>(string jobId, string queueName, T message, string cronExpression) where T : class
    {
        // Регистрируем повторяющуюся задачу
        RecurringJob.AddOrUpdate(
            jobId,
            () => _rabbitMQService.PublishMessage(queueName, message),
            cronExpression);
    }
}

public void DeclareOptimizedQueue(string queueName, bool highThroughput = false)
{
    using var connection = _connectionFactory.CreateConnection();
    using var channel = connection.CreateModel();
    
    var args = new Dictionary<string, object>();
    
    if (highThroughput)
    {
        // Для очередей с высокой нагрузкой
        args["x-queue-mode"] = "lazy"; // Сообщения сбрасываются на диск быстрее
        args["x-max-length"] = 100000; // Ограничиваем длину очереди
        args["x-overflow"] = "reject-publish"; // Отклоняем новые сообщения при достижении лимита
    }
    else
    {
        // Для стандартных очередей
        args["x-dead-letter-exchange"] = ""; // Настройка Dead Letter Exchange
        args["x-dead-letter-routing-key"] = $"{queueName}.dlq"; // Очередь для неуспешных сообщений
    }
    
    channel.QueueDeclare(
        queue: queueName,
        durable: true,
        exclusive: false,
        autoDelete: false,
        arguments: args);
    
    // Создаем очередь для "мертвых" сообщений
    if (!highThroughput)
    {
        channel.QueueDeclare(
            queue: $"{queueName}.dlq",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
    }
}

public async Task PublishWithRetry<T>(string queue, T message, int maxRetries = 3) where T : class
{
    int attempts = 0;
    while (true)
    {
        try
        {
            PublishMessage(queue, message);
            return;
        }
        catch (Exception ex)
        {
            attempts++;
            if (attempts > maxRetries)
            {
                _logger.LogError($"Превышено количество попыток отправки сообщения: {ex.Message}");
                throw;
            }
            
            var delay = TimeSpan.FromSeconds(Math.Pow(2, attempts)); // Экспоненциальная задержка
            _logger.LogWarning($"Попытка {attempts} не удалась. Повторная попытка через {delay.TotalSeconds} сек.");
            await Task.Delay(delay);
        }
    }
}

public class MessageSerializer
{
    private readonly JsonSerializerOptions _options;
    
    public MessageSerializer()
    {
        _options = new JsonSerializerOptions
        {
            WriteIndented = false,
            DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
            PropertyNameCaseInsensitive = true
        };
        
        // Добавляем конвертер для DateTime
        _options.Converters.Add(new JsonStringEnumConverter());
    }
    
    public byte[] Serialize<T>(T message) where T : class
    {
        return JsonSerializer.SerializeToUtf8Bytes(message, _options);
    }
    
    public T Deserialize<T>(byte[] messageBody) where T : class
    {
        return JsonSerializer.Deserialize<T>(messageBody, _options);
    }
}

public static class JobTemplates
{
    public static void RegisterEmailNotificationJob(
        this JobScheduler scheduler, 
        string recipientEmail, 
        string subject, 
        string body, 
        DateTimeOffset sendTime)
    {
        var message = new EmailMessage
        {
            To = recipientEmail,
            Subject = subject,
            Body = body
        };
        
        scheduler.ScheduleMessageDelivery("email-notifications", message, sendTime);
    }
    
    public static void RegisterDailyReportJob(
        this JobScheduler scheduler,
        int reportType,
        string[] recipients)
    {
        var message = new ReportGenerationMessage
        {
            ReportType = reportType,
            Recipients = recipients,
            GenerationTime = DateTime.UtcNow
        };
        
        scheduler.RegisterRecurringTask(
            $"daily-report-{reportType}",
            "report-generation",
            message,
            "0 7 * * *" // Каждый день в 7 утра
        );
    }
}

public class RabbitMQConsumer<T> : BackgroundService where T : class
{
    private readonly IConnectionFactory _connectionFactory;
    private readonly string _queueName;
    private readonly ILogger _logger;
    private readonly IServiceProvider _serviceProvider;
    private IConnection _connection;
    private IModel _channel;
 
    public RabbitMQConsumer(
        IConnectionFactory connectionFactory,
        string queueName,
        ILogger logger,
        IServiceProvider serviceProvider)
    {
        _connectionFactory = connectionFactory;
        _queueName = queueName;
        _logger = logger;
        _serviceProvider = serviceProvider;
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            _connection = _connectionFactory.CreateConnection();
            _channel = _connection.CreateModel();
 
            _channel.QueueDeclare(
                queue: _queueName,
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: null);
 
            // Ограничение на количество одновременно обрабатываемых сообщений
            _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
 
            _logger.LogInformation($"Запущен обработчик для очереди {_queueName}");
 
            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += OnMessageReceived;
 
            _channel.BasicConsume(queue: _queueName, autoAck: false, consumer: consumer);
 
            while (!stoppingToken.IsCancellationRequested)
            {
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError($"Ошибка в потребителе очереди {_queueName}: {ex.Message}");
            throw;
        }
        finally
        {
            Dispose();
        }
    }
 
    private async void OnMessageReceived(object sender, BasicDeliverEventArgs ea)
    {
        var messageId = Guid.NewGuid().ToString();
        _logger.LogInformation($"Получено сообщение {messageId} из очереди {_queueName}");
 
        try
        {
            var messageSerializer = new MessageSerializer();
            var message = messageSerializer.Deserialize<T>(ea.Body.ToArray());
 
            using (var scope = _serviceProvider.CreateScope())
            {
                var processor = scope.ServiceProvider.GetRequiredService<IMessageProcessor<T>>();
                await processor.ProcessAsync(message);
            }
 
            // Подтверждаем успешную обработку
            _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            _logger.LogInformation($"Сообщение {messageId} успешно обработано");
        }
        catch (Exception ex)
        {
            _logger.LogError($"Ошибка при обработке сообщения {messageId}: {ex.Message}");
            
            // Возвращаем сообщение в очередь или отправляем в DLQ
            _channel.BasicNack(
                deliveryTag: ea.DeliveryTag,
                multiple: false, 
                requeue: ShouldRequeue(ex)); // Возвращаем в очередь только при временных ошибках
        }
    }
 
    private bool ShouldRequeue(Exception ex)
    {
        // Решаем, возвращать ли сообщение в очередь, в зависимости от типа ошибки
        return ex is TimeoutException || ex is DbException || ex is HttpRequestException;
    }
 
    public override void Dispose()
    {
        _channel?.Close();
        _connection?.Close();
        base.Dispose();
    }
}

public interface IMessageProcessor<T> where T : class
{
    Task ProcessAsync(T message);
}

public class EmailNotificationProcessor : IMessageProcessor<EmailMessage>
{
    private readonly IEmailService _emailService;
    private readonly ILogger<EmailNotificationProcessor> _logger;
 
    public EmailNotificationProcessor(
        IEmailService emailService,
        ILogger<EmailNotificationProcessor> logger)
    {
        _emailService = emailService;
        _logger = logger;
    }
 
    public async Task ProcessAsync(EmailMessage message)
    {
        _logger.LogInformation($"Отправка email для {message.To}: {message.Subject}");
        await _emailService.SendEmailAsync(message.To, message.Subject, message.Body);
    }
}

// Регистрация фабрики соединений RabbitMQ
services.AddSingleton<IConnectionFactory>(sp => 
{
    var configuration = sp.GetRequiredService<IConfiguration>();
    return new ConnectionFactory
    {
        HostName = configuration["RabbitMQ:Host"],
        UserName = configuration["RabbitMQ:Username"],
        Password = configuration["RabbitMQ:Password"],
        VirtualHost = configuration["RabbitMQ:VirtualHost"]
    };
});
 
// Регистрация процессоров сообщений
services.AddScoped<IMessageProcessor<EmailMessage>, EmailNotificationProcessor>();
services.AddScoped<IMessageProcessor<ReportGenerationMessage>, ReportGenerationProcessor>();
 
// Регистрация потребителей как фоновые службы
services.AddHostedService(sp => new RabbitMQConsumer<EmailMessage>(
    sp.GetRequiredService<IConnectionFactory>(),
    "email-notifications",
    sp.GetRequiredService<ILogger<RabbitMQConsumer<EmailMessage>>>(),
    sp
));
 
services.AddHostedService(sp => new RabbitMQConsumer<ReportGenerationMessage>(
    sp.GetRequiredService<IConnectionFactory>(),
    "report-generation",
    sp.GetRequiredService<ILogger<RabbitMQConsumer<ReportGenerationMessage>>>(),
    sp
));

public T DeserializeWithFallback<T>(byte[] messageBody) where T : class
{
    try
    {
        return JsonSerializer.Deserialize<T>(messageBody, _options);
    }
    catch (JsonException ex)
    {
        _logger.LogWarning($"Не удалось десериализовать сообщение: {ex.Message}");
        
        // Пробуем десериализовать как динамический объект
        var dynamicObject = JsonSerializer.Deserialize<JsonElement>(messageBody);
        
        // Создаем новый экземпляр T и заполняем его доступными свойствами
        var instance = Activator.CreateInstance<T>();
        var properties = typeof(T).GetProperties();
        
        foreach (var prop in properties)
        {
            if (dynamicObject.TryGetProperty(prop.Name, out var value))
            {
                try
                {
                    var convertedValue = ConvertJsonValue(value, prop.PropertyType);
                    prop.SetValue(instance, convertedValue);
                }
                catch
                {
                    // Игнорируем ошибки конвертации отдельных свойств
                }
            }
        }
        
        return instance;
    }
}
 
private object ConvertJsonValue(JsonElement element, Type targetType)
{
    // Реализация конвертации различных типов JsonElement в целевой тип
    // ...
}

static async Task Main(string[] args)
{
    // Настройка DI
    var services = ConfigureServices();
    var serviceProvider = services.BuildServiceProvider();
    
    // Получаем планировщик задач
    var scheduler = serviceProvider.GetRequiredService<JobScheduler>();
    
    // Планируем отправку email через 30 секунд
    scheduler.ScheduleMessageDelivery(
        "email-notifications",
        new EmailMessage
        {
            To = "recipient@example.com",
            Subject = "Тестовое сообщение",
            Body = "Это сообщение было отправлено через Hangfire и RabbitMQ"
        },
        DateTimeOffset.Now.AddSeconds(30));
    
    Console.WriteLine("Задача запланирована. Нажмите любую клавишу для выхода...");
    Console.ReadKey();
}
 
static IServiceCollection ConfigureServices()
{
    var services = new ServiceCollection();
    
    // Конфигурация
    var configuration = new ConfigurationBuilder()
        .AddJsonFile("appsettings.json")
        .Build();
    services.AddSingleton<IConfiguration>(configuration);
    
    // Логирование
    services.AddLogging(builder =>
    {
        builder.AddConsole();
        builder.SetMinimumLevel(LogLevel.Information);
    });
    
    // Hangfire
    services.AddHangfire(config => config
        .UseMemoryStorage());
    services.AddHangfireServer();
    
    // RabbitMQ
    services.AddSingleton<RabbitMQService>();
    services.AddSingleton<JobScheduler>();
    
    // Регистрация всех необходимых зависимостей...
    
    return services;
}

// Неправильно
channel.BasicConsume(queue: "my-queue", autoAck: true, consumer: consumer);
 
// Правильно
channel.BasicConsume(queue: "my-queue", autoAck: false, consumer: consumer);
 
// После успешной обработки
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

// Создание постоянной очереди
channel.QueueDeclare(
    queue: "important-queue",
    durable: true,  // Очередь переживёт перезапуск брокера
    exclusive: false,
    autoDelete: false,
    arguments: null);
 
// Отправка персистентного сообщения
var properties = channel.CreateBasicProperties();
properties.Persistent = true;  // Сообщение будет сохранено на диск
channel.BasicPublish(..., basicProperties: properties, ...);

// Добавляем аргументы DLX при создании очереди
var arguments = new Dictionary<string, object>
{
    {"x-dead-letter-exchange", "dlx.exchange"},
    {"x-dead-letter-routing-key", "dlx.queue"}
};
 
channel.QueueDeclare("main-queue", true, false, false, arguments);
 
// Отклоняем сообщение с requeue: false, оно попадёт в DLX
channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: false);

// Пример добавления метрик в код обработчика
public async Task ProcessMessage(T message, CancellationToken cancellationToken)
{
    var stopwatch = Stopwatch.StartNew();
    
    try
    {
        await _actualProcessor.ProcessAsync(message, cancellationToken);
        stopwatch.Stop();
        
        // Запись метрики времени обработки
        _metrics.RecordMessageProcessingTime(_queueName, stopwatch.ElapsedMilliseconds);
        _metrics.IncrementSuccessCount(_queueName);
    }
    catch (Exception ex)
    {
        stopwatch.Stop();
        _metrics.IncrementErrorCount(_queueName);
        _metrics.RecordExceptionType(_queueName, ex.GetType().Name);
        throw;
    }
}

// Для ресурсоёмких операций лучше брать по одному сообщению
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
 
// Для быстрых операций можно увеличить
channel.BasicQos(prefetchSize: 0, prefetchCount: 10, global: false);

public async Task ProcessBatch(IEnumerable<T> messages, CancellationToken token)
{
    // Например, для вставки в БД одним запросом
    await _repository.BulkInsertAsync(messages.Select(m => MapToEntity(m)), token);
}

// При публикации выбираем шард по хэшу ключа
int shardCount = 4;
string routingKey = $"my-queue-{message.TenantId.GetHashCode() % shardCount}";
channel.BasicPublish("", routingKey, null, messageBody);

// Хранение счётчика повторов в заголовках сообщения
public async Task ProcessWithRetryLimit(BasicDeliverEventArgs ea)
{
    // Проверяем, есть ли заголовок с числом повторов
    int retryCount = 0;
    if (ea.BasicProperties.Headers != null && 
        ea.BasicProperties.Headers.TryGetValue("x-retry-count", out var value))
    {
        retryCount = (int)value;
    }
    
    if (retryCount >= MaxRetries)
    {
        // Превышен лимит - отправляем в DLQ
        _channel.BasicNack(ea.DeliveryTag, false, false);
        await LogFailedMessage(ea);
        return;
    }
    
    try
    {
        // Пытаемся обработать
        await ProcessMessage(ea);
        _channel.BasicAck(ea.DeliveryTag, false);
    }
    catch (Exception ex)
    {
        // Инкрементируем счётчик и возвращаем в очередь
        var properties = _channel.CreateBasicProperties();
        properties.Headers = new Dictionary<string, object>
        {
            ["x-retry-count"] = retryCount + 1
        };
        
        // Публикуем копию с обновленным счётчиком
        _channel.BasicPublish("", _queueName, properties, ea.Body);
        
        // Подтверждаем оригинал
        _channel.BasicAck(ea.DeliveryTag, false);
    }
}

private bool ShouldRetry(Exception ex)
{
    // Повторяем при временных проблемах
    return ex is TimeoutException ||
           ex is DbException ||
           ex is HttpRequestException ||
           (ex is SocketException socketEx && IsTransientSocketError(socketEx.ErrorCode));
           
    // При проблемах с данными повторы бесполезны
    return !(ex is FormatException ||
             ex is ArgumentException ||
             ex is ValidationException);
}

// Вычисление задержки на основе номера попытки
private TimeSpan CalculateDelay(int attempt)
{
    // Базовая задержка 1 секунда, максимум 1 час
    var seconds = Math.Min(3600, Math.Pow(2, attempt));
    return TimeSpan.FromSeconds(seconds);
}

// Пример настройки задачи в Quartz.NET
public class EmailReminderJob : IJob
{
    private readonly IEmailService _emailService;
    
    public EmailReminderJob(IEmailService emailService)
    {
        _emailService = emailService;
    }
    
    public async Task Execute(IJobExecutionContext context)
    {
        var data = context.JobDetail.JobDataMap;
        var email = data.GetString("Email");
        var subject = data.GetString("Subject");
        
        await _emailService.SendAsync(email, subject);
    }
}
 
// Регистрация задачи
var job = JobBuilder.Create<EmailReminderJob>()
    .WithIdentity("reminderJob", "emails")
    .UsingJobData("Email", "customer@example.com")
    .UsingJobData("Subject", "Не забудьте о встрече!")
    .Build();
    
var trigger = TriggerBuilder.Create()
    .WithIdentity("dailyTrigger", "emails")
    .WithCronSchedule("0 0 9 ? * MON-FRI")  // В 9:00 по будням
    .Build();
    
await scheduler.ScheduleJob(job, trigger);

// Пример Azure Function с триггером по таймеру
public static class ScheduledEmailFunction
{
    [FunctionName("DailyEmailReminder")]
    public static async Task Run(
        [TimerTrigger("0 0 9 * * *")] TimerInfo timer,
        [ServiceBusTrigger("email-queue")] EmailMessage message,
        ILogger log)
    {
        log.LogInformation($"Отправка напоминания для: {message.Email}");
        await SendEmailAsync(message);
    }
}