使用 Spring Integration 和 Google Cloud Pub/Sub 傳送訊息

1. 總覽

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、EMAC 或 Nano

您會如何使用這項教學課程？

僅閱讀 閱讀並完成練習

您對建構 HTML/CSS 網頁應用程式的體驗滿意嗎？

新手 中級 熟練

您對使用 Google Cloud Platform 服務的體驗有何評價？

新手 中級 熟練

2. 設定和需求

自修實驗室環境設定

  1. 登入 Google Cloud 控制台，然後建立新專案或重複使用現有專案。如果沒有 Gmail 或 Google Workspace 帳戶，請先建立帳戶。

b35bf95b8bf3d5d8.png

a99b7ace416376c4.png

bd84a6d3004737c5.png

  • 專案名稱是這個專案參與者的顯示名稱。這是 Google API 未使用的字元字串。你隨時可以更新。
  • 專案 ID 在所有 Google Cloud 專案中都是不重複的，而且設定後即無法變更。Cloud 控制台會自動產生專屬字串，通常您不需要在意該字串為何。在大多數程式碼研究室中，您需要參照專案 ID (通常標示為 PROJECT_ID)。如果您不喜歡產生的 ID，可以產生另一個隨機 ID。你也可以嘗試使用自己的名稱，看看是否可用。完成這個步驟後就無法變更，且專案期間會維持不變。
  • 請注意，有些 API 會使用第三個值，也就是「專案編號」。如要進一步瞭解這三種值，請參閱說明文件。
  1. 接著，您需要在 Cloud 控制台中啟用帳單，才能使用 Cloud 資源/API。完成這個程式碼研究室的費用不高，甚至可能完全免費。如要關閉資源，避免在本教學課程結束後繼續產生費用，請刪除您建立的資源或專案。Google Cloud 新使用者可參加 $300 美元的免費試用計畫。

Google Cloud Shell

雖然您可以透過筆電遠端操作 Google Cloud，但在本程式碼研究室中，我們將使用 Google Cloud Shell，這是可在雲端執行的指令列環境。

啟用 Cloud Shell

  1. 在 Cloud 控制台，點選「啟用 Cloud Shell」 圖示 853e55310c205094.png。

55efc1aaa7a4d3ad.png

如果您是首次啟動 Cloud Shell，系統會顯示中繼畫面，說明這個指令列環境。如果出現中繼畫面，請按一下「繼續」。

9c92662c6a846a5c.png

佈建並連至 Cloud Shell 預計只需要幾分鐘。

9f0e51b578fecce5.png

這部虛擬機器已載入所有必要的開發工具，並提供永久的 5 GB 主目錄，而且可在 Google Cloud 運作，大幅提升網路效能並強化驗證功能。本程式碼研究室幾乎所有工作都可在瀏覽器上完成。

連至 Cloud Shell 後，您應該會看到驗證已完成，專案也已設為獲派的專案 ID。

  1. 在 Cloud Shell 中執行下列指令，確認您已通過驗證：
gcloud auth list

指令輸出

 Credentialed Accounts
ACTIVE  ACCOUNT
*       <my_account>@<my_domain.com>

To set the active account, run:
    $ gcloud config set account `ACCOUNT`
  1. 在 Cloud Shell 中執行下列指令，確認 gcloud 指令知道您的專案：
gcloud config list project

指令輸出

[core]
project = <PROJECT_ID>

如未設定，請輸入下列指令手動設定專案：

gcloud config set project <PROJECT_ID>

指令輸出

Updated property [core/project].

3. 佈建 Pub/Sub 資源

前往 Google Cloud Pub/Sub 主題頁面。

按一下 [Create Topic] (建立主題)。

4c938409dc7169a6.png

輸入 exampleTopic 做為主題名稱，然後按一下「建立」。

e2daeec91537f672.png

建立主題後，請留在「主題」頁面。找到剛建立的主題，按下行尾的三個垂直點，然後點選「新增訂閱」。

975efa26e5054936.png

在訂閱名稱文字方塊中輸入 exampleSubscription，然後按一下「建立」。

f7a91d9e1cb48009.png

4. 初始化 Spring Boot 應用程式

啟動 Cloud Shell 後，您可以使用指令列，透過 Spring Initializr 產生兩個新的 Spring Boot 應用程式：

$ curl https://start.spring.io/starter.tgz \
  -d bootVersion=3.0.5 \
  -d dependencies=web,integration,cloud-gcp-pubsub \
  -d type=maven-project \
  -d baseDir=spring-integration-sender | tar -xzvf -

$ curl https://start.spring.io/starter.tgz \
  -d bootVersion=3.0.5 \
  -d dependencies=web,integration,cloud-gcp-pubsub \
  -d type=maven-project \
  -d baseDir=spring-integration-receiver | tar -xzvf -

5. 建立應用程式來傳送訊息

現在來建立訊息傳送應用程式。切換至傳送應用程式的目錄。

$ 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 com.google.cloud.spring.pubsub.core.PubSubTemplate;
import com.google.cloud.spring.pubsub.integration.outbound.PubSubMessageHandler;

import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
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 要求的端點，但必須同時將 @RestController 註解新增至 DemoApplication 類別，將其標示為 REST 控制器。

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

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

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

確認 JAVA_HOME 已設為正確版本。

export JAVA_HOME=/usr/lib/jvm/java-1.17.0-openjdk-amd64

執行傳送端應用程式。

# 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 要求，但稍後會再說明這點。

6. 建立應用程式以接收訊息

我們剛建立的應用程式會透過 Google Cloud Pub/Sub 傳送訊息。現在，我們要建立另一個應用程式，接收並處理這些訊息。

按一下「+」，開啟新的 Cloud Shell 工作階段。

9799bee5fea95aa6.png

接著，在新的 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 com.google.cloud.spring.pubsub.core.PubSubTemplate;
import com.google.cloud.spring.pubsub.integration.inbound.PubSubInboundChannelAdapter;

import org.springframework.beans.factory.annotation.Qualifier;

@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 訂閱項目中的新訊息。

我們有一個頻道，會發布收到的訊息，但該如何處理這些訊息呢？

讓我們使用 @ServiceActivator 處理這些訊息，該函式會在新訊息送達 pubsubInputChannel 時觸發。在本例中，我們只會記錄訊息酬載。

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

確認 JAVA_HOME 已設為正確版本。

export JAVA_HOME=/usr/lib/jvm/java-1.17.0-openjdk-amd64

執行接收器應用程式。

$ ./mvnw spring-boot:run -Dspring-boot.run.jvmArguments="-Dserver.port=8081"

現在，您傳送至傳送端應用程式的任何訊息，都會記錄在接收端應用程式中。如要測試，請開啟新的 Cloud Shell 工作階段，並向傳送端應用程式發出 HTTP POST 要求。

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

然後，確認接收端應用程式是否已記錄您傳送的訊息！

INFO: Message arrived! Payload: Hello world!

7. 清除

刪除您在本練習中建立的訂閱項目和主題。

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

8. 摘要

您設定了兩個 Spring Boot 應用程式，使用適用於 Google Cloud Pub/Sub 的 Spring Integration 管道轉接程式。這些應用程式會彼此交換訊息，完全不會與 Google Cloud Pub/Sub API 互動。

9. 恭喜！

您已瞭解如何使用 Google Cloud Pub/Sub 的 Spring Integration 管道轉接程式！

瞭解詳情

授權

這項內容採用的授權為 Creative Commons 姓名標示 2.0 通用授權。