引导性GCP：带有Google Cloud Pub / Sub的Spring Cloud Stream

我最近在Sprint Central的工程博客上阅读了Josh Long的Bootiful GCP系列，特别喜欢关于使用Google Cloud的Pub / Sub的第四部分。我受到该系列的启发，同时我还在为我的一个新项目评估Spring Cloud Stream。我以为，我会继续讨论乔希（Josh）停下的那篇文章。本文介绍了如何将Spring Cloud Stream与Google Cloud Pub / Sub一起使用，以实现简单的生产者和使用者应用程序。

介绍

如果您之前阅读过Josh的文章，则可以安全地跳过此部分。如果您还没有这样做，请不用担心，我将在此处快速总结一些关键点。

什么是Google Cloud Pub / Sub？

Google通过以下方式定义发布/订阅。

Cloud Pub / Sub将面向企业消息的中间件的可伸缩性，灵活性和可靠性带到了云中。通过提供将发送者和接收者分离的多对多异步消息传递，它可以在独立编写的应用程序之间进行安全且高度可用的通信。

https://cloud.google.com/pubsub/docs/overview

简而言之，Pub / Sub是Google的解决方案，用于支持开发人员将应用程序组件与Google规模的消息代理连接起来。顾名思义，此解决方案使用您期望的相同概念来实现发布/订阅机制。可以将邮件提交到主题，并且某个主题的所有订阅者都可以接收已发布的消息。

在这里需要强调的是，Pub / Sub为每个提交的消息至少提供一次传递。 如果要确保只发送一次消息，则必须自己照顾。

什么是Spring Integration？

Spring Integration是其投资组合中的一个Spring项目。整篇文章甚至整本书都可以写在上面，因为它本身就是一个巨大的框架。总之，Spring Integration是一个框架，可以帮助您使用EIP模式设计和集成应用程序。 Spring Integration构建的两个最基本的原语是Message<T>和MessageChannel 。在这方面，开发人员可以使组件彼此分离和隔离。您可以想到这种机制，就好像Spring Integration将以某种方式甚至不需要组件彼此了解而是通过交换消息来进一步依赖注入的想法一样。

通道可以将组件彼此连接，如果它们位于相同的JVM中，或者即使它们是由网络分布和分隔的。此时，要了解的相关概念是什么是通道适配器。它们基本上是用来将Spring Framework消息通过消息通道时转换为一段可由外部系统使用的数据。

Spring Integration提供了许多适配器，可以帮助开发人员连接数据库，消息代理和许多其他外部系统。在这种情况下，将使用适配器向Google Cloud Pub / Sub提交消息或从Google Cloud Pub / Sub接收消息。 Spring Cloud GCP项目为Pub / Sub提供了入站和出站适配器，从Spring Integration消息流的角度来看，这使得消息交换变得透明。

如果您阅读Josh的文章，他的工作是他正在介绍Spring Integration，以一种干净，一致的方式使用Pub / Sub。这意味着将删除PubSubTemplate的直接引用，因此，如果您想将该文章中的示例修改为例如RabbitMQ，您要做的就是相应地替换通道适配器。

什么是Spring Cloud Stream？

消息传递非常适合微服务世界，在微服务世界中，一组分布式组件相互通信。由于消息和渠道是Spring Integration中的头等公民，因此非常适合。另一方面，Spring Integration是专门为实现那些EIP模式而设计的。

但是，在现代应用程序开发中，我们不一定要与旧系统集成，在这种情况下，我们宁愿与RabbitMQ ， Apache Kafka等现代消息代理集成，也要与GCP Pub / Sub集成。就是说，就能够与各种外部系统集成而言，我们不需要Spring Integration的全部功能。这种额外的灵活性要求我们配置适配器，而这是我们不需要的。如果我们仅使用GCP Pub / Sub或前面提到的任何其他现代消息代理，那么必须为每个单个组件定义和配置适配器就变得很麻烦。

我们确实希望能够灵活地处理消息，并且希望利用消息代理，但是我们不想编写Spring Integration所需的太多代码。 Spring Cloud Stream建立在Spring Integration之上，并利用了相同的原语（例如消息和通道），但减轻了开发人员的负担，不必将这些组件连接在一起。因为渠道是通过特定于中间件的Binder实现连接到外部代理的。

将Spring Cloud Stream与Google Cloud Pub / Sub结合使用

我想我已经充分讨论了Spring Cloud Stream，Spring Integration和Google Cloud Pub / Sub的背景。现在该看一些代码了。有两个非常简单的Spring Boot应用程序，它们交换一个简单的字符串作为消息的有效负载。让我们从发布者开始。

