Control de simultaneidad

En este documento, se proporciona información sobre el uso del control de simultaneidad con los mensajes publicados en un tema.

El control de simultaneidad te ayuda a anular la cantidad predeterminada de subprocesos en segundo plano (E/S) que usa la biblioteca cliente para publicar mensajes. Esto permite que los clientes de publicadores envíen mensajes en paralelo.

El control de simultaneidad es una función disponible en la biblioteca cliente de alto nivel de Pub/Sub. También puedes implementar tu propio control de simultaneidad cuando usas una biblioteca de bajo nivel.

La compatibilidad con el control de simultaneidad depende del lenguaje de programación de la biblioteca cliente. En el caso de las implementaciones de lenguajes que admiten subprocesos paralelos, como C++, Go y Java, las bibliotecas cliente realizan una elección predeterminada para la cantidad de subprocesos.

En esta página, se explica el concepto de control de simultaneidad y cómo configurar la función para tus clientes publicadores. Para configurar tus clientes suscriptores para el control de simultaneidad, consulta Procesa más mensajes con el control de simultaneidad.

Antes de comenzar

Antes de configurar el flujo de trabajo de publicación, asegúrate de haber completado las siguientes tareas:

Roles requeridos

Para obtener los permisos que necesitas para publicar mensajes en un tema, pídele a tu administrador que te otorgue el rol de IAM de Publicador de Pub/Sub (roles/pubsub.publisher) en el tema. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

También puedes obtener los permisos necesarios a través de roles personalizados o cualquier otro rol predefinido.

Necesitas permisos adicionales para crear o actualizar temas y suscripciones.

Configuraciones de control de simultaneidad

Los valores predeterminados de las variables de control de simultaneidad y los nombres de las variables pueden diferir entre las bibliotecas cliente. Por ejemplo, en la biblioteca cliente de Java, los métodos para configurar el control de simultaneidad son setExecutorProvider() y setChannelProvider(). Para obtener más información, consulta la documentación de referencia de la API.

  • setExecutorProvider() te permite personalizar el proveedor de ejecutores que se usa para procesar las respuestas de publicación. Por ejemplo, puedes cambiar el proveedor del ejecutor por uno que devuelva un solo ejecutor compartido con una cantidad limitada de subprocesos en varios clientes publicadores. Esta configuración ayuda a limitar la cantidad de subprocesos creados.

  • setChannelProvider() te permite personalizar el proveedor de canales que se usa para abrir conexiones a Pub/Sub. Por lo general, no se configura este valor, a menos que desees usar el mismo canal en varios clientes publicadores. Si se reutiliza un canal en demasiados clientes, es posible que se produzcan errores GOAWAY o ENHANCE_YOUR_CALM. Si ves estos errores en los registros de tu aplicación o en los registros de Cloud, crea más canales.

Muestras de código para el control de simultaneidad

C++

Antes de probar esta muestra, sigue las instrucciones de configuración de C++ en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para C++. 

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::GrpcBackgroundThreadPoolSizeOption;
using ::google::cloud::Options;
using ::google::cloud::StatusOr;
[](std::string project_id, std::string topic_id) {
  auto topic = pubsub::Topic(std::move(project_id), std::move(topic_id));
  // Override the default number of background (I/O) threads. By default the
  // library uses `std::thread::hardware_concurrency()` threads.
  auto options = Options{}.set<GrpcBackgroundThreadPoolSizeOption>(8);
  auto publisher = pubsub::Publisher(
      pubsub::MakePublisherConnection(std::move(topic), std::move(options)));

  std::vector<future<void>> ids;
  for (char const* data : {"1", "2", "3", "go!"}) {
    ids.push_back(
        publisher.Publish(pubsub::MessageBuilder().SetData(data).Build())
            .then([data](future<StatusOr<std::string>> f) {
              auto s = f.get();
              if (!s) return;
              std::cout << "Sent '" << data << "' (" << *s << ")\n";
            }));
  }
  publisher.Flush();
  // Block until they are actually sent.
  for (auto& id : ids) id.get();
}

