圧縮レートでメッセージをパブリッシュ

パブリッシャーを作成して圧縮を有効にし圧縮バイト数のしきい値を設定して、メッセージをパブリッシュします。

このコードサンプルを含む詳細なドキュメントについては、以下をご覧ください。

メッセージを圧縮

コードサンプル

C++

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある C++ 向けの手順に従って設定を行ってください。詳細については、Pub/Sub C++ API のリファレンスドキュメントをご覧ください。

Pub/Sub に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の認証を設定するをご覧ください。

namespace g = ::google::cloud;
namespace pubsub = ::google::cloud::pubsub;
[](std::string project_id, std::string topic_id) {
  auto topic = pubsub::Topic(std::move(project_id), std::move(topic_id));
  auto publisher = pubsub::Publisher(pubsub::MakePublisherConnection(
      std::move(topic),
      g::Options{}
          // Compress any batch of messages over 10 bytes. By default, no
          // messages are compressed, set this to 0 to compress all batches,
          // regardless of their size.
          .set<pubsub::CompressionThresholdOption>(10)
          // Compress using the GZIP algorithm. By default, the library uses
          // GRPC_COMPRESS_DEFLATE.
          .set<pubsub::CompressionAlgorithmOption>(GRPC_COMPRESS_GZIP)));
  auto message_id = publisher.Publish(
      pubsub::MessageBuilder{}.SetData("Hello World!").Build());
  auto done = message_id.then([](g::future<g::StatusOr<std::string>> f) {
    auto id = f.get();
    if (!id) throw std::move(id).status();
    std::cout << "Hello World! published with id=" << *id << "\n";
  });
  // Block until the message is published
  done.get();
}

C#

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある C# 向けの手順に従って設定を行ってください。詳細については、Pub/Sub C# API のリファレンスドキュメントをご覧ください。


using Google.Cloud.PubSub.V1;
using System.Threading.Tasks;

public class PublishCompressedMessagesAsyncSample
{
    public async Task PublishCompressedMessagesAsync(string projectId, string topicId, string messageText)
    {
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);

        var customSettings = new PublisherClient.Settings
        {
            EnableCompression = true,
            // Compress any batch of messages over 10 bytes. By default,
            // the threshold is taken from PublisherClient.Settings.DefaultCompressionBytesThreshold.
            CompressionBytesThreshold = 10
        };

        PublisherClient publisher = await new PublisherClientBuilder
        {
            TopicName = topicName,
            Settings = customSettings
        }.BuildAsync();

        await publisher.PublishAsync(messageText);
    }
}

Go

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Go 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Go API のリファレンスドキュメントをご覧ください。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub"
)

func publishWithCompression(w io.Writer, projectID, topicID string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	// msg := "Hello World"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub: NewClient: %w", err)
	}
	defer client.Close()

	t := client.Topic(topicID)
	// Enable compression and configure the compression threshold to 10 bytes (default to 240 B).
	// Publish requests of sizes > 10 B (excluding the request headers) will get compressed.
	t.PublishSettings.EnableCompression = true
	t.PublishSettings.CompressionBytesThreshold = 10
	result := t.Publish(ctx, &pubsub.Message{
		Data: []byte("This is a test message"),
	})
	// Block until the result is returned and a server-generated
	// ID is returned for the published message.
	id, err := result.Get(ctx)
	if err != nil {
		return fmt.Errorf("pubsub: result.Get: %w", err)
	}
	fmt.Fprintf(w, "Published a message; msg ID: %v\n", id)
	return nil
}

Java

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Java 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Java API のリファレンスドキュメントをご覧ください。

import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class PublishWithGrpcCompressionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    // Choose an existing topic.
    String topicId = "your-topic-id";

    publishWithGrpcCompressionExample(projectId, topicId);
  }

  public static void publishWithGrpcCompressionExample(String projectId, String topicId)
      throws IOException, ExecutionException, InterruptedException {
    TopicName topicName = TopicName.of(projectId, topicId);

    // Create a publisher and set enable compression to true.
    Publisher publisher = null;
    try {
      // Enable compression and configure the compression threshold to 10 bytes (default to 240 B).
      // Publish requests of sizes > 10 B (excluding the request headers) will get compressed.
      // The number of messages in a publish request is determined by publisher batch settings.
      // Batching is turned off by default, i.e. each publish request contains only one message.
      publisher =
          Publisher.newBuilder(topicName)
              .setEnableCompression(true)
              .setCompressionBytesThreshold(10L)
              .build();

      byte[] bytes = new byte[1024];
      ByteString data = ByteString.copyFrom(bytes);
      PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

      // Once published, returns a server-assigned message id (unique within the topic).
      // You can look up the actual size of the outbound data using the Java Logging API.
      // Configure logging properties as shown in
      // https://github.com/googleapis/java-pubsub/tree/main/samples/snippets/src/main/resources/logging.properties
      // and look for "OUTBOUND DATA" with "length=" in the output log.
      ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
      String messageId = messageIdFuture.get();
      System.out.println("Published a compressed message of message ID: " + messageId);
    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }
}

PHP

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある PHP 向けの手順に従って設定を行ってください。詳細については、Pub/Sub PHP API のリファレンスドキュメントをご覧ください。


/**
 * Publish a message for a Pub/Sub topic with compression enabled.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $message  The message to publish.
 */
function publisher_with_compression(
    string $projectId,
    string $topicName,
    string $message
): void {
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);

    // Enable compression and configure the compression threshold to
    // 10 bytes (default to 240 B). Publish requests of sizes > 10 B
    // (excluding the request headers) will get compressed.
    $topic = $pubsub->topic(
        $topicName,
        [
            'enableCompression' => true,
            'compressionBytesThreshold' => 10
        ]
    );
    $result = $topic->publish((new MessageBuilder)->setData($message)->build());

    printf(
        'Published a compressed message of message ID: %s' . PHP_EOL,
        $result['messageIds'][0]
    );
}

Ruby

このサンプルを試す前に、Pub/Sub クイックスタート: クライアントライブラリの使用にある Ruby 向けの手順に従って設定を行ってください。詳細については、Pub/Sub Ruby API のリファレンスドキュメントをご覧ください。

require "google/cloud/pubsub"

##
# Shows how to create a BigQuery subscription where messages published
# to a topic populates a BigQuery table.
#
# @param project_id [String]
# Your Google Cloud project (e.g. "my-project")
# @param topic_id [String]
# Your topic name (e.g. "my-secret")
#
def pubsub_publisher_with_compression project_id:, topic_id:
  pubsub = Google::Cloud::Pubsub.new project_id: project_id

  # Enable compression and configure the compression threshold to 10 bytes (default to 240 B).
  # Publish requests of sizes > 10 B (excluding the request headers) will get compressed.
  topic = pubsub.topic topic_id, async: {
    compress: true,
    compression_bytes_threshold: 10
  }

  begin
    topic.publish_async "This is a test message." do |result|
      raise "Failed to publish the message." unless result.succeeded?
      puts "Published a compressed message of message ID: #{result.message_id}"
    end

    # Stop the async_publisher to send all queued messages immediately.
    topic.async_publisher.stop.wait!
  rescue StandardError => e
    puts "Received error while publishing: #{e.message}"
  end
end

次のステップ

他の Google Cloud プロダクトに関連するコードサンプルの検索およびフィルタ検索を行うには、Google Cloud のサンプルをご覧ください。

圧縮レートでメッセージをパブリッシュ

もっと見る

コードサンプル

C++

C#

Go

Java

PHP

Ruby

次のステップ