การรับส่งข้อความที่มี Spring Integration และ Google Cloud Pub/Sub

การผสานรวมฤดูใบไม้ผลิมีกลไกการรับส่งข้อความเพื่อแลกเปลี่ยน Messages ผ่าน MessageChannels โดยจะใช้อะแดปเตอร์ช่องเพื่อสื่อสารกับระบบภายนอก

ในแบบฝึกหัดนี้ เราจะสร้างแอป 2 แอปที่สื่อสารโดยใช้อะแดปเตอร์ช่องการผสานรวม Spring ซึ่งโดย Spring Cloud GCP อะแดปเตอร์เหล่านี้ทําให้ Spring Integration ใช้ Google Cloud Pub/Sub เป็นแบ็กเอนด์การแลกเปลี่ยนข้อความได้

ดูวิธีใช้ Cloud Shell และคําสั่ง gcloud ของ Cloud SDK

บทแนะนํานี้ใช้ตัวอย่างโค้ดจากคู่มือเริ่มต้นใช้งาน Spring

สิ่งที่จะได้เรียนรู้

• วิธีแลกเปลี่ยนข้อความระหว่างแอปกับ Google Cloud Pub/Sub โดยใช้ Spring Integration และ Spring Cloud GCP

สิ่งที่ต้องมี

• โครงการ Google Cloud Platform
• เบราว์เซอร์ เช่น Chrome หรือ Firefox
• ทําความคุ้นเคยกับเครื่องมือแก้ไขข้อความ Linux มาตรฐาน เช่น Vim, EMAC หรือ Nano

คุณจะใช้บทแนะนํานี้อย่างไร

อ่านจบเท่านั้น อ่านและตอบแบบฝึกหัด

คุณจะให้คะแนนประสบการณ์ในการสร้างเว็บแอป HTML/CSS อย่างไร

มือใหม่ ปานกลาง เชี่ยวชาญ

คุณจะให้คะแนนประสบการณ์ในการใช้บริการของ Google Cloud Platform อย่างไร

มือใหม่ ปานกลาง เชี่ยวชาญ

การตั้งค่าสภาพแวดล้อมด้วยตนเอง

หากยังไม่มีบัญชี Google (Gmail หรือ Google Apps) คุณต้องสร้างบัญชี ลงชื่อเข้าใช้คอนโซล Google Cloud Platform (console.cloud.google.com) และสร้างโปรเจ็กต์ใหม่ ดังนี้

โปรดทราบว่ารหัสโปรเจ็กต์ ซึ่งเป็นชื่อที่ไม่ซ้ํากันสําหรับโปรเจ็กต์ Google Cloud ทั้งหมด (ชื่อข้างต้นมีผู้อื่นนําไปใช้แล้ว ขออภัยในความไม่สะดวก) และจะเรียกใน Codelab นี้ว่า PROJECT_ID ในภายหลัง

จากนั้นคุณจะต้องเปิดใช้การเรียกเก็บเงินใน Cloud Console เพื่อใช้ทรัพยากรของ Google Cloud

การเรียกใช้ Codelab นี้ไม่ควรมีค่าใช้จ่ายเกิน 2-3 ดอลลาร์ แต่อาจมากกว่านั้นหากคุณตัดสินใจใช้ทรัพยากรเพิ่มเติมหรือปล่อยให้ทรัพยากรทํางาน (ดู "cleanup" ในตอนท้ายของเอกสารนี้)

ผู้ใช้ใหม่ของ Google Cloud Platform มีสิทธิ์รับช่วงทดลองใช้ฟรี $300

Google Cloud Shell

แม้ว่า Google Cloud จะทํางานจากแล็ปท็อปได้จากระยะไกล แต่ใน Codelab นี้ เราจะใช้ Google Cloud Shell ซึ่งเป็นสภาพแวดล้อมบรรทัดคําสั่งที่ทํางานในระบบคลาวด์

เปิดใช้งาน Google Cloud Shell

จากคอนโซล GCP ให้คลิกไอคอน Cloud Shell บนแถบเครื่องมือด้านขวาบน ดังนี้

จากนั้นคลิก "เริ่ม Cloud Shell"

ระบบจะใช้เวลาเพียงครู่เดียวในการจัดสรรและเชื่อมต่อกับสภาพแวดล้อม

เครื่องเสมือนนี้โหลดด้วยเครื่องมือการพัฒนาทั้งหมดที่คุณต้องการ โดยมีไดเรกทอรีหน้าแรกขนาด 5 GB ถาวรและทํางานอยู่ใน Google Cloud ซึ่งช่วยปรับปรุงประสิทธิภาพและการตรวจสอบสิทธิ์ของเครือข่ายได้อย่างมาก ไม่เพียงเท่านั้น งานของคุณในห้องทดลองนี้ก็ทําได้โดยใช้เพียงเบราว์เซอร์หรือ Google Chromebook

