Trong bài viết trước, mình đã hướng dẫn các bạn cách publish một message vào Google Pub/Sub topic sử dụng Spring Cloud GCP là như thế nào? Trong bài viết này, chúng ta hãy cùng nhau tìm hiểu cách subscribe nó để nhận message với Spring Cloud GCP các bạn nhé!
Mình cũng tạo một Spring Boot với GCP Message dependency như sau:
Kết quả:
Mình cũng sẽ implement interface CommandLineRunner cho class SpringGcpPubsubSubscribeApplication như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
package com.huongdanjava.springgcppubsub.subscribe; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class SpringGcpPubsubSubscribeApplication implements CommandLineRunner { public static void main(String[] args) { SpringApplication.run(SpringGcpPubsubSubscribeApplication.class, args); } @Override public void run(String... args) throws Exception { } } |
Để subscribe và nhận message từ GCP Pub/Sub topic, chúng ta cũng inject class PubSubTemplate và sử dụng phương thức subscribe() của đối tượng này.
Tham số của phương thức subscribe() này lần lượt là tên subscription được gắn với topic và một Consumer object với type là BasicAcknowledgeablePubsubMessage để consume message, cụ thể của mình như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
package com.huongdanjava.springgcppubsub.subscribe; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import com.google.cloud.spring.pubsub.core.PubSubTemplate; import com.google.pubsub.v1.PubsubMessage; @SpringBootApplication public class SpringGcpPubsubSubscribeApplication implements CommandLineRunner { @Autowired private PubSubTemplate pubSubTemplate; public static void main(String[] args) { SpringApplication.run(SpringGcpPubsubSubscribeApplication.class, args); } @Override public void run(String... args) throws Exception { String subscription = "projects/crested-trainer-341707/subscriptions/huongdanjava-sub"; pubSubTemplate.subscribe(subscription, c -> { PubsubMessage pubsubMessage = c.getPubsubMessage(); System.out.println(pubsubMessage.getData().toStringUtf8()); }); Thread.sleep(100000L); } } |
Data nhận được từ GCP Pub/Sub topic là một đối tượng PubsubMessage của thư viện PubSub client. Các bạn có thể lấy được chính xác nội dung message sau khi getData() và gọi phương thức toStringUtf8() như mình làm như trên.
Cho phần authentication với GCP Pub/Sub, mình sẽ sử dụng property spring.cloud.gcp.credentials.location cùng với property spring.cloud.gcp.project-id, như sau:
|
1 2 |
spring.cloud.gcp.project-id=crested-trainer-341707 spring.cloud.gcp.credentials.location=file:/Users/khanh/Downloads/crested-trainer-341707-3c56ce84473e.json |
Kết quả khi mình public một message trên Google Pub/Sub topic:
như sau:
Thông thường sau khi nhận và đã xử lý message từ GCP Pub/Sub topic, không có lỗi gì xảy ra, chúng ta sẽ send một acknowledgement để GCP Pub/Sub topic xoá message đó đi. Các bạn có thể gọi phương thức ack() của đối tượng BasicAcknowledgeablePubsubMessage để làm việc này, như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
package com.huongdanjava.springgcppubsub.subscribe; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.CommandLineRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import com.google.cloud.spring.pubsub.core.PubSubTemplate; import com.google.pubsub.v1.PubsubMessage; @SpringBootApplication public class SpringGcpPubsubSubscribeApplication implements CommandLineRunner { @Autowired private PubSubTemplate pubSubTemplate; public static void main(String[] args) { SpringApplication.run(SpringGcpPubsubSubscribeApplication.class, args); } @Override public void run(String... args) throws Exception { String subscription = "projects/crested-trainer-341707/subscriptions/huongdanjava-sub"; pubSubTemplate.subscribe(subscription, c -> { PubsubMessage pubsubMessage = c.getPubsubMessage(); System.out.println(pubsubMessage.getData().toStringUtf8()); c.ack(); }); Thread.sleep(100000L); } } |
Lúc này, nếu chạy lại ví dụ, các bạn sẽ thấy message mà chúng ta publish vào topic chỉ được consume và in ra một lần thay vì hai lần ở trên:
