Gestire i picchi temporanei con il controllo del flusso

A volte, le Data pipelines registrano picchi di traffico pubblicato. Se non ti prepari, i picchi di traffico possono sovraccaricare gli abbonati. Una soluzione semplice per evitare i picchi di traffico è aumentare dinamicamente le risorse degli abbonati Pub/Sub per elaborare più messaggi. Tuttavia, questa soluzione potrebbe aumentare i costi o non funzionare immediatamente. Ad esempio, potresti aver bisogno di molte VM.

Il controllo del flusso sul lato dell'abbonato consente all'abbonato di regolare la velocità di inserimento dei messaggi. Il controllo del flusso gestisce quindi i picchi di traffico senza aumentare i costi o fino a quando l'abbonato non viene scalato.

Il controllo del flusso è una funzionalità disponibile nella libreria client di alto livello Pub/Sub. Puoi anche implementare la tua programmazione di controllo del flusso quando utilizzi una libreria client di basso livello.

Punto chiave: se si verificano picchi improvvisi di traffico per i messaggi pubblicati, utilizza il controllo del flusso nel client dell'abbonato pull per risolvere i picchi temporanei.

La necessità di un controllo del flusso indica che i messaggi vengono pubblicati a una velocità superiore a quella di consumo. Se questo scenario è uno stato persistente, anziché un picco temporaneo nel volume dei messaggi, valuta la possibilità di aumentare il numero di istanze del client dell'abbonato.

Configurazione del controllo del flusso

Il controllo del flusso consente di configurare il numero massimo di byte allocati per le richieste in sospeso e il numero massimo di messaggi in sospeso consentiti. Imposta questi limiti in base alla capacità di throughput delle macchine client.

I valori predefiniti per le variabili di controllo del flusso e i nomi delle variabili potrebbero variare a seconda delle librerie client. Ad esempio, nella libreria client Java, le seguenti variabili configurano il controllo del flusso:

• setMaxOutstandingElementCount(). Definisce il numero massimo di messaggi per i quali Pub/Sub non ha ricevuto riconoscimenti o riconoscimenti negativi.

• setMaxOutstandingRequestBytes(). Definisce la dimensione massima dei messaggi per i quali Pub/Sub non ha ricevuto riconoscimenti o riconoscimenti negativi.

Se viene superato il limite per setMaxOutstandingElementCount() o setMaxOutstandingRequestBytes(), il client dell'abbonato non esegue il pull di altri messaggi. Questo comportamento continua finché i messaggi di cui è già stato eseguito il pull non vengono riconosciuti o riconosciuti negativamente. In questo modo, possiamo allineare il throughput al costo associato all'esecuzione di più abbonati.

Esempi di codice per il controllo del flusso

Per controllare la velocità con cui il client dell'abbonato riceve i messaggi, utilizza le funzionalità di controllo del flusso dell'abbonato. Queste funzionalità di controllo del flusso sono illustrate negli esempi seguenti:

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));
};