Mit Gleichzeitigkeitssteuerung mehr Nachrichten verarbeiten

Die Nebenläufigkeitserkennung ist eine verfügbare Funktion in der Pub/Sub-Clientbibliothek auf hoher Ebene. Wenn Sie eine Low-Level-Bibliothek verwenden, können Sie auch Ihre eigene Parallelitätssteuerung implementieren.

Die Unterstützung für die Parallelitätssteuerung hängt von der Programmiersprache der Clientbibliothek ab. Bei Sprachimplementierungen, die parallele Threads unterstützen, z. B. C++, Go und Java, wird in den Clientbibliotheken standardmäßig eine bestimmte Anzahl von Threads verwendet.

Diese Auswahl ist möglicherweise nicht optimal für Ihre Anwendung. Wenn Ihre Abonnentenanwendung beispielsweise nicht mit dem eingehenden Nachrichtenvolumen mithalten kann und nicht CPU-gebunden ist, müssen Sie die Anzahl der Threads erhöhen. Bei CPU-intensiven Vorgängen zur Nachrichtenverarbeitung kann es sinnvoll sein, die Anzahl der Threads zu reduzieren.

Auf dieser Seite wird das Konzept der Concurrency Control erläutert und beschrieben, wie du die Funktion für deine Abonnentenclients einrichtest. Informationen zum Konfigurieren Ihrer Publisher-Clients für die Nebenläufigkeitserkennung finden Sie unter Nebenläufigkeitserkennung.

Konfigurationen für die Parallelitätssteuerung

Die Standardwerte für die Variablen zur Nebenläufigkeitserkennung und die Namen der Variablen können sich je nach Clientbibliothek unterscheiden. Weitere Informationen finden Sie in der API-Referenzdokumentation. In der Java-Clientbibliothek sind die Methoden zum Konfigurieren der Nebenläufigkeitserkennung beispielsweise setParallelPullCount(), setExecutorProvider(), setSystemExecutorProvider() und setChannelProvider().

  • Mit setParallelPullCount() können Sie festlegen, wie viele Streams geöffnet werden sollen. Sie können weitere Streams öffnen, wenn Ihr Abonnentenclient mehr Daten verarbeiten kann, als in einem einzelnen Stream gesendet werden (10 MBps).

  • Mit setExecutorProvider() können Sie den Executor-Provider anpassen, der für die Verarbeitung von Nachrichten verwendet wird. Sie können beispielsweise den Executor-Anbieter in einen ändern, der einen einzelnen, gemeinsam genutzten Executor mit einer begrenzten Anzahl von Threads für mehrere Abonnentenclients zurückgibt. Diese Konfiguration trägt dazu bei, die Anzahl der erstellten Threads zu begrenzen. Die Gesamtzahl der Threads, die für die Parallelitätssteuerung verwendet werden, hängt vom Executor-Anbieter ab, der in der Clientbibliothek übergeben wird, und von der Anzahl der parallelen Pull-Vorgänge.

  • Mit setSystemExecutorProvider() können Sie den Executor-Anbieter anpassen, der für die Lease-Verwaltung verwendet wird. Normalerweise konfigurieren Sie diesen Wert nur, wenn Sie denselben Executor-Anbieter in setExecutorProvider und setSystemExecutorProvider verwenden möchten. Sie können beispielsweise denselben Executor-Anbieter verwenden, wenn Sie eine Reihe von Abos mit geringem Durchsatz haben. Wenn Sie denselben Wert verwenden, wird die Anzahl der Threads im Client begrenzt.

  • Mit setChannelProvider() können Sie den Channelanbieter anpassen, der zum Öffnen von Verbindungen zu Pub/Sub verwendet wird. Normalerweise konfigurieren Sie diesen Wert nicht, es sei denn, Sie möchten denselben Channel für mehrere Abonnentenclients verwenden. Wenn ein Channel für zu viele Clients wiederverwendet wird, kann dies zu GOAWAY- oder ENHANCE_YOUR_CALM-Fehlern führen. Wenn diese Fehler in den Logs Ihrer Anwendung oder in Cloud Logging angezeigt werden, erstellen Sie weitere Channels.

