Se usó la API de Cloud Translation para traducir esta página.
Switch to English

Usa Pub/Sub en aplicaciones de Spring

En esta página, se describe cómo usar Pub/Sub en aplicaciones Java compiladas con el framework Spring.

Spring Cloud GCP tiene varios módulos para enviar mensajes a temas de Pub/Sub y recibir mensajes de suscripciones de Pub/Sub mediante el framework Spring. Puedes usar estos módulos por separado o combinarlos en diferentes casos de uso:

Antes de comenzar

  1. Accede a tu Cuenta de Google.

    Si todavía no tienes una cuenta, regístrate para obtener una nueva.

  2. Configura un proyecto de Cloud Console.

    Configurar un proyecto

    Haz clic para realizar alguna de las siguientes acciones:

    • Crea o selecciona un proyecto.
    • Habilita la API de Pub/Sub para ese proyecto.
    • Crea una cuenta de servicio.
    • Descargar una clave privada como JSON

    Puedes ver y administrar estos recursos en cualquier momento desde Cloud Console.

  3. Configura la variable de entorno GOOGLE_APPLICATION_CREDENTIALS en la ruta del archivo JSON que contiene la clave de tu cuenta de servicio. Esta variable solo se aplica a la sesión actual de shell. Por lo tanto, si abres una sesión nueva, deberás volver a configurar la variable.

  4. Configura la variable de entorno GOOGLE_CLOUD_PROJECT como el ID de tu proyecto de Cloud.

Usa Spring Cloud GCP Pub/Sub Starter

El módulo Spring Cloud GCP Pub/Sub Starter instala la biblioteca cliente de Java para Pub/Sub con el módulo Spring Cloud GCP Pub/Sub. Puedes llamar a la API de Pub/Sub desde tu aplicación de Spring con las clases que proporciona Spring Cloud Pub/Sub Starter o la biblioteca cliente de Java para Pub/Sub. Si usas las clases que proporciona Spring Cloud GCP Pub/Sub Starter, puedes anular las configuraciones de Pub/Sub predeterminadas.

Instala el módulo

Para instalar el módulo de Spring Cloud GCP Pub/Sub Starter, agrega estas dependencias a tu archivo pom.xml:

  1. La lista de materiales (BOM) de Spring Cloud GCP:

    <dependencyManagement>
        <dependency>
          <groupId>org.springframework.cloud</groupId>
          <artifactId>spring-cloud-gcp-dependencies</artifactId>
          <version>1.2.6.RELEASE</version>
          <type>pom</type>
          <scope>import</scope>
        </dependency>
      </dependencies>
    </dependencyManagement>
  2. El artefacto de Spring Cloud GCP Pub/Sub Starter:

    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
    </dependency>

Operaciones admitidas

El módulo de Spring Cloud GCP Pub/Sub Starter incluye las siguientes clases:

  • PubSubAdmin para operaciones administrativas:
    • Crea temas y suscripciones.
    • Obtén temas y suscripciones.
    • Muestra una lista de temas y suscripciones.
    • Borra temas y suscripciones.
    • Obtén y establece plazos de confirmación en una suscripción.
  • PubSubTemplate para enviar y recibir mensajes:
    • Publica mensajes en temas.
    • Extrae mensajes de las suscripciones de forma síncrona.
    • Extrae mensajes de las suscripciones de forma asíncrona.
    • Confirma recepción de mensajes.
    • Modifica los plazos de confirmación.
    • Convierte mensajes de Pub/Sub en objetos antiguos y sin formato basados en Java (POJO).

Usa adaptadores de canal de integración de Spring

Si tu aplicación de Spring usa canales de mensaje de integración de Spring, puedes enrutar los mensajes entre tus canales de mensajes y Pub/Sub con adaptadores de canal.

Instala los módulos

A fin de instalar módulos para adaptadores de canal de integración de Spring, agrega lo siguiente a tu archivo pom.xml:

  1. La BOM de Spring Cloud GCP.

    <dependencyManagement>
        <dependency>
          <groupId>org.springframework.cloud</groupId>
          <artifactId>spring-cloud-gcp-dependencies</artifactId>
          <version>1.2.6.RELEASE</version>
          <type>pom</type>
          <scope>import</scope>
        </dependency>
      </dependencies>
    </dependencyManagement>
  2. Los artefactos de Spring Cloud GCP Pub/Sub Starter y Spring Integration Core:

    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-core</artifactId>
    </dependency>

Recibe mensajes de Pub/Sub

Para recibir mensajes de una suscripción de Pub/Sub en tu aplicación de Spring, usa un adaptador de canal entrante. El adaptador de canal entrante convierte los mensajes de Pub/Sub entrantes en POJO y, luego, reenvía los POJO a un canal de mensajes.

// Create a message channel for messages arriving from the subscription `sub-one`.
@Bean
public MessageChannel inputMessageChannel() {
  return new PublishSubscribeChannel();
}

// Create an inbound channel adapter to listen to the subscription `sub-one` and send
// messages to the input message channel.
@Bean
public PubSubInboundChannelAdapter inboundChannelAdapter(
    @Qualifier("inputMessageChannel") MessageChannel messageChannel,
    PubSubTemplate pubSubTemplate) {
  PubSubInboundChannelAdapter adapter =
      new PubSubInboundChannelAdapter(pubSubTemplate, "sub-one");
  adapter.setOutputChannel(messageChannel);
  adapter.setAckMode(AckMode.MANUAL);
  adapter.setPayloadType(String.class);
  return adapter;
}

// Define what happens to the messages arriving in the message channel.
@ServiceActivator(inputChannel = "inputMessageChannel")
public void messageReceiver(
    String payload,
    @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) BasicAcknowledgeablePubsubMessage message) {
  LOGGER.info("Message arrived via an inbound channel adapter from sub-one! Payload: " + payload);
  message.ack();
}

En el ejemplo anterior, se usan los siguientes beans de Spring y el siguiente recurso de Pub/Sub:

  • El bean del canal de mensajes llamado inputMessageChannel.
  • Un bean adaptador de canal entrante llamado inboundChannelAdapter de tipo PubSubInboundChannelAdapter.
  • Un ID de suscripción a Pub/Sub con el nombre sub-one.

El inboundChannelAdapter extrae de forma asíncrona los mensajes de sub-one mediante un PubSubTemplate y los envía a inputMessageChannel.

inboundChannelAdapter establece el modo de confirmación en MANUAL para que la aplicación pueda confirmar los mensajes después de procesarlos. El modo de confirmación predeterminado de los tipos PubSubInboundChannelAdapter es AUTO.

El bean messageReceiver de ServiceActivator registra cada mensaje que llega a inputMessageChannel en la salida estándar y, luego, confirma la recepción del mensaje.

Publica mensajes en Pub/Sub

Para publicar mensajes de un canal de mensajes en un tema de Pub/Sub, usa un adaptador de canal saliente. El adaptador de canal saliente convierte los POJO en mensajes de Pub/Sub y, luego, envía los mensajes a un tema de Pub/Sub.

// Create an outbound channel adapter to send messages from the input message channel to the
// topic `topic-two`.
@Bean
@ServiceActivator(inputChannel = "inputMessageChannel")
public MessageHandler messageSender(PubSubTemplate pubsubTemplate) {
  PubSubMessageHandler adapter = new PubSubMessageHandler(pubsubTemplate, "topic-two");

  adapter.setPublishCallback(
      new ListenableFutureCallback<String>() {
        @Override
        public void onFailure(Throwable throwable) {
          LOGGER.info("There was an error sending the message.");
        }

        @Override
        public void onSuccess(String result) {
          LOGGER.info("Message was sent via the outbound channel adapter to topic-two!");
        }
      });
  return adapter;
}

En el ejemplo anterior, se usan los siguientes beans de Spring y el siguiente recurso de Pub/Sub:

  • El bean del canal de mensajes llamado inputMessageChannel.
  • Un adaptador de canal saliente llamado messageSender, de tipo PubSubMessageHandler,
  • Un ID de tema de Pub/Sub llamado topic-two.

El bean ServiceActivator aplica la lógica en messageSender a cada mensaje en inputMessageChannel.

El PubSubMessageHandler en messageSender publica mensajes en inputMessageChannel con un PubSubTemplate. El PubSubMessageHandler publica mensajes en el tema de Pub/Sub topic-two.

Usa Spring Cloud Stream Binder

Para llamar a la API de Pub/Sub en una aplicación de Spring Cloud Stream, usa el módulo Spring Cloud Pub/Sub Stream Binder.

Instala el módulo

Para instalar el módulo de Spring Cloud Stream Binder, agrega lo siguiente al archivo pom.xml:

  1. La BOM de Spring Cloud GCP.

    <dependencyManagement>
        <dependency>
          <groupId>org.springframework.cloud</groupId>
          <artifactId>spring-cloud-gcp-dependencies</artifactId>
          <version>1.2.6.RELEASE</version>
          <type>pom</type>
          <scope>import</scope>
        </dependency>
      </dependencies>
    </dependencyManagement>
  2. El artefacto de Spring Cloud Stream Binder:

    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-gcp-pubsub-stream-binder</artifactId>
    </dependency>

Recibe mensajes de Pub/Sub

Para usar tu aplicación como un receptor de eventos, configura el vinculador de entrada especificando lo siguiente:

  • Un bean Consumer que define la lógica del manejo de mensajes. Por ejemplo, el siguiente bean Consumer se llama receiveMessageFromTopicTwo:

    // Create an input binder to receive messages from `topic-two` using a Consumer bean.
    @Bean
    public Consumer<Message<String>> receiveMessageFromTopicTwo() {
      return message -> {
        LOGGER.info(
            "Message arrived via an input binder from topic-two! Payload: " + message.getPayload());
      };
    }
  • Un ID de tema de Pub/Sub en el archivo de configuración application.properties. Por ejemplo, el siguiente archivo de configuración usa un ID de tema de Pub/Sub llamado topic-two:

    # Bind the Pub/Sub topic `topic-two` to the Consumer bean
    # `receiveMessageFromTopicTwo`. Your Spring application will
    # automatically create and attach a subscription to the topic.
    spring.cloud.stream.bindings.receiveMessageFromTopicTwo-in-0.destination=topic-two

En el código de ejemplo, se reciben mensajes de Pub/Sub. En el ejemplo, se realizan las acciones siguientes:

  1. Buscar el ID de tema de Pub/Sub topic-two en el destino de vinculación de entrada en application.properties.
  2. Crear una suscripción a Pub/Sub para topic-two.
  3. Usar el nombre de vinculación receiveMessageFromTopicTwo-in-0 para encontrar el bean Consumer llamado receiveMessageFromTopicTwo.
  4. Imprimir los mensajes entrantes en la salida estándar y reconocerlos automáticamente.

Publica mensajes en Pub/Sub

Si quieres usar tu aplicación como fuente de eventos, configura el vinculador de salida mediante la especificación siguiente:

  • Un bean Supplier que define de dónde provienen los mensajes dentro de tu aplicación. Por ejemplo, el siguiente bean Supplier se llama sendMessageToTopicOne:

    // Create an output binder to send messages to `topic-one` using a Supplier bean.
    @Bean
    public Supplier<Flux<Message<String>>> sendMessageToTopicOne() {
      return () ->
          Flux.<Message<String>>generate(
              sink -> {
                try {
                  Thread.sleep(10000);
                } catch (InterruptedException e) {
                  // stop sleep earlier.
                }
    
                Message<String> message =
                    MessageBuilder.withPayload("message-" + rand.nextInt(1000)).build();
                LOGGER.info(
                    "Sending a message via the output binder to topic-one! Payload: "
                        + message.getPayload());
                sink.next(message);
              })
              .subscribeOn(Schedulers.elastic());
    }
  • Un ID de tema de Pub/Sub en el archivo de configuración application.properties. Por ejemplo, el siguiente archivo de configuración usa un ID de tema de Pub/Sub llamado topic-one:

    # Bind the Supplier bean `sendMessageToTopicOne` to the Pub/Sub topic
    # `topic-one`. If the topic does not exist, one will be created.
    spring.cloud.stream.bindings.sendMessageToTopicOne-out-0.destination=topic-one

El código de ejemplo publica mensajes en Pub/Sub. En el ejemplo, se realizan las acciones siguientes:

  1. Busca el ID de tema de Pub/Sub topic-one en el destino de vinculación de salida en application.properties.
  2. Usa el nombre de vinculación sendMessageToTopicOne-out-0 para encontrar el bean Supplier llamado sendMessageToTopicOne.
  3. Envía un mensaje numerado a topic-one cada 10 segundos.