เมื่อเชื่อมต่อกับ Cloud Shell แล้ว ระบบจะตรวจสอบว่าคุณตรวจสอบสิทธิ์แล้ว และมีการตั้งค่าโปรเจ็กต์เป็น PROJECT_ID แล้ว

เรียกใช้คําสั่งต่อไปนี้ใน Cloud Shell เพื่อยืนยันว่าคุณได้รับการตรวจสอบสิทธิ์

gcloud auth list

เอาต์พุตจากคําสั่ง

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

หมายเหตุ: gcloud คือเครื่องมือบรรทัดคําสั่งแบบรวมที่มีประสิทธิภาพสําหรับ Google Cloud Platform ดูเอกสารประกอบฉบับเต็มได้ที่ https://cloud.google.com/sdk/gcloud โดยมีการติดตั้งไว้ล่วงหน้าบน Cloud Shell คุณจะสังเกตเห็นการรองรับสําหรับการเติมแท็บ

gcloud config list project

เอาต์พุตจากคําสั่ง

[core]
project = <PROJECT_ID>

หากไม่ใช่ คุณจะตั้งค่าได้โดยใช้คําสั่งนี้

gcloud config set project <PROJECT_ID>

เอาต์พุตจากคําสั่ง

Updated property [core/project].

ไปที่หน้าหัวข้อ Google Cloud Pub/Sub และเปิดใช้ API

คลิกสร้างหัวข้อ

พิมพ์ exampleTopic เป็นชื่อหัวข้อ แล้วคลิกสร้าง

หลังจากสร้างหัวข้อแล้ว ให้อยู่ในหน้าหัวข้อ มองหาหัวข้อที่คุณเพิ่งสร้าง กดจุดแนวตั้ง 3 จุดที่ท้ายบรรทัด แล้วคลิกการสมัครใช้บริการใหม่

พิมพ์ exampleSubscription ในกล่องข้อความชื่อการสมัครใช้บริการ แล้วคลิกสร้าง

นอกจากนี้คุณอาจใช้ gcloud CLI เพื่อสร้างหัวข้อและการสมัครใช้บริการใหม่ เช่น

$ gcloud beta pubsub topics create exampleTopic

$ gcloud beta pubsub subscriptions create exampleSubscription --topic exampleTopic

หลังจากเปิดตัว Cloud Shell คุณจะใช้บรรทัดคําสั่งเพื่อสร้างแอปพลิเคชัน Spring Boot ใหม่ 2 แอปพลิเคชันด้วย Spring Initializr ได้ดังนี้

$ 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 เพื่อให้แอปเขียนไปที่ช่องได้ หากใช้เครื่องมือแก้ไขข้อความจาก vim, emacs หรือ nano ให้ประกาศอินเทอร์เฟซ PubsubOutboundGateway ภายในชั้นเรียน DemoApplication

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 ซึ่งเป็นอะแดปเตอร์ช่องทางขาออกเพื่อเผยแพร่ข้อความไปยังหัวข้อ exampleTopic ของ Google Cloud Pub/Sub&#39

เมื่อมีอะแดปเตอร์ช่องแล้ว เราจึงสามารถต่อสายออบเจ็กต์ 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 {
  ...
}

เราต้องการเพิ่มทรัพยากร Dependency ที่จําเป็นเพื่อให้แอปทํางาน

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

แอปกําลังฟังคําขอ POST ที่มีข้อความในพอร์ต 8080 และปลายทาง /postMessage แต่เราจะดําเนินการในภายหลัง

เราเพิ่งสร้างแอปที่ส่งข้อความผ่าน 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

เรามีช่องที่มีการโพสต์ข้อความขาเข้า แต่ควรทําอย่างไรกับข้อความเหล่านั้น

มาประมวลผลข้อความเหล่านั้นด้วย @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);
        }
}

ในกรณีนี้ เราจะบันทึกเพย์โหลดของข้อความเท่านั้น

เราต้องเพิ่มทรัพยากร Dependency ที่จําเป็น

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 2 แอปซึ่งใช้ Spring Integration Channel Adapters สําหรับ Google Cloud Pub/Sub โดยจะแลกเปลี่ยนข้อความกันเองโดยไม่ต้องโต้ตอบกับ Google Cloud Pub/Sub API

คุณได้เรียนรู้วิธีใช้อะแดปเตอร์ช่องการผสานรวม Spring สําหรับ Google Cloud Pub/Sub แล้ว

ดูข้อมูลเพิ่มเติม

• Google Cloud Pub/Sub: https://cloud.google.com/pubsub/
• Spring ในโปรเจ็กต์ GCP: http://cloud.spring.io/spring-cloud-gcp/
• ฤดูใบไม้ผลิบนที่เก็บ GitHub ของ GCP: https://github.com/spring-cloud/spring-cloud-gcp
• Java ใน Google Cloud Platform: https://cloud.google.com/java/

ใบอนุญาต

ผลงานนี้ได้รับอนุญาตภายใต้สัญญาอนุญาตทั่วไปของ Creative Commons Attribution 2.0