我最近在Sprint Central的工程博客上閱讀了Josh Long的Bootiful GCP系列 ，特別喜歡關(guān)于使用Google Cloud的Pub / Sub的第四部分 。 我受到該系列的啟發(fā)，同時(shí)我還在為我的一個(gè)新項(xiàng)目評(píng)估Spring Cloud Stream。 我以為，我會(huì)繼續(xù)討論喬希（Josh）停下的那篇文章。 本文介紹了如何將Spring Cloud Stream與Google Cloud Pub / Sub一起使用，以實(shí)現(xiàn)簡單的生產(chǎn)者和使用者應(yīng)用程序。

介紹

如果您之前閱讀過Josh的文章，則可以安全地跳過此部分。 如果您還沒有這樣做，請(qǐng)不用擔(dān)心，我將在此處快速總結(jié)一些關(guān)鍵點(diǎn)。

什么是Google Cloud Pub / Sub？

Google通過以下方式定義發(fā)布/訂閱 。

Cloud Pub / Sub將面向企業(yè)消息的中間件的可伸縮性，靈活性和可靠性帶到了云中。 通過提供將發(fā)送者和接收者分離的多對(duì)多異步消息傳遞，它可以在獨(dú)立編寫的應(yīng)用程序之間進(jìn)行安全且高度可用的通信。

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

簡而言之，Pub / Sub是Google的解決方案，用于支持開發(fā)人員將應(yīng)用程序組件與Google規(guī)模的消息代理連接起來。 顧名思義，此解決方案使用您期望的相同概念來實(shí)現(xiàn)發(fā)布/訂閱機(jī)制。 可以將郵件提交到主題，并且某個(gè)主題的所有訂閱者都可以接收已發(fā)布的消息。

在這里需要強(qiáng)調(diào)的是，Pub / Sub為每個(gè)提交的消息至少提供一次傳遞。 如果要確保只發(fā)送一次消息，則必須自己照顧。

什么是Spring Integration？

Spring Integration是其投資組合中的一個(gè)Spring項(xiàng)目。 整篇文章甚至整本書都可以寫在上面，因?yàn)樗旧砭褪且粋€(gè)巨大的框架。 總之，Spring Integration是一個(gè)框架，可以幫助您使用EIP模式設(shè)計(jì)和集成應(yīng)用程序。 Spring Integration構(gòu)建的兩個(gè)最基本的原語是Message<T>和MessageChannel 。 在這方面，開發(fā)人員可以使組件彼此分離和隔離。 您可以想到這種機(jī)制，就好像Spring Integration將以某種方式甚至不需要組件彼此了解而是通過交換消息來進(jìn)一步依賴注入的想法一樣。

通道可以將組件彼此連接，如果它們位于相同的JVM中，或者即使它們是由網(wǎng)絡(luò)分布和分隔的。 此時(shí)，要了解的相關(guān)概念是什么是通道適配器。 它們基本上是用來將Spring Framework消息通過消息通道時(shí)轉(zhuǎn)換為一段可由外部系統(tǒng)使用的數(shù)據(jù)。

Spring Integration提供了許多適配器，可以幫助開發(fā)人員連接數(shù)據(jù)庫，消息代理和許多其他外部系統(tǒng)。 在這種情況下，將使用適配器向Google Cloud Pub / Sub提交消息或從Google Cloud Pub / Sub接收消息。 Spring Cloud GCP項(xiàng)目為Pub / Sub提供了入站和出站適配器，從Spring Integration消息流的角度來看，這使得消息交換變得透明。

如果您閱讀Josh的文章 ，他的工作是他正在介紹Spring Integration，以一種干凈，一致的方式使用Pub / Sub。 這意味著將刪除PubSubTemplate的直接引用，因此，如果您想將該文章中的示例修改為例如RabbitMQ，您要做的就是相應(yīng)地替換通道適配器。

什么是Spring Cloud Stream？

消息傳遞非常適合微服務(wù)世界，在微服務(wù)世界中，一組分布式組件相互通信。 由于消息和渠道是Spring Integration中的頭等公民，因此非常適合。 另一方面，Spring Integration是專門為實(shí)現(xiàn)那些EIP模式而設(shè)計(jì)的。

但是，在現(xiàn)代應(yīng)用程序開發(fā)中，我們不一定要與舊系統(tǒng)集成，在這種情況下，我們寧愿與RabbitMQ ， Apache Kafka等現(xiàn)代消息代理集成，也要與GCP Pub / Sub集成。 就是說，就能夠與各種外部系統(tǒng)集成而言，我們不需要Spring Integration的全部功能。 這種額外的靈活性要求我們配置適配器，而這是我們不需要的。 如果我們僅使用GCP Pub / Sub或前面提到的任何其他現(xiàn)代消息代理，那么必須為每個(gè)單個(gè)組件定義和配置適配器就變得很麻煩。

我們確實(shí)希望能夠靈活地處理消息，并且希望利用消息代理，但是我們不想編寫Spring Integration所需的太多代碼。 Spring Cloud Stream建立在Spring Integration之上，并利用了相同的原語（例如消息和通道），但減輕了開發(fā)人員的負(fù)擔(dān)，不必將這些組件連接在一起。 因?yàn)榍朗峭ㄟ^特定于中間件的Binder實(shí)現(xiàn)連接到外部代理的。