Go

En el siguiente ejemplo, se usa la versión principal de la biblioteca cliente de Pub/Sub de Go (v2). Si aún usas la biblioteca de la versión 1, consulta la guía de migración a la versión 2. Para ver una lista de muestras de código de la versión 1, consulta las muestras de código obsoletas.

Antes de probar esta muestra, sigue las instrucciones de configuración de Go en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Go.

import (
	"context"
	"fmt"
	"io"

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

func publishSingleGoroutine(w io.Writer, projectID, topicID, msg string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	// msg := "Hello World"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	// client.Publisher can be passed a topic ID (e.g. "my-topic") or
	// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
	// If a topic ID is provided, the project ID from the client is used.
	publisher := client.Publisher(topicID)
	publisher.PublishSettings.NumGoroutines = 1

	result := publisher.Publish(ctx, &pubsub.Message{Data: []byte(msg)})
	// Block until the result is returned and a server-generated
	// ID is returned for the published message.
	id, err := result.Get(ctx)
	if err != nil {
		return fmt.Errorf("Get: %w", err)
	}
	fmt.Fprintf(w, "Published a message; msg ID: %v\n", id)
	return nil
}

Java

Antes de probar esta muestra, sigue las instrucciones de configuración de Java en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Java. 


import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutures;
import com.google.api.gax.core.ExecutorProvider;
import com.google.api.gax.core.InstantiatingExecutorProvider;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

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

    publishWithConcurrencyControlExample(projectId, topicId);
  }

  public static void publishWithConcurrencyControlExample(String projectId, String topicId)
      throws IOException, ExecutionException, InterruptedException {
    TopicName topicName = TopicName.of(projectId, topicId);
    Publisher publisher = null;
    List<ApiFuture<String>> messageIdFutures = new ArrayList<>();

    try {
      // Provides an executor service for processing messages. The default
      // `executorProvider` used by the publisher has a default thread count of
      // 5 * the number of processors available to the Java virtual machine.
      ExecutorProvider executorProvider =
          InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(4).build();

      // `setExecutorProvider` configures an executor for the publisher.
      publisher = Publisher.newBuilder(topicName).setExecutorProvider(executorProvider).build();

      // schedule publishing one message at a time : messages get automatically batched
      for (int i = 0; i < 100; i++) {
        String message = "message " + i;
        ByteString data = ByteString.copyFromUtf8(message);
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

        // Once published, returns a server-assigned message id (unique within the topic)
        ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
        messageIdFutures.add(messageIdFuture);
      }
    } finally {
      // Wait on any pending publish requests.
      List<String> messageIds = ApiFutures.allAsList(messageIdFutures).get();

      System.out.println("Published " + messageIds.size() + " messages with concurrency control.");

      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }
}

Ruby

En el siguiente ejemplo, se usa la versión 3 de la biblioteca cliente de Pub/Sub de Ruby. Si aún usas la biblioteca de la versión 2, consulta la guía de migración a la versión 3. Para ver una lista de muestras de código de Ruby v2, consulta las muestras de código obsoletas.

Antes de probar esta muestra, sigue las instrucciones de configuración de Ruby en la guía de inicio rápido sobre el uso de bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Ruby. 

# topic_id = "your-topic-id"

pubsub = Google::Cloud::PubSub.new
publisher = pubsub.publisher topic_id, async: {
  threads: {
    # Use exactly one thread for publishing message and exactly one thread
    # for executing callbacks
    publish:  1,
    callback: 1
  }
}

publisher.publish_async "This is a test message." do |result|
  raise "Failed to publish the message." unless result.succeeded?
  puts "Message published asynchronously."
end

# Stop the async_publisher to send all queued messages immediately.
publisher.async_publisher.stop.wait!

¿Qué sigue?