Partilhar via

Feed de mudanças de processo no Armazenamento de Blobs do Azure

O feed de alterações fornece registos de transações de todas as alterações que ocorrem nos blobs e nos metadados de blob na sua conta de armazenamento. Este artigo mostra como ler registos do fluxo de alterações utilizando a biblioteca do processador de alterações de blobs.

Para saber mais sobre o feed de alterações, consulte Feed de Alterações no Armazenamento de Blobs do Azure.

Configure o seu projeto

Esta secção guia-o na preparação de um projeto para trabalhar com a biblioteca cliente do Blob Change Feed para .NET.

Instalar pacotes

No diretório do projeto, instale o pacote usando o dotnet add package comando para a biblioteca cliente do Feed de Alteração de Blobs de Armazenamento do Azure para .NET. Neste exemplo, adicionamos o sinalizador --prerelease ao comando para instalar a versão de visualização mais recente.

dotnet add package Azure.Storage.Blobs.ChangeFeed --prerelease

Os exemplos de código neste artigo também usam os pacotes Azure Blob Storage e Azure Identity .

dotnet add package Azure.Identity
dotnet add package Azure.Storage.Blobs

Adicionar using diretivas

Adicione as seguintes using diretivas ao seu arquivo de código:

using Azure.Identity;
using Azure.Storage.Blobs;
using Azure.Storage.Blobs.ChangeFeed;

Criar um objeto cliente

Para conectar o aplicativo ao Armazenamento de Blob, crie uma instância da BlobServiceClient classe. O exemplo a seguir mostra como criar um objeto cliente usando DefaultAzureCredential para autorização. Para saber mais, consulte Autorizar acesso e conectar-se ao Blob Storage. Para trabalhar com o feed de alterações, você precisa do Leitor de Dados de Blob de Armazenamento do Azure RBAC ou superior.

// TODO: Replace <storage-account-name> with the name of your storage account
string accountName = "<storage-account-name>";

BlobServiceClient client = new(
        new Uri($"https://{accountName}.blob.core.windows.net"),
        new DefaultAzureCredential());

O objeto cliente é passado como um parâmetro para alguns dos métodos mostrados neste artigo.

Ler registos no feed de alterações

Nota

O feed de alterações é uma entidade imutável e somente leitura na sua conta de armazenamento. Qualquer número de aplicativos pode ler e processar o feed de alterações simultaneamente e de forma independente, de acordo com sua própria conveniência. Os registros não são removidos do feed de alterações quando um aplicativo os lê. O estado de leitura ou iteração de cada leitor consumidor é independente e mantido apenas pelo seu aplicativo.

O exemplo de código a seguir itera por todos os registros no feed de alterações, adiciona-os a uma lista e retorna a lista de eventos do feed de alterações:

public async Task<List<BlobChangeFeedEvent>> ChangeFeedAsync(BlobServiceClient client)
{
    // Create a new BlobChangeFeedClient
    BlobChangeFeedClient changeFeedClient = client.GetChangeFeedClient();

    List<BlobChangeFeedEvent> changeFeedEvents = [];

    // Get all the events in the change feed
    await foreach (BlobChangeFeedEvent changeFeedEvent in changeFeedClient.GetChangesAsync())
    {
        changeFeedEvents.Add(changeFeedEvent);
    }

    return changeFeedEvents;
}

O exemplo de código a seguir imprime alguns valores da lista de eventos de feed de alteração:

public void showEventData(List<BlobChangeFeedEvent> changeFeedEvents)
{
    foreach (BlobChangeFeedEvent changeFeedEvent in changeFeedEvents)
    {
        string subject = changeFeedEvent.Subject;
        string eventType = changeFeedEvent.EventType.ToString();
        BlobOperationName operationName = changeFeedEvent.EventData.BlobOperationName;

        Console.WriteLine("Subject: " + subject + "\n" +
        "Event Type: " + eventType + "\n" +
        "Operation: " + operationName.ToString());
    }
}

Retomar a leitura de registos a partir de uma posição guardada

Você pode optar por salvar sua posição de leitura no feed de alterações e, em seguida, retomar a iteração pelos registros em um momento futuro. Você pode salvar a posição de leitura obtendo o cursor do feed de alterações. O cursor é uma cadeia de caracteres e seu aplicativo pode salvá-la de qualquer maneira que faça sentido para o design do seu aplicativo, por exemplo, em um arquivo ou banco de dados.

Este exemplo itera por todos os registos no fluxo de alterações, adiciona-os a uma lista e salva o cursor. A lista e o cursor são retornados ao chamador.

public async Task<(string, List<BlobChangeFeedEvent>)> ChangeFeedResumeWithCursorAsync(
    BlobServiceClient client,
    string cursor)
{
    // Get a new change feed client
    BlobChangeFeedClient changeFeedClient = client.GetChangeFeedClient();
    List<BlobChangeFeedEvent> changeFeedEvents = new List<BlobChangeFeedEvent>();

    IAsyncEnumerator<Page<BlobChangeFeedEvent>> enumerator = changeFeedClient
        .GetChangesAsync(continuationToken: cursor)
        .AsPages(pageSizeHint: 10)
        .GetAsyncEnumerator();

    await enumerator.MoveNextAsync();

    foreach (BlobChangeFeedEvent changeFeedEvent in enumerator.Current.Values)
    {

        changeFeedEvents.Add(changeFeedEvent);
    }

    // Update the change feed cursor. The cursor is not required to get each page of events,
    // it's intended to be saved and used to resume iterating at a later date.
    cursor = enumerator.Current.ContinuationToken;
    return (cursor, changeFeedEvents);
}

Processamento em fluxo de registos

Você pode optar por processar os registros de feed de alteração à medida que eles estão comprometidos com o feed de alterações. Consulte as especificações. Os eventos de alteração são publicados no feed de alterações em um período de 60 segundos, em média. Recomendamos que você pesquise novas alterações com esse período em mente ao especificar seu intervalo de pesquisa.

Este exemplo sonda periodicamente as alterações. Se existirem registos de alteração, este código processa esses registos e guarda o cursor do feed de alterações. Dessa forma, se o processo for interrompido e, em seguida, iniciado novamente, o aplicativo pode usar o cursor para retomar o processamento de registros onde parou pela última vez. Este exemplo salva o cursor em um arquivo local para fins de demonstração, mas seu aplicativo pode salvá-lo de qualquer forma que faça mais sentido para seu cenário.

public async Task ChangeFeedStreamAsync(
    BlobServiceClient client,
    int waitTimeMs,
    string cursor)
{
    // Get a new change feed client
    BlobChangeFeedClient changeFeedClient = client.GetChangeFeedClient();

    while (true)
    {
        IAsyncEnumerator<Page<BlobChangeFeedEvent>> enumerator = changeFeedClient
        .GetChangesAsync(continuationToken: cursor).AsPages().GetAsyncEnumerator();

        while (true)
        {
            var result = await enumerator.MoveNextAsync();

            if (result)
            {
                foreach (BlobChangeFeedEvent changeFeedEvent in enumerator.Current.Values)
                {
                    string subject = changeFeedEvent.Subject;
                    string eventType = changeFeedEvent.EventType.ToString();
                    BlobOperationName operationName = changeFeedEvent.EventData.BlobOperationName;

                    Console.WriteLine("Subject: " + subject + "\n" +
                        "Event Type: " + eventType + "\n" +
                        "Operation: " + operationName.ToString());
                }

                // Helper method to save cursor
                SaveCursor(enumerator.Current.ContinuationToken);
            }
            else
            {
                break;
            }

        }
        await Task.Delay(waitTimeMs);
    }
}

void SaveCursor(string cursor)
{
    // Specify the path to the file where you want to save the cursor
    string filePath = "path/to/cursor.txt";

    // Write the cursor value to the file
    File.WriteAllText(filePath, cursor);
}

Ler registros dentro de um intervalo de tempo específico

Você pode ler registros que se enquadram em um intervalo de tempo específico. Este exemplo itera através de todos os registos no feed de alterações que estão dentro de um período de tempo específico, adiciona-os a uma lista e retorna a lista.

async Task<List<BlobChangeFeedEvent>> ChangeFeedBetweenDatesAsync(BlobServiceClient client)
{
    // Get a new change feed client
    BlobChangeFeedClient changeFeedClient = client.GetChangeFeedClient();
    List<BlobChangeFeedEvent> changeFeedEvents = new List<BlobChangeFeedEvent>();

    // Create the start and end time.  The change feed client will round start time down to
    // the nearest hour, and round endTime up to the next hour if you provide DateTimeOffsets
    // with minutes and seconds.
    DateTimeOffset startTime = new DateTimeOffset(2024, 3, 1, 0, 0, 0, TimeSpan.Zero);
    DateTimeOffset endTime = new DateTimeOffset(2024, 6, 1, 0, 0, 0, TimeSpan.Zero);

    // You can also provide just a start or end time.
    await foreach (BlobChangeFeedEvent changeFeedEvent in changeFeedClient.GetChangesAsync(
        start: startTime,
        end: endTime))
    {
        changeFeedEvents.Add(changeFeedEvent);
    }

    return changeFeedEvents;
}

A hora de início fornecida é arredondada para a hora mais próxima e a hora de fim é arredondada para a hora mais próxima. É possível que os utilizadores vejam eventos que ocorreram antes da hora de início e depois da hora de fim. Também é possível que alguns eventos que ocorrem entre a hora de início e a hora de término não apareçam. Isso porque os eventos podem ser registrados durante a hora anterior à hora de início ou durante a hora após a hora de término.

Próximos passos

  • Last updated on