將Spring Cloud Stream與Google Cloud Pub / Sub結(jié)合使用

我想我已經(jīng)充分討論了Spring Cloud Stream，Spring Integration和Google Cloud Pub / Sub的背景。 現(xiàn)在該看一些代碼了。 有兩個(gè)非常簡單的Spring Boot應(yīng)用程序，它們交換一個(gè)簡單的字符串作為消息的有效負(fù)載。 讓我們從發(fā)布者開始。

發(fā)行人

這基本上是一個(gè)簡單的控制器，它發(fā)送一個(gè)簡單的String作為消息的有效負(fù)載。 如果您以前使用過Spring Integration，則發(fā)送部分沒有什么特別的。

@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());}}

有趣的是消息通道如何綁定到實(shí)際消息代理的資源。 在第6-8行中，注入了一個(gè)bean（ Channels ），它似乎持有對(duì)傳出消息通道的引用。

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

Channels反過來只是一個(gè)接口，可以定義任意數(shù)量的消息通道并用@Input或@Output標(biāo)記。 Spring Cloud Stream負(fù)責(zé)實(shí)例化一個(gè)代理對(duì)象，該代理對(duì)象負(fù)責(zé)返回對(duì)MessageChannel對(duì)象的引用。

@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注釋標(biāo)記Channels作為一個(gè)可綁定接口和對(duì)一個(gè)邏輯綁定的域名（ outgoing ）與目的地。 目的地的含義因活頁夾的不同而不同，對(duì)于發(fā)布/訂閱，它意味著消息生產(chǎn)者的主題和消息消費(fèi)者的訂閱。 這些綁定可以在application.yml定義。

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

訂戶

訂閱者比發(fā)布者更簡單，它只是一個(gè)類。

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());}}

這里值得一提的是水槽是什么？ 正如我們已經(jīng)看到的， @EnableBinding可以采用接口，然后該框架隱藏了將入站和出站消息適配器連接到消息通道的復(fù)雜性，并且還配置了相關(guān)的基礎(chǔ)結(jié)構(gòu)。 大多數(shù)應(yīng)用程序僅向單個(gè)通道發(fā)送消息或從單個(gè)通道接收消息。 這就是Spring Cloud Stream提供Source ， Sink和Processor接口以幫助您減少代碼的原因。 就是說，我們也可以為發(fā)布者使用Source而不是定義Channels ，但是我想展示框架的功能。

運(yùn)行演示

為了能夠運(yùn)行示例，您需要完成以下步驟。

如果已經(jīng)有一個(gè)，則可以跳過此步驟。

如果您不需要安裝任何軟件，我認(rèn)為會(huì)更容易。 默認(rèn)情況下， Google Cloud Shell隨附了Google Cloud SDK ，Git，Maven和Java。

啟用發(fā)布/訂閱API

由于Spring Cloud Stream是一個(gè)自以為是的框架，因此在其之上構(gòu)建的應(yīng)用程序?qū)⒆孕袆?chuàng)建主題和訂閱。 也就是說，在此處手動(dòng)創(chuàng)建主題和訂閱是可選的。 不過，您必須啟用發(fā)布/訂閱API。

% gcloud services enable pubsub.googleapis.com % gcloud pubsub topics create reservations % gcloud pubsub subscriptions create reservations --topic=reservations

克隆

% git clone https://github.com/springuni/springuni-examples.git

啟動(dòng)發(fā)布者

% cd ~/springuni-examples/spring-cloud/spring-cloud-stream-pubsub-publisher % mvn spring-boot:run

啟動(dòng)訂戶

Google Cloud Shell帶有tmux支持，這也意味著它默認(rèn)情況下會(huì)啟動(dòng)tmux會(huì)話。 當(dāng)然可以禁用。 重要的一點(diǎn)是，您不必打開新的外殼，只需單擊Ctrl-B和C即可打開一個(gè)新窗口。有關(guān)更多詳細(xì)信息，請(qǐng)參閱Tmux鍵綁定 。

% cd ~/springuni-examples/spring-cloud/spring-cloud-stream-pubsub-subscriber % mvn spring-boot:run

發(fā)送信息

像以前一樣再次打開一個(gè)新窗口并發(fā)送消息。

% curl -XPOST http://localhost:8080/publish/test

您應(yīng)該看到訂閱者收到它。

問題

• 您認(rèn)為如果啟動(dòng)更多訂閱者會(huì)發(fā)生什么？
• 他們都會(huì)收到同一條消息還是只收到其中一條？
• 那為什么呢？

在下面發(fā)表評(píng)論，讓我知道您的想法！

結(jié)論

我們已經(jīng)了解了什么是Google Cloud Pub / Sub，什么是Spring Integration，以及為何Spring Cloud Stream建立在Spring Integration上以幫助開發(fā)人員更快地創(chuàng)建消息驅(qū)動(dòng)的微服務(wù)的原因。 在上面的代碼示例中，我進(jìn)一步介紹了Josh的示例，并使用Spring Cloud Stream代替了Spring Integration，最終減少了更多代碼。

翻譯自: https://www.javacodegeeks.com/2018/12/bootiful-spring-cloud-stream.html

總結(jié)

以上是生活随笔為你收集整理的引导性GCP：带有Google Cloud Pub / Sub的Spring Cloud Stream的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。