Membuat langganan Lite yang menulis pesan ke topik Pub/Sub
Jelajahi lebih lanjut
Untuk dokumentasi mendetail yang menyertakan contoh kode ini, lihat artikel berikut:
Contoh kode
Go
Untuk melakukan autentikasi ke Pub/Sub Lite, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.
import (
"context"
"fmt"
"io"
"cloud.google.com/go/pubsublite"
)
func createPubsubExportSubscription(w io.Writer, projectID, region, location, topicID, subID, pubsubTopicID string) error {
// projectID := "my-project-id"
// region := "us-central1"
// NOTE: location can be either a region ("us-central1") or a zone ("us-central1-a")
// For a list of valid locations, see https://cloud.google.com/pubsub/lite/docs/locations.
// location := "us-central1"
// NOTE: topic and subscription must be in the same region/zone (e.g. "us-central1-a")
// topicID := "my-topic"
// subID := "my-subscription"
// pubsubTopicID := "destination-topic-id"
ctx := context.Background()
client, err := pubsublite.NewAdminClient(ctx, region)
if err != nil {
return fmt.Errorf("pubsublite.NewAdminClient: %w", err)
}
defer client.Close()
// Initialize the subscription to the oldest retained messages for each
// partition.
targetLocation := pubsublite.AtTargetLocation(pubsublite.Beginning)
sub, err := client.CreateSubscription(ctx, pubsublite.SubscriptionConfig{
Name: fmt.Sprintf("projects/%s/locations/%s/subscriptions/%s", projectID, location, subID),
Topic: fmt.Sprintf("projects/%s/locations/%s/topics/%s", projectID, location, topicID),
DeliveryRequirement: pubsublite.DeliverImmediately, // Can also be DeliverAfterStored.
// Configures an export subscription that writes messages to a Pub/Sub topic.
ExportConfig: &pubsublite.ExportConfig{
DesiredState: pubsublite.ExportActive, // Can also be ExportPaused.
Destination: &pubsublite.PubSubDestinationConfig{
Topic: fmt.Sprintf("projects/%s/topics/%s", projectID, pubsubTopicID),
},
},
}, targetLocation)
if err != nil {
return fmt.Errorf("client.CreateSubscription got err: %w", err)
}
fmt.Fprintf(w, "Created export subscription: %s\n", sub.Name)
return nil
}
Java
Untuk melakukan autentikasi ke Pub/Sub Lite, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.
import com.google.api.gax.rpc.AlreadyExistsException;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.BacklogLocation;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.CloudRegionOrZone;
import com.google.cloud.pubsublite.CloudZone;
import com.google.cloud.pubsublite.ProjectNumber;
import com.google.cloud.pubsublite.SeekTarget;
import com.google.cloud.pubsublite.SubscriptionName;
import com.google.cloud.pubsublite.SubscriptionPath;
import com.google.cloud.pubsublite.TopicName;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.proto.ExportConfig;
import com.google.cloud.pubsublite.proto.ExportConfig.PubSubConfig;
import com.google.cloud.pubsublite.proto.ExportConfig.State;
import com.google.cloud.pubsublite.proto.Subscription;
import com.google.cloud.pubsublite.proto.Subscription.DeliveryConfig;
import com.google.cloud.pubsublite.proto.Subscription.DeliveryConfig.DeliveryRequirement;
import java.util.concurrent.ExecutionException;
public class CreatePubsubExportSubscriptionExample {
public static void main(String... args) throws Exception {
// TODO(developer): Replace these variables before running the sample.
String cloudRegion = "your-cloud-region";
char zoneId = 'b';
String topicId = "your-topic-id";
String subscriptionId = "your-subscription-id";
String pubsubTopicId = "destination-topic-id";
long projectNumber = Long.parseLong("123456789");
// True if using a regional location. False if using a zonal location.
// https://cloud.google.com/pubsub/lite/docs/topics
boolean regional = false;
createPubsubExportSubscriptionExample(
cloudRegion, zoneId, projectNumber, topicId, subscriptionId, pubsubTopicId, regional);
}
public static void createPubsubExportSubscriptionExample(
String cloudRegion,
char zoneId,
long projectNumber,
String topicId,
String subscriptionId,
String pubsubTopicId,
boolean regional)
throws Exception {
CloudRegionOrZone location;
if (regional) {
location = CloudRegionOrZone.of(CloudRegion.of(cloudRegion));
} else {
location = CloudRegionOrZone.of(CloudZone.of(CloudRegion.of(cloudRegion), zoneId));
}
TopicPath topicPath =
TopicPath.newBuilder()
.setProject(ProjectNumber.of(projectNumber))
.setLocation(location)
.setName(TopicName.of(topicId))
.build();
SubscriptionPath subscriptionPath =
SubscriptionPath.newBuilder()
.setLocation(location)
.setProject(ProjectNumber.of(projectNumber))
.setName(SubscriptionName.of(subscriptionId))
.build();
com.google.pubsub.v1.TopicName pubsubTopicName =
com.google.pubsub.v1.TopicName.of(String.valueOf(projectNumber), pubsubTopicId);
Subscription subscription =
Subscription.newBuilder()
.setDeliveryConfig(
// Possible values for DeliveryRequirement:
// - `DELIVER_IMMEDIATELY`
// - `DELIVER_AFTER_STORED`
// You may choose whether to wait for a published message to be successfully written
// to storage before the server delivers it to subscribers. `DELIVER_IMMEDIATELY` is
// suitable for applications that need higher throughput.
DeliveryConfig.newBuilder()
.setDeliveryRequirement(DeliveryRequirement.DELIVER_IMMEDIATELY))
.setExportConfig(
// Configures an export subscription that writes messages to a Pub/Sub topic.
ExportConfig.newBuilder()
// Possible values for State:
// - `ACTIVE`: enable message processing.
// - `PAUSED`: suspend message processing.
.setDesiredState(State.ACTIVE)
.setPubsubConfig(
PubSubConfig.newBuilder().setTopic(pubsubTopicName.toString())))
.setName(subscriptionPath.toString())
.setTopic(topicPath.toString())
.build();
// Target location within the message backlog that the subscription should be initialized to.
SeekTarget initialLocation = SeekTarget.of(BacklogLocation.BEGINNING);
AdminClientSettings adminClientSettings =
AdminClientSettings.newBuilder().setRegion(location.extractRegion()).build();
// Initialize client that will be used to send requests. This client only needs to be created
// once, and can be reused for multiple requests. After completing all of your requests, call
// the "close" method on the client to safely clean up any remaining background resources, or
// use "try-with-close" statement to do this automatically.
try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
Subscription response = adminClient.createSubscription(subscription, initialLocation).get();
System.out.println(response.getAllFields() + " created successfully.");
} catch (ExecutionException e) {
try {
throw e.getCause();
} catch (AlreadyExistsException alreadyExists) {
System.out.println("This subscription already exists.");
} catch (Throwable throwable) {
throwable.printStackTrace();
}
}
}
}Python
Untuk melakukan autentikasi ke Pub/Sub Lite, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.
from google.api_core.exceptions import AlreadyExists
from google.cloud.pubsub_v1 import PublisherClient
from google.cloud.pubsublite import AdminClient, Subscription, ExportConfig
from google.cloud.pubsublite.types import (
BacklogLocation,
CloudRegion,
CloudZone,
SubscriptionPath,
TopicPath,
)
def create_lite_pubsub_export_subscription(
project_number,
cloud_region="us-central1",
zone_id="a",
topic_id="my-topic-id",
subscription_id="my-subscription-id",
pubsub_topic_id="destination-topic-id",
regional=True,
target_location=BacklogLocation.BEGINNING,
):
if regional:
location = CloudRegion(cloud_region)
else:
location = CloudZone(CloudRegion(cloud_region), zone_id)
topic_path = TopicPath(project_number, location, topic_id)
subscription_path = SubscriptionPath(project_number, location, subscription_id)
destination_topic_path = PublisherClient.topic_path(project_number, pubsub_topic_id)
subscription = Subscription(
name=str(subscription_path),
topic=str(topic_path),
delivery_config=Subscription.DeliveryConfig(
# Possible values for delivery_requirement:
# - `DELIVER_IMMEDIATELY`
# - `DELIVER_AFTER_STORED`
# You may choose whether to wait for a published message to be successfully written
# to storage before the server delivers it to subscribers. `DELIVER_IMMEDIATELY` is
# suitable for applications that need higher throughput.
delivery_requirement=Subscription.DeliveryConfig.DeliveryRequirement.DELIVER_IMMEDIATELY,
),
# Configures an export subscription that writes messages to a Pub/Sub topic.
export_config=ExportConfig(
# Possible values for desired_state:
# - `ACTIVE`: enable message processing.
# - `PAUSED`: suspend message processing.
desired_state=ExportConfig.State.ACTIVE,
pubsub_config=ExportConfig.PubSubConfig(
topic=destination_topic_path,
),
),
)
# Initialize client that will be used to send requests across threads. This
# client only needs to be created once, and can be reused for multiple requests.
client = AdminClient(cloud_region)
try:
response = client.create_subscription(subscription, target_location)
print(f"{response.name} created successfully.")
except AlreadyExists:
print(f"{subscription_path} already exists.")
Langkah selanjutnya
Untuk menelusuri dan memfilter contoh kode untuk produk Google Cloud lainnya, lihat browser contoh Google Cloud.