Codebeispiele für die Steuerung der Parallelität

C++

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für C++ in der Kurzanleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Pub/Sub C++ API-Referenzdokumentation. 

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::GrpcBackgroundThreadPoolSizeOption;
using ::google::cloud::Options;
using ::google::cloud::StatusOr;
auto sample = [](std::string project_id, std::string subscription_id) {
  // Create a subscriber with 16 threads handling I/O work, by default the
  // library creates `std::thread::hardware_concurrency()` threads.
  auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(
      pubsub::Subscription(std::move(project_id), std::move(subscription_id)),
      Options{}
          .set<pubsub::MaxConcurrencyOption>(8)
          .set<GrpcBackgroundThreadPoolSizeOption>(16)));

  // Create a subscription where up to 8 messages are handled concurrently. By
  // default the library uses `std::thread::hardware_concurrency()` as the
  // maximum number of concurrent callbacks.
  auto session = subscriber.Subscribe(
      [](pubsub::Message const& m, pubsub::AckHandler h) {
        // This handler executes in the I/O threads, applications could use,
        // std::async(), a thread-pool, or any other mechanism to transfer the
        // execution to other threads.
        std::cout << "Received message " << m << "\n";
        std::move(h).ack();
        PleaseIgnoreThisSimplifiesTestingTheSamples();
      });
  return std::make_pair(subscriber, std::move(session));
};

Go

Im folgenden Beispiel wird die Hauptversion der Go Pub/Sub-Clientbibliothek (v2) verwendet. Wenn Sie noch die v1-Bibliothek verwenden, finden Sie hier die Migrationsanleitung zu v2. Eine Liste der Codebeispiele für Version 1 finden Sie unter Eingestellte Codebeispiele.

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Go-Einrichtungsanleitung in der Kurzanleitung zur Verwendung von Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Pub/Sub Go API.

import (
	"context"
	"fmt"
	"io"
	"sync/atomic"
	"time"

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

func pullMsgsConcurrencyControl(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 {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
	// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
	// If a subscription ID is provided, the project ID from the client is used.
	sub := client.Subscriber(subID)
	// NumGoroutines determines the number of streams sub.Receive will spawn to pull
	// messages. It is recommended to set this to 1, unless your throughput
	// is greater than 10 MB/s, as even having 1 stream can still result in
	// messages being handled asynchronously.
	sub.ReceiveSettings.NumGoroutines = 1
	// MaxOutstandingMessages limits the number of concurrent handlers of messages.
	// In this case, up to 8 unacked messages can be handled concurrently.
	sub.ReceiveSettings.MaxOutstandingMessages = 8

	// Receive messages for 10 seconds, which simplifies testing.
	// Comment this out in production, since `Receive` should
	// be used as a long running operation.
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	var received int32

	// Receive blocks until the context is cancelled or an error occurs.
	err = sub.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
		atomic.AddInt32(&received, 1)
		msg.Ack()
	})
	if err != nil {
		return fmt.Errorf("sub.Receive returned error: %w", err)
	}
	fmt.Fprintf(w, "Received %d messages\n", received)

	return nil
}

Java

Bevor Sie dieses Beispiel testen, folgen Sie der Einrichtungsanleitung für Java in der Kurzanleitung: Clientbibliotheken verwenden. Weitere Informationen finden Sie in der Referenzdokumentation zur Pub/Sub Java API. 


