订购Google Cloud Pub/Sub messages - java示例程序

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.core.CredentialsProvider;
import com.google.api.gax.core.FixedCredentialsProvider;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.SubscriptionName;

import java.io.FileInputStream;
import java.util.concurrent.TimeUnit;

public class PubSubSubscriberExample {
    public static void main(String[] args) throws Exception {
        String projectId = "your-project-id";
        String subscriptionId = "your-subscription-id";

        // 设置Google Cloud凭证
        CredentialsProvider credentialsProvider = FixedCredentialsProvider.create(GoogleCredentials.fromStream(new FileInputStream("path/to/credentials.json")));

        // 创建订阅者
        ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(projectId, subscriptionId);
        MessageReceiver receiver = new MessageReceiver() {
            @Override
            public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
                // 处理接收到的消息
                System.out.println("Received message: " + message.getData().toStringUtf8());

                // 手动确认消息已被处理
                consumer.ack();
            }
        };
        Subscriber subscriber = null;
        try {
            subscriber = Subscriber.newBuilder(subscriptionName, receiver)
                    .setCredentialsProvider(credentialsProvider)
                    .build();
            subscriber.start();

            // 等待一段时间后停止订阅
            subscriber.awaitTerminated(30, TimeUnit.SECONDS);
        } finally {
            if (subscriber != null) {
                subscriber.stopAsync();
            }
        }
    }
}