Управление блокировками распределенной системы с использованием хранилища Azure

Распределенные системы существуют уже достаточно давно, и при их проектировании уже установлены хорошо известные шаблоны. Сегодня мы обсудим один из популярных шаблонов: “блокировки.”

Проще говоря, блокировки — это способ, с помощью которого процессы получают эксклюзивный доступ к ресурсу для выполнения определенного действия. Например, представьте, что в учетной записи хранения есть множество Blob-ов, и вам нужно, чтобы один экземпляр вашего сервиса обрабатывал каждый blob, чтобы избежать дублирующей обработки. Способ сделать это — захватить блокировку на blob, завершить обработку и освободить ее. Однако возникает потенциальная проблема, если процесс завершает работу до освобождения блокировки, либо потому, что процесс завершился, либо из-за сетевого разрыва, оставляя ресурс заблокированным на неопределенный срок. Это может привести к взаимным блокировкам и конкуренции за ресурсы.

Чтобы предотвратить взаимные блокировки, одна из стратегий, которую можно использовать, — это использование таймаутов или блокировок на основе аренды.

Таймаут блокировки

• В этом случае существует предопределенный таймаут, на который процесс запрашивает блокировку. Если блокировка не освобождается до истечения таймаута, система обеспечивает ее окончательное освобождение.

Блокировка аренды

• Для блокировок на основе аренды предоставляется API продления аренды наряду с механизмом таймаута. Процесс, удерживающий блокировку, должен вызвать этот API до истечения срока аренды, чтобы сохранить исключительный доступ к ресурсу. Если процесс не успевает продлить аренду вовремя, блокировка автоматически освобождается, что позволяет другим процессам ее захватить.

Плюсы и минусы блокировок на основе таймаута и аренды

Обе вышеуказанные стратегии являются способом быстрого восстановления после сбоев процессов или сетевых разделений в распределенных системах.

Стратегия блокировки аренды с Azure Storage

Давайте посмотрим, как использовать стратегию блокировки аренды с Azure Storage. Это также охватывает стратегию блокировки по таймауту.

Шаг 1: Импортировать Nuget хранилища блоков

“12.23.0” – последняя версия на момент написания этой статьи. Последние версии можно найти на хранилище блоков Azure.

<ItemGroup>

  <PackageReference Include="Azure.Storage.Blobs" Version="12.23.0" />

</ItemGroup>

Шаг 2: Получение аренды

Ниже приведен код для получения аренды.

    public async Task<string> TryAcquireLeaseAsync(string blobName, TimeSpan durationInSeconds, string leaseId = default)

    {

        BlobContainerClient blobContainerClient = new BlobContainerClient(new Uri($"https://{storageName}.blob.core.windows.net/processors"), tokenCredential, blobClientOptions);

        BlobLeaseClient blobLeaseClient = blobContainerClient.GetBlobClient(blobName).GetBlobLeaseClient(leaseId);

​

        try

        {

            BlobLease lease = await blobLeaseClient.AcquireAsync(durationInSeconds).ConfigureAwait(false);

            return lease.LeaseId;

        }

        catch (RequestFailedException ex) when (ex.Status == 409)

        {

            return default;

        }

    }

1. Сначала мы создаем клиент контейнера блобов и получаем клиент блоба для конкретного блоба, на котором мы хотим получить аренду.
2. Во-вторых, метод “Acquire Async” пытается получить аренду на определенный срок. Если получение прошло успешно, возвращается идентификатор аренды, в противном случае выбрасывается 409 (код состояния для конфликта).
3. Метод “Acquire Async” является ключевым здесь. Остальной код можно настроить/изменить в соответствии с вашими потребностями.

Шаг 3: Обновить аренду

• “Renew Async” – метод в SDK хранилища .NET, используемый для обновления аренды.
• Если обновление не удалось, выбрасывается исключение вместе с причиной сбоя.

public async Task ReleaseLeaseAsync(string blobName, string leaseId)

{

  BlobLeaseClient blobLeaseClient = this.blobContainerClient.GetBlobClient(blobName).GetBlobLeaseClient(leaseId);

​

  await blobLeaseClient.RenewAsync().ConfigureAwait(false);

}

Шаг 4: Организовать методы получения и обновления аренды

• Изначально мы вызываем “Try Acquire Lease Async” для получения идентификатора аренды из Шага 2. Как только это происходит успешно, запускается фоновая задача, которая вызывает “Renew Lease Async” из Шага 3 каждые X секунд. Просто убедитесь, что между временем ожидания и вызовом метода обновления аренды достаточно времени.

string leaseId = await this.blobReadProcessor.TryAcquireLeaseAsync(blobName, TimeSpan.FromSeconds(60)).ConfigureAwait(false);

​

Task leaseRenwerTask = this.taskFactory.StartNew(

        async () =>

        {

            while (leaseId != default && !cancellationToken.IsCancellationRequested)

            {

                await Task.Delay(renewLeaseMillis).ConfigureAwait(false);

                await this.blobReadProcessor.RenewLeaseAsync(blobName, leaseId).ConfigureAwait(false);

            }

        },

        CancellationToken.None,

        TaskCreationOptions.LongRunning,

        TaskScheduler.Default);

• Токен отмены используется для благоприятной остановки задачи обновления аренды, когда она больше не нужна.

Шаг 5: Отмена обновления аренды

• Когда вызывается метод “Cancel Async”, “IsCancellationRequested” в Шаге 4 становится true, из-за чего мы больше не входим в цикл while и не запрашиваем обновление аренды.

await cancellationTokenSource.CancelAsync().ConfigureAwait(false);

await leaseRenwerTask.WaitAsync(Timeout.InfiniteTimeSpan).ConfigureAwait(false);

Шаг 6: Освобождение аренды

Наконец, чтобы освободить аренду, просто вызовите метод “Release Async”.

public async Task ReleaseLeaseAsync(string blobName, string leaseId)

{

  BlobLeaseClient blobLeaseClient = this.blobContainerClient.GetBlobClient(blobName).GetBlobLeaseClient(leaseId);

​

  await blobLeaseClient.ReleaseAsync().ConfigureAwait(false);

}

Заключение

Блокировки являются одними из основных шаблонов в распределенных системах для получения эксклюзивного доступа к ресурсам. Необходимо помнить о возможных проблемах при работе с ними для беспроблемного выполнения операций. Используя Azure Storage, мы можем реализовать эффективные механизмы блокировки, которые могут предотвратить бесконечное блокирование и, в то же время, обеспечить упругость в том, как управляются блокировки.