import com.google.api.gax.core.ExecutorProvider;
import com.google.api.gax.core.InstantiatingExecutorProvider;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class SubscribeWithConcurrencyControlExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String subscriptionId = "your-subscription-id";

    subscribeWithConcurrencyControlExample(projectId, subscriptionId);
  }

  public static void subscribeWithConcurrencyControlExample(
      String projectId, String subscriptionId) {
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          // Handle incoming message, then ack the received message.
          System.out.println("Id: " + message.getMessageId());
          System.out.println("Data: " + message.getData().toStringUtf8());
          consumer.ack();
        };

    Subscriber subscriber = null;
    try {
      // Provides an executor service for processing messages. The default `executorProvider` used
      // by the subscriber has a default thread count of 5.
      ExecutorProvider executorProvider =
          InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(4).build();

      // `setParallelPullCount` determines how many StreamingPull streams the subscriber will open
      // to receive message. It defaults to 1. `setExecutorProvider` configures an executor for the
      // subscriber to process messages. Here, the subscriber is configured to open 2 streams for
      // receiving messages, each stream creates a new executor with 4 threads to help process the
      // message callbacks. In total 2x4=8 threads are used for message processing.
      subscriber =
          Subscriber.newBuilder(subscriptionName, receiver)
              .setParallelPullCount(2)
              .setExecutorProvider(executorProvider)
              .build();

      // Start the subscriber.
      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (TimeoutException timeoutException) {
      // Shut down the subscriber after 30s. Stop receiving messages.
      subscriber.stopAsync();
    }
  }
}

Ruby

Im folgenden Beispiel wird die Ruby Pub/Sub-Clientbibliothek v3 verwendet. Wenn Sie noch die v2-Bibliothek verwenden, finden Sie hier die Migrationsanleitung für v3. Eine Liste der Ruby v2-Codebeispiele finden Sie unter eingestellte Codebeispiele.

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Anleitung zur Einrichtung von Ruby in der Kurzanleitung zur Verwendung von Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Pub/Sub Ruby API. 

# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::PubSub.new
subscriber = pubsub.subscriber subscription_id

# Use 2 threads for streaming, 4 threads for executing callbacks and 2 threads
# for sending acknowledgements and/or delays
listener = subscriber.listen streams: 2, threads: {
  callback: 4,
  push:     2
} do |received_message|
  puts "Received message: #{received_message.data}"
  received_message.acknowledge!
end

listener.start
# Let the main thread sleep for 60 seconds so the thread for listening
# messages does not quit
sleep 60
listener.stop.wait!

Rust

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Einrichtungsanleitung für Rust in der Kurzanleitung zur Verwendung von Clientbibliotheken. Weitere Informationen finden Sie in der Pub/Sub Rust API-Referenzdokumentation. 

use google_cloud_pubsub::client::Subscriber;
use google_cloud_pubsub::subscriber::MessageStream;
use std::time::Duration;

pub async fn sample(project_id: &str, subscription_id: &str) -> anyhow::Result<()> {
    // Assume an available concurrency of 2 for this sample.
    const NCPU: usize = 2;

    let subscription_name = format!("projects/{project_id}/subscriptions/{subscription_id}");
    let client = Subscriber::builder()
        // Configure the subscriber to use multiple gRPC channels. This lets the
        // client multiplex its open streams and its acknowledgement RPCs.
        .with_grpc_subchannel_count(NCPU)
        .build()
        .await?;
    let tasks: Vec<_> = (0..2 * NCPU)
        .map(|i| {
            // Pub/Sub caps the throughput of a single stream to 10 MB/s. To achieve
            // higher throughput, you should open multiple streams.
            let stream = client.subscribe(&subscription_name).build();
            tokio::spawn(subscribe_task(i, stream))
        })
        .collect();

    for t in tasks {
        t.await??;
    }
    println!("done listening for messages");
    Ok(())
}

async fn subscribe_task(index: usize, mut stream: MessageStream) -> anyhow::Result<()> {
    // Terminate the example after 10 seconds.
    let shutdown_token = stream.shutdown_token();
    let shutdown = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(10)).await;
        shutdown_token.shutdown().await;
    });

    println!("listening for messages on stream {index}...");

    while let Some((m, h)) = stream.next().await.transpose()? {
        println!("received message: {m:?}");
        h.ack();
    }
    shutdown.await?;

    Ok(())
}

Nächste Schritte

Weitere Lieferoptionen, die Sie für ein Abo konfigurieren können: