Processar picos temporários com controle de fluxo

Data pipelines às vezes recebem picos no tráfego publicado. Esses picos podem sobrecarregar os assinantes, a menos que você esteja preparado para isso. Uma solução simples para evitar picos de tráfego é aumentar dinamicamente os recursos do assinante do Pub/Sub para processar mais mensagens. No entanto, essa solução pode aumentar os custos ou não funcionar instantaneamente. Por exemplo, você pode precisar de muitas VMs.

O controle de fluxo no lado do assinante permite que ele regule a taxa de ingestão de mensagens. Assim, o controle de fluxo processa picos de tráfego sem aumentar os custos ou até que o assinante seja escalonado.

O controle de fluxo é um recurso disponível na biblioteca de cliente de alto nível do Pub/Sub. Também é possível implementar sua própria programação de controle de fluxo ao usar uma biblioteca de cliente de baixo nível.

Ponto principal: se você receber picos repentinos de tráfego para mensagens publicadas, use o controle de fluxo no cliente do assinante de extração para resolver os picos temporários.

A necessidade do controle de fluxo indica que as mensagens são publicadas com uma taxa maior do que são consumidas. Se esse cenário for um estado constante, em vez de um pico transitório no volume de mensagens, considere aumentar o número de instâncias do cliente assinante.

Configuração do controle de fluxo

O controle de fluxo permite configurar o número máximo de bytes alocados para solicitações pendentes e o número máximo de mensagens pendentes permitidas. Defina esses limites de acordo com a capacidade de processamento dos computadores clientes.

Os valores padrão das variáveis de controle de fluxo e os nomes das variáveis podem variar entre as bibliotecas de cliente. Por exemplo, na biblioteca de cliente Java, as seguintes variáveis configuram o controle de fluxo:

  • setMaxOutstandingElementCount(). Define o número máximo de mensagens para as quais o Pub/Sub não recebeu confirmações ou confirmações negativas.

  • setMaxOutstandingRequestBytes(). Define o tamanho máximo de mensagens para as quais o Pub/Sub não recebeu confirmações ou confirmações negativas.

Se o limite de setMaxOutstandingElementCount() ou setMaxOutstandingRequestBytes() for ultrapassado, o cliente do assinante não extrairá mais mensagens. Esse comportamento continua até que as mensagens já extraídas sejam confirmadas ou confirmadas negativamente. Assim, podemos alinhar a capacidade de processamento com o custo associado à execução de mais assinantes.

Exemplos de código para controle de fluxo

Para controlar a taxa com que o cliente do assinante recebe mensagens, use os recursos de controle de fluxo do assinante. Esses recursos de controle de fluxo são ilustrados nas amostras a seguir:

C++

Antes de tentar esse exemplo, siga as instruções de configuração do C++ em Guia de início rápido: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Pub/Sub C++. 

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::Options;
using ::google::cloud::StatusOr;
auto sample = [](std::string project_id, std::string subscription_id) {
  // Change the flow control watermarks, by default the client library uses
  // 0 and 1,000 for the message count watermarks, and 0 and 10MiB for the
  // size watermarks. Recall that the library stops requesting messages if
  // any of the high watermarks are reached, and the library resumes
  // requesting messages when *both* low watermarks are reached.
  auto constexpr kMiB = 1024 * 1024L;
  auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(
      pubsub::Subscription(std::move(project_id), std::move(subscription_id)),
      Options{}
          .set<pubsub::MaxOutstandingMessagesOption>(1000)
          .set<pubsub::MaxOutstandingBytesOption>(8 * kMiB)));

  auto session = subscriber.Subscribe(
      [](pubsub::Message const& m, pubsub::AckHandler h) {
        std::move(h).ack();
        std::cout << "Received message " << m << "\n";
        PleaseIgnoreThisSimplifiesTestingTheSamples();
      });
  return std::make_pair(subscriber, std::move(session));
};

C#

Antes de tentar esse exemplo, siga as instruções de configuração do C# em Guia de início rápido: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Pub/Sub C# . 


using Google.Api.Gax;
using Google.Cloud.PubSub.V1;
using System;
using System.Threading;
using System.Threading.Tasks;

public class PullMessagesWithFlowControlAsyncSample
{
    public async Task<int> PullMessagesWithFlowControlAsync(string projectId, string subscriptionId, bool acknowledge)
    {
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);
        int messageCount = 0;
        SubscriberClient subscriber = await new SubscriberClientBuilder
        {
            SubscriptionName = subscriptionName,
            Settings = new SubscriberClient.Settings
            {
                AckExtensionWindow = TimeSpan.FromSeconds(4),
                AckDeadline = TimeSpan.FromSeconds(10),
                FlowControlSettings = new FlowControlSettings(maxOutstandingElementCount: 100, maxOutstandingByteCount: 10240)
            }
        }.BuildAsync();
        // SubscriberClient runs your message handle function on multiple
        // threads to maximize throughput.
        Task startTask = subscriber.StartAsync((PubsubMessage message, CancellationToken cancel) =>
        {
            string text = message.Data.ToStringUtf8();
            Console.WriteLine($"Message {message.MessageId}: {text}");
            Interlocked.Increment(ref messageCount);
            return Task.FromResult(acknowledge ? SubscriberClient.Reply.Ack : SubscriberClient.Reply.Nack);
        });
        // Run for 5 seconds.
        await Task.Delay(5000);
        await subscriber.StopAsync(CancellationToken.None);
        // Lets make sure that the start task finished successfully after the call to stop.
        await startTask;
        return messageCount;
    }
}

Go

O exemplo a seguir usa a versão principal da biblioteca de cliente do Go Pub/Sub (v2). Se você ainda estiver usando a biblioteca v1, consulte o guia de migração para a v2. Para conferir uma lista de exemplos de código da v1, consulte os exemplos de código obsoletos.

Antes de tentar esse exemplo, siga as instruções de configuração do Go em Guia de início rápido: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Pub/Sub Go.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
)

func pullMsgsFlowControlSettings(w io.Writer, projectID, subID string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {