Time to sync up! Tune in for Google I/O on May 14, 2024. Register now.

使用 Spring Integration 和 Google Cloud Pub/Sub 进行消息传递

Spring Integration 为您提供了通过 MessageChannels 交换 Messages 的消息传递机制。它使用通道适配器与外部系统进行通信。

在本练习中，我们将创建两个应用，它们使用 Spring Cloud GCP 提供的 Spring Integration 通道适配器进行通信。Spring Integration 通过这些适配器使用 Google Cloud Pub/Sub 作为消息交换后端。

您将学习如何使用 Cloud Shell 和 Cloud SDK gcloud 命令。

本教程使用 Spring Boot 入门指南中的示例代码。

学习内容

如何使用 Spring Integration 和 Spring Cloud GCP 在应用与 Google Cloud Pub/Sub 之间交换消息

所需条件

一个 Google Cloud Platform 项目
一个浏览器，例如 Chrome 或 Firefox
熟悉标准的 Linux 文本编辑器，例如 Vim、EMACs 或 Nano

您将如何使用本教程？

仅阅读教程内容

阅读并完成练习

您如何评价自己在构建 HTML/CSS Web 应用方面的经验水平？

新手水平

中等水平

熟练水平

您如何评价自己在使用 Google Cloud Platform 服务方面的经验水平？

新手水平

中等水平

熟练水平

自定进度的环境设置

如果您还没有 Google 帐号（Gmail 或 Google Apps），则必须创建一个。登录 Google Cloud Platform Console (console.cloud.google.com) 并创建一个新项目：

2016-02-10 12:45:26 的屏幕截图.png

请记住项目 ID，它在所有 Google Cloud 项目中都是唯一名称（很抱歉，上述名称已被占用，您无法使用！）。它稍后将在此 Codelab 中被称为 PROJECT_ID。

接下来，您需要在 Cloud Console 中启用结算功能，才能使用 Google Cloud 资源。

在此 Codelab 中运行仅花费几美元，但是如果您决定使用更多资源或继续让它们运行，费用可能更高（请参阅本文档末尾的“清理”部分）。

Google Cloud Platform 的新用户有资格获享 $300 免费试用。

Google Cloud Shell

虽然 Google Cloud 可以从笔记本电脑远程操作，但在此 Codelab 中，我们将使用 Google Cloud Shell，这是一个在云端运行的命令行环境。

激活 Google Cloud Shell

在 GCP 控制台中，点击右上角工具栏上的 Cloud Shell 图标：

然后点击“启动 Cloud Shell”：

配置和连接到环境应该只需要片刻时间：

这个虚拟机已加载了您需要的所有开发工具。它提供了一个永久性的 5GB 主目录，并且在 Google Cloud 上运行，从而大大增强了网络性能和身份验证功能。只需使用一个浏览器或 Google Chromebook 即可完成本实验中的大部分（甚至全部）工作。

在连接到 Cloud Shell 后，您应该会看到自己已通过身份验证，并且相关项目已设置为您的 PROJECT_ID。

在 Cloud Shell 中运行以下命令以确认您已通过身份验证：

gcloud auth list

命令输出

Credentialed accounts:
 - <myaccount>@<mydomain>.com (active)

gcloud config list project

命令输出

[core]
project = <PROJECT_ID>

如果不是上述结果，您可以使用以下命令进行设置：

gcloud config set project <PROJECT_ID>

命令输出

Updated property [core/project].

导航到 Google Cloud Pub/Sub 主题页面并启用该 API。

点击创建主题。

输入 exampleTopic 作为主题名称，然后点击创建。

创建主题后，留在“主题”页面中。找到您刚刚创建的主题，按行末的三个垂直点，然后点击新建订阅。

在订阅名称文本框中输入 exampleSubscription，然后点击创建。

在 Cloud Shell 启动后，您可以使用 Spring Initializr 通过命令行生成两个新的 Spring Boot 应用：

$ curl https://start.spring.io/starter.tgz \
  -d bootVersion=2.0.6.RELEASE \
  -d dependencies=web \
  -d baseDir=spring-integration-sender | tar -xzvf -
$ curl https://start.spring.io/starter.tgz \
  -d bootVersion=2.0.6.RELEASE \
  -d baseDir=spring-integration-receiver | tar -xzvf -

现在，我们来创建消息发送应用。切换到正在发送的应用的目录。

$ cd spring-integration-sender

我们希望应用向通道写入消息。消息进入通道后，将由出站通道适配器提取。该适配器将消息从常规 Spring 消息转换为 Google Cloud Pub/Sub 消息，并将其发布到 Google Cloud Pub/Sub 主题。

为了让我们的应用向通道写入数据，可以使用 Spring Integration 消息传递网关。使用来自 vim、emacs 或 nano 的文本编辑器在 DemoApplication 类中声明 PubsubOutboundGateway 接口。

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.integration.annotation.MessagingGateway;

@SpringBootApplication
public class DemoApplication {

        ...

        @MessagingGateway(defaultRequestChannel = "pubsubOutputChannel")
        public interface PubsubOutboundGateway {

                void sendToPubsub(String text);
        }
}

我们现在有一种向通道发送消息的机制，但是这些消息在进入通道后去了哪里？

我们需要一个出站通道适配器来接收通道中的新消息，并将其发布到 Google Cloud Pub/Sub 主题。

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.cloud.gcp.pubsub.core.PubSubTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.cloud.gcp.pubsub.integration.outbound.PubSubMessageHandler;
import org.springframework.messaging.MessageHandler;

@SpringBootApplication
public class DemoApplication {

        ...

        @Bean
        @ServiceActivator(inputChannel = "pubsubOutputChannel")
        public MessageHandler messageSender(PubSubTemplate pubsubTemplate) {
                return new PubSubMessageHandler(pubsubTemplate, "exampleTopic");
        }
}

@ServiceActivator 注释可将此 MessageHandler 应用于 inputChannel 中的任何新消息。在本例中，我们将调用出站通道适配器 PubSubMessageHandler，将消息发布到 Google Cloud Pub/Sub 的 exampleTopic 主题。

设置好通道适配器后，我们现在可以自动连接 PubsubOutboundGateway 对象，并使用它向通道写入消息。

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.servlet.view.RedirectView;

@SpringBootApplication
public class DemoApplication {

...

        @Autowired
        private PubsubOutboundGateway messagingGateway;

        @PostMapping("/postMessage")
        public RedirectView postMessage(@RequestParam("message") String message) {
                this.messagingGateway.sendToPubsub(message);
                return new RedirectView("/");
        }
}

得益于 @PostMapping 注释，我们现在有了一个端点来侦听 HTTP POST 请求，但还要向 DemoApplication 类添加 @RestController 注释来将其标记为 REST 控制器。

src/main/java/com/example/demo/DemoApplication.java

import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
@RestController
public class DemoApplication {
  ...
}

我们只需添加所需的依赖项就能让应用运行。

pom.xml

<project>
  ...
  <!-- Add Spring Cloud GCP Dependency BOM -->
  <dependencyManagement>
        <dependencies>
                <dependency>
                        <groupId>org.springframework.cloud</groupId>
                        <artifactId>spring-cloud-gcp-dependencies</artifactId>
                        <version>1.0.0.RELEASE</version>
                        <type>pom</type>
                        <scope>import</scope>
                </dependency>
        </dependencies>
  </dependencyManagement>

  <dependencies>
        ...
        <!-- Add Pub/Sub Starter -->
        <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
        </dependency>

        <!-- Add Spring Integration -->
        <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-core</artifactId>
        </dependency>

  </dependencies>

</project>

运行发送者应用。

# Set the Project ID in environmental variable
$ export GOOGLE_CLOUD_PROJECT=`gcloud config list \
  --format 'value(core.project)'`
$ ./mvnw spring-boot:run

该应用正在侦听包含端口 8080 和端点 /postMessage 上消息的 POST 请求，但我们稍后会对此进行说明。

我们刚刚创建了一个通过 Google Cloud Pub/Sub 发送消息的应用。现在，我们将再创建一个接收这些消息并进行处理的应用。

点击 + 打开新的 Cloud Shell 会话。

然后，在新的 Cloud Shell 会话中，将目录更改为接收者应用的目录：

$ cd spring-integration-receiver

在上一个应用中，消息传递网关声明为我们创建了出站通道。由于我们不使用消息传递网关来接收消息，因此需要声明自己的 MessageChannel（收到的消息将到达此处）。

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;

@SpringBootApplication
public class DemoApplication {

        ...

        @Bean
        public MessageChannel pubsubInputChannel() {
                return new DirectChannel();
        }
}

我们需要入站通道适配器接收来自 Google Cloud Pub/Sub 的消息，并将其中继到 pubsubInputChannel。

src/main/java/com/example/demo/DemoApplication.java

...
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.gcp.pubsub.core.PubSubTemplate;
import org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubInboundChannelAdapter;

@SpringBootApplication
public class DemoApplication {
        ...

        @Bean
        public PubSubInboundChannelAdapter messageChannelAdapter(
                        @Qualifier("pubsubInputChannel") MessageChannel inputChannel,
                        PubSubTemplate pubSubTemplate) {
                PubSubInboundChannelAdapter adapter =
                                new PubSubInboundChannelAdapter(pubSubTemplate, "exampleSubscription");
                adapter.setOutputChannel(inputChannel);

                return adapter;
        }
}

此适配器会将其自身绑定到 pubsubInputChannel，并侦听 Google Cloud Pub/Sub exampleSubscription 订阅中的新消息。

我们有一个通道，收到的消息会发布到其中，但该如何处理这些消息呢？

让我们使用在 pubsubInputChannel 收到新消息时触发的 @ServiceActivator 来处理它们。

src/main/java/com/example/demo/DemoApplication.java

...
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.annotation.ServiceActivator;

@SpringBootApplication
public class DemoApplication {

        ...

        private static final Log LOGGER = LogFactory.getLog(DemoApplication.class);

        @ServiceActivator(inputChannel = "pubsubInputChannel")
        public void messageReceiver(String payload) {
                LOGGER.info("Message arrived! Payload: " + payload);
        }
}

在本例中，我们只需记录消息载荷。

我们需要添加必要的依赖项。

pom.xml

<project>
  ...
  <!-- Add Spring Cloud GCP Dependency BOM -->
  <dependencyManagement>
        <dependencies>
                <dependency>
                        <groupId>org.springframework.cloud</groupId>
                        <artifactId>spring-cloud-gcp-dependencies</artifactId>
                        <version>1.0.0.RELEASE</version>
                        <type>pom</type>
                        <scope>import</scope>
                </dependency>
        </dependencies>
  </dependencyManagement>

  <dependencies>
        ...
        <!-- Add Pub/Sub Starter -->
        <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
        </dependency>

        <!-- Add Spring Integration -->
        <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-core</artifactId>
        </dependency>

  </dependencies>

</project>

运行接收者应用。

$ ./mvnw spring-boot:run

现在，您发送到发送者应用的任何消息都会记录在接收者应用中。如需进行测试，请打开新的 Cloud Shell 会话，并向发送者应用发出 HTTP POST 请求。

$ curl --data "message=Hello world!" localhost:8080/postMessage

然后，验证接收者应用是否记录了您发送的消息！

INFO: Message arrived! Payload: Hello world!

删除在此练习中创建的订阅和主题。

$ gcloud beta pubsub subscriptions delete exampleSubscription
$ gcloud beta pubsub topics delete exampleTopic

您设置了两个 Spring Boot 应用，它们使用适用于 Google Cloud Pub/Sub 的 Spring Integration 通道适配器。它们彼此之间交换消息，而无需与 Google Cloud Pub/Sub API 进行交互。

您已经了解如何使用适用于 Google Cloud Pub/Sub 的 Spring Integration 通道适配器！

了解详情

Google Cloud Pub/Sub：https://cloud.google.com/pubsub/
GCP 上的 Spring 项目：http://cloud.spring.io/spring-cloud-gcp/
GCP GitHub 代码库上的 Spring：https://github.com/spring-cloud/spring-cloud-gcp
Google Cloud Platform 上的 Java：https://cloud.google.com/java/

许可

此作品已获得 Creative Commons Attribution 2.0 通用许可授权。