发行人

这基本上是一个简单的控制器，它发送一个简单的String作为消息的有效负载。如果您以前使用过Spring Integration，则发送部分没有什么特别的。

@RestController
public class PublisherController {private final MessageChannel outgoing;public PublisherController(Channels channels) {outgoing = channels.outgoing();}@PostMapping("/publish/{name}")public void publish(@PathVariable String name) {outgoing.send(MessageBuilder.withPayload("Hello " + name + "!").build());}}

有趣的是消息通道如何绑定到实际消息代理的资源。在第6-8行中，注入了一个bean（ Channels ），它似乎持有对传出消息通道的引用。

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;public interface Channels {@OutputMessageChannel outgoing();}

Channels反过来只是一个接口，可以定义任意数量的消息通道并用@Input或@Output标记。 Spring Cloud Stream负责实例化一个代理对象，该代理对象负责返回对MessageChannel对象的引用。

@EnableBinding(Channels.class)
@SpringBootApplication
public class PubsubPublisherApplication {public static void main(String[] args) {SpringApplication.run(PubsubPublisherApplication.class, args);}}

Spring Cloud Stream依赖于Spring Boot和Spring Integration。所述@EnableBinding注释标记Channels作为一个可绑定接口和对一个逻辑绑定的域名（ outgoing ）与目的地。目的地的含义因活页夹的不同而不同，对于发布/订阅，它意味着消息生产者的主题和消息消费者的订阅。这些绑定可以在application.yml定义。

spring:cloud:stream:bindings:outgoing:destination: reservations

订户

订阅者比发布者更简单，它只是一个类。

import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;@Slf4j
@EnableBinding(Sink.class)
@SpringBootApplication
public class PubsubSubscriberApplication {public static void main(String[] args) {SpringApplication.run(PubsubSubscriberApplication.class, args);}@StreamListener(Sink.INPUT)public void handleMessage(Message<String> message) {log.info("Received: {}.", message.getPayload());}}

这里值得一提的是水槽是什么？正如我们已经看到的， @EnableBinding可以采用接口，然后该框架隐藏了将入站和出站消息适配器连接到消息通道的复杂性，并且还配置了相关的基础结构。大多数应用程序仅向单个通道发送消息或从单个通道接收消息。这就是Spring Cloud Stream提供Source ， Sink和Processor接口以帮助您减少代码的原因。就是说，我们也可以为发布者使用Source而不是定义Channels ，但是我想展示框架的功能。

运行演示

为了能够运行示例，您需要完成以下步骤。

1. 如果已经有一个，则可以跳过此步骤。
2. 如果您不需要安装任何软件，我认为会更容易。默认情况下， Google Cloud Shell随附了Google Cloud SDK ，Git，Maven和Java。
3. 启用发布/订阅API
  
  由于Spring Cloud Stream是一个自以为是的框架，因此在其之上构建的应用程序将自行创建主题和订阅。也就是说，在此处手动创建主题和订阅是可选的。不过，您必须启用发布/订阅API。
```
% gcloud services enable pubsub.googleapis.com
% gcloud pubsub topics create reservations
% gcloud pubsub subscriptions create reservations --topic=reservations
```
4. 克隆
```
% git clone https://github.com/springuni/springuni-examples.git
```
5. 启动发布者
```
% cd ~/springuni-examples/spring-cloud/spring-cloud-stream-pubsub-publisher
% mvn spring-boot:run
```
6. 启动订户
  
  Google Cloud Shell带有tmux支持，这也意味着它默认情况下会启动tmux会话。当然可以禁用。重要的一点是，您不必打开新的外壳，只需单击Ctrl-B和C即可打开一个新窗口。有关更多详细信息，请参阅Tmux键绑定。
```
% cd ~/springuni-examples/spring-cloud/spring-cloud-stream-pubsub-subscriber
% mvn spring-boot:run
```
7. 发送信息
  
  像以前一样再次打开一个新窗口并发送消息。
```
% curl -XPOST http://localhost:8080/publish/test
```
  您应该看到订阅者收到它。

问题

您认为如果启动更多订阅者会发生什么？
他们都会收到同一条消息还是只收到其中一条？
那为什么呢？

在下面发表评论，让我知道您的想法！

结论

我们已经了解了什么是Google Cloud Pub / Sub，什么是Spring Integration，以及为何Spring Cloud Stream建立在Spring Integration上以帮助开发人员更快地创建消息驱动的微服务的原因。在上面的代码示例中，我进一步介绍了Josh的示例，并使用Spring Cloud Stream代替了Spring Integration，最终减少了更多代码。