‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

פרסום הודעות בנושאים

במסמך הזה מוסבר איך לפרסם הודעות.

אפליקציית בעל התוכן הדיגיטלי יוצרת ושולחת הודעות לנושא. ‫Pub/Sub מציע מסירת הודעות לפחות פעם אחת וסידור לפי מיטב היכולת למנויים קיימים.

התהליך הכללי של בקשה להצטרף לתוכנית השותפים של בעלי אתרים הוא:

יוצרים הודעה שמכילה את הנתונים.
שליחת בקשה לשרת Pub/Sub כדי לפרסם את ההודעה בנושא שצוין.

לפני שמתחילים

לפני שמגדירים את תהליך העבודה של הפרסום, צריך לבצע את המשימות הבאות:

התפקידים הנדרשים

כדי לקבל את ההרשאות שנדרשות לפרסום הודעות בנושא, צריך לבקש מהאדמין להקצות לכם ב-IAM את התפקיד פרסום הודעות ב-Pub/Sub (roles/pubsub.publisher) בנושא. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

יכול להיות שאפשר לקבל את ההרשאות הנדרשות גם באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש.

כדי ליצור או לעדכן נושאים ומינויים, צריך הרשאות נוספות.

פורמט הודעה

הודעה מורכבת משדות עם נתוני ההודעה ומטא-נתונים. ציינו בהודעה לפחות אחד מהפרטים הבאים:

נתוני ההודעה
מפתח הזמנה
מאפיינים עם מטא-נתונים נוספים

שירות Pub/Sub מוסיף להודעה את השדות הבאים:

מזהה הודעה ייחודי לנושא
חותמת זמן שמציינת מתי שירות Pub/Sub קיבל את ההודעה

מידע נוסף על הודעות זמין במאמר פורמט של הודעות.

פרסום הודעות

אפשר לפרסם הודעות באמצעות מסוף Google Cloud , Google Cloud CLI, ‏Pub/Sub API וספריות הלקוח. ספריות הלקוח יכולות לפרסם הודעות באופן אסינכרוני.

בדוגמאות הבאות מוסבר איך לפרסם הודעה בנושא.

המסוף

כדי לפרסם הודעה:

נכנסים לדף Pub/Sub topics במסוף Google Cloud .

כניסה לדף Pub/Sub topics
לוחצים על מזהה הנושא.
בדף Topic details, בקטע Messages לוחצים על Publish message.
בשדה גוף ההודעה, מזינים את נתוני ההודעה.
לוחצים על פרסום.

gcloud

כדי לפרסם הודעה, משתמשים בפקודה gcloud pubsub topics publish:

gcloud pubsub topics publish TOPIC_ID \
  --message=MESSAGE_DATA \
  [--attribute=KEY="VALUE",...]

מחליפים את מה שכתוב בשדות הבאים:

‫TOPIC_ID: מזהה הנושא
‫MESSAGE_DATA: מחרוזת עם נתוני ההודעה
‫KEY: המפתח של מאפיין ההודעה
‫VALUE: הערך של המפתח של מאפיין ההודעה

REST

כדי לפרסם הודעה, שולחים בקשת POST כמו זו:

POST  https://pubsub.googleapis.com/v1/projects/PROJECT_ID/topics/TOPIC_ID:publish
Content-Type: application/json
Authorization: Bearer $(gcloud auth application-default print-access-token)

מחליפים את מה שכתוב בשדות הבאים:

‫PROJECT_ID: מזהה הפרויקט עם הנושא
‫TOPIC_ID: מזהה הנושא

מציינים את השדות הבאים בגוף הבקשה:

{
  "messages": [
    {
      "attributes": {
        "KEY": "VALUE",
        ...
      },
      "data": "MESSAGE_DATA",
    }
  ]
}

מחליפים את מה שכתוב בשדות הבאים:

‫KEY: המפתח של מאפיין ההודעה
‫VALUE: הערך של המפתח של מאפיין ההודעה
‫MESSAGE_DATA: מחרוזת בקידוד base64 עם נתוני ההודעה

ההודעה חייבת להכיל שדה נתונים לא ריק או לפחות מאפיין אחד.

אם הבקשה מצליחה, התגובה היא אובייקט JSON עם מזהה ההודעה. הדוגמה הבאה היא תשובה עם מזהה הודעה:

{
  "messageIds": [
    "19916711285",
  ]
}

C++‎

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C++‎ במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף זמין במאמרי העזרה של Pub/Sub C++ API.

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::StatusOr;
[](pubsub::Publisher publisher) {
  auto message_id = publisher.Publish(
      pubsub::MessageBuilder{}.SetData("Hello World!").Build());
  auto done = message_id.then([](future<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#‎

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C# ‎ במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub C# API.


using Google.Cloud.PubSub.V1;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public class PublishMessagesAsyncSample
{
    public async Task<int> PublishMessagesAsync(string projectId, string topicId, IEnumerable<string> messageTexts)
    {
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        PublisherClient publisher = await PublisherClient.CreateAsync(topicName);

        int publishedMessageCount = 0;
        var publishTasks = messageTexts.Select(async text =>
        {
            try
            {
                string message = await publisher.PublishAsync(text);
                Console.WriteLine($"Published message {message}");
                Interlocked.Increment(ref publishedMessageCount);
            }
            catch (Exception exception)
            {
                Console.WriteLine($"An error occurred when publishing message {text}: {exception.Message}");
            }
        });
        await Task.WhenAll(publishTasks);
        // PublisherClient instance should be shutdown after use.
        // The TimeSpan specifies for how long to attempt to publish locally queued messages.
        await publisher.ShutdownAsync(TimeSpan.FromSeconds(15));
        return publishedMessageCount;
    }
}

המשך

בדוגמה הבאה נעשה שימוש בגרסה הראשית של ספריית הלקוח של Pub/Sub ב-Go‏ (v2). אם אתם עדיין משתמשים בספרייה של גרסה 1, כדאי לעיין במדריך להעברת נתונים לגרסה 2. כאן תוכלו למצוא רשימה של דוגמאות קוד שהוצאו משימוש בגרסה 1.

לפני שתנסו את הדוגמה הזו, פעלו לפי הוראות ההגדרה של Go במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במסמכי העזר של ה-API של Pub/Sub Go.

import (
	"context"
	"fmt"
	"io"
	"strconv"
	"sync"
	"sync/atomic"

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

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

	var wg sync.WaitGroup
	var totalErrors uint64

	// client.Publisher can be passed a topic ID (e.g. "my-topic") or
	// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
	// If a topic ID is provided, the project ID from the client is used.
	// Reuse this publisher for all publish calls to send messages in batches.
	publisher := client.Publisher(topicID)

	for i := 0; i < n; i++ {
		result := publisher.Publish(ctx, &pubsub.Message{
			Data: []byte("Message " + strconv.Itoa(i)),
		})

		wg.Add(1)
		go func(i int, res *pubsub.PublishResult) {
			defer wg.Done()
			// The Get method blocks until a server-generated ID or
			// an error is returned for the published message.
			id, err := res.Get(ctx)
			if err != nil {
				// Error handling code can be added here.
				fmt.Fprintf(w, "Failed to publish: %v", err)
				atomic.AddUint64(&totalErrors, 1)
				return
			}
			fmt.Fprintf(w, "Published message %d; msg ID: %v\n", i, id)
		}(i, result)
	}

	wg.Wait()

	if totalErrors > 0 {
		return fmt.Errorf("%d of %d messages did not publish successfully", totalErrors, n)
	}
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Java API.


import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.rpc.ApiException;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class PublishWithErrorHandlerExample {

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

    publishWithErrorHandlerExample(projectId, topicId);
  }

  public static void publishWithErrorHandlerExample(String projectId, String topicId)
      throws IOException, InterruptedException {
    TopicName topicName = TopicName.of(projectId, topicId);
    Publisher publisher = null;

    try {
      // Create a publisher instance with default settings bound to the topic
      publisher = Publisher.newBuilder(topicName).build();

      List<String> messages = Arrays.asList("first message", "second message");

      for (final String message : messages) {
        ByteString data = ByteString.copyFromUtf8(message);
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

        // Once published, returns a server-assigned message id (unique within the topic)
        ApiFuture<String> future = publisher.publish(pubsubMessage);

        // Add an asynchronous callback to handle success / failure
        ApiFutures.addCallback(
            future,
            new ApiFutureCallback<String>() {

              @Override
              public void onFailure(Throwable throwable) {
                if (throwable instanceof ApiException) {
                  ApiException apiException = ((ApiException) throwable);
                  // details on the API exception
                  System.out.println(apiException.getStatusCode().getCode());
                  System.out.println(apiException.isRetryable());
                }
                System.out.println("Error publishing message : " + message);
              }

              @Override
              public void onSuccess(String messageId) {
                // Once published, returns server-assigned message ids (unique within the topic)
                System.out.println("Published message ID: " + messageId);
              }
            },
            MoreExecutors.directExecutor());
      }
    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }
}

Node.js

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Node.js API.

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const data = JSON.stringify({foo: 'bar'});

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function publishMessage(topicNameOrId, data) {
  // Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
  const dataBuffer = Buffer.from(data);

  // Cache topic objects (publishers) and reuse them.
  const topic = pubSubClient.topic(topicNameOrId);

  try {
    const messageId = await topic.publishMessage({data: dataBuffer});
    console.log(`Message ${messageId} published.`);
  } catch (error) {
    console.error(`Received error while publishing: ${error.message}`);
    process.exitCode = 1;
  }
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const data = JSON.stringify({foo: 'bar'});

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function publishMessage(topicNameOrId: string, data: string) {
  // Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
  const dataBuffer = Buffer.from(data);

  // Cache topic objects (publishers) and reuse them.
  const topic = pubSubClient.topic(topicNameOrId);

  try {
    const messageId = await topic.publishMessage({data: dataBuffer});
    console.log(`Message ${messageId} published.`);
  } catch (error) {
    console.error(
      `Received error while publishing: ${(error as Error).message}`,
    );
    process.exitCode = 1;
  }
}

PHP

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של PHP במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub PHP API.

use Google\Cloud\PubSub\MessageBuilder;
use Google\Cloud\PubSub\PubSubClient;

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

    $topic = $pubsub->topic($topicName);
    $topic->publish((new MessageBuilder)->setData($message)->build());

    print('Message published' . PHP_EOL);
}

Python

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Python במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של ה-API בשפת Python של Pub/Sub.

"""Publishes multiple messages to a Pub/Sub topic with an error handler."""
from concurrent import futures
from google.cloud import pubsub_v1
from typing import Callable

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)
publish_futures = []

def get_callback(
    publish_future: pubsub_v1.publisher.futures.Future, data: str
) -> Callable[[pubsub_v1.publisher.futures.Future], None]:
    def callback(publish_future: pubsub_v1.publisher.futures.Future) -> None:
        try:
            # Wait 60 seconds for the publish call to succeed.
            print(publish_future.result(timeout=60))
        except futures.TimeoutError:
            print(f"Publishing {data} timed out.")

    return callback

for i in range(10):
    data = str(i)
    # When you publish a message, the client returns a future.
    publish_future = publisher.publish(topic_path, data.encode("utf-8"))
    # Non-blocking. Publish failures are handled in the callback function.
    publish_future.add_done_callback(get_callback(publish_future, data))
    publish_futures.append(publish_future)

# Wait for all the publish futures to resolve before exiting.
futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)

print(f"Published messages with error handler to {topic_path}.")

Ruby

בדוגמה הבאה נעשה שימוש בגרסה 3 של ספריית הלקוח של Pub/Sub ב-Ruby. אם אתם עדיין משתמשים בספרייה v2, עיינו במדריך להעברת נתונים (מיגרציה) ל-v3. לרשימה של דוגמאות קוד ב-Ruby v2, עיינו ב דוגמאות הקוד שהוצאו משימוש.

לפני שתנסו את הדוגמה הזו, פעלו לפי הוראות ההגדרה של Ruby במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Ruby API.

# topic_id = "your-topic-id"

pubsub = Google::Cloud::PubSub.new
publisher = pubsub.publisher topic_id

begin
  publisher.publish_async "This is a test message." do |result|
    raise "Failed to publish the message." unless result.succeeded?
    puts "Message published asynchronously."
  end
  # Stop the async_publisher to send all queued messages immediately.
  publisher.async_publisher.stop.wait!
rescue StandardError => e
  puts "Received error while publishing: #{e.message}"
end

Rust

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Rust במדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Rust API.

use futures::future::join_all;
use google_cloud_pubsub::client::Publisher;
use google_cloud_pubsub::model::Message;

pub async fn sample(project_id: &str, topic_id: &str) -> anyhow::Result<()> {
    let topic_name = format!("projects/{project_id}/topics/{topic_id}");
    let publisher = Publisher::builder(topic_name).build().await?;

    let results = join_all(
        (0..10).map(|i| publisher.publish(Message::new().set_data(format!("message {i}")))),
    )
    .await;

    for (i, result) in results.into_iter().enumerate() {
        match result {
            Ok(message_id) => println!("published message {i}; message ID: {message_id}"),
            // Handle the error, e.g., log it, retry, or move the message to a queue.
            Err(e) => eprintln!("failed to publish message {i}: {e}"),
        }
    }

    Ok(())
}

אחרי שמפרסמים הודעה, שירות Pub/Sub מחזיר את מזהה ההודעה למפרסם.

שימוש במאפיינים לפרסום הודעה

אתם יכולים להטמיע מאפיינים מותאמים אישית כמטא-נתונים בהודעות Pub/Sub. מאפיינים משמשים כדי לספק מידע נוסף על ההודעה, כמו העדיפות, המקור או היעד שלה. אפשר גם להשתמש במאפיינים כדי לסנן הודעות במינוי.

כדי להשתמש במאפיינים בהודעות, צריך לפעול לפי ההנחיות הבאות:

לכל הודעה יכולים להיות עד 100 מאפיינים.
המפתחות והערכים של המאפיינים חייבים להיות מסוג מחרוזת. אין קידוד נדרש.
מפתחות המאפיינים לא יכולים להתחיל ב-goog ולא יכולים להיות גדולים מ-256 בייטים.
ערכי המאפיינים לא יכולים להיות גדולים מ-1,024 בייטים.

אפשר לייצג את סכימת ההודעה באופן הבא:

{
  "data": string,
  "attributes": {
    string: string,
    ...
  },
  "messageId": string,
  "publishTime": string,
  "orderingKey": string
}

בכפילויות בצד הפרסום, יכול להיות שיופיעו ערכי publishTime שונים לאותה הודעה מקורית בצד הלקוח, גם עם אותו messageId.

סכימת ה-JSON של PubsubMessage מתפרסמת כחלק מהמסמכים של REST ו-RPC. אפשר להשתמש במאפיינים מותאמים אישית לחותמות זמן של אירועים.

בדוגמאות הבאות מוסבר איך לפרסם הודעה עם מאפיינים בנושא.

המסוף

כדי לפרסם הודעה עם מאפיינים:

נכנסים לדף Topics במסוף Google Cloud .

כניסה לדף Pub/Sub topics
לוחצים על הנושא שבו רוצים לפרסם הודעות.
בדף הפרטים של הנושא, לוחצים על הודעות.
לוחצים על פרסום ההודעה.
בשדה גוף ההודעה, מזינים את נתוני ההודעה.
בקטע Message attributes, לוחצים על Add an attribute.
מזינים צמד מפתח/ערך.
מוסיפים עוד מאפיינים, אם צריך.
לוחצים על פרסום.

gcloud

gcloud pubsub topics publish my-topic --message="hello" \
  --attribute="origin=gcloud-sample,username=gcp,eventTime='2021-01-01T12:00:00Z'"

C++‎

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::StatusOr;
[](pubsub::Publisher publisher) {
  std::vector<future<void>> done;
  for (int i = 0; i != 10; ++i) {
    auto message_id = publisher.Publish(
        pubsub::MessageBuilder{}
            .SetData("Hello World! [" + std::to_string(i) + "]")
            .SetAttribute("origin", "cpp-sample")
            .SetAttribute("username", "gcp")
            .Build());
    done.push_back(message_id.then([i](future<StatusOr<std::string>> f) {
      auto id = f.get();
      if (!id) throw std::move(id).status();
      std::cout << "Message " << i << " published with id=" << *id << "\n";
    }));
  }
  publisher.Flush();
  // Block until all the messages are published (optional)
  for (auto& f : done) f.get();
}

C#‎


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

public class PublishMessageWithCustomAttributesAsyncSample
{
    public async Task PublishMessageWithCustomAttributesAsync(string projectId, string topicId, string messageText)
    {
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        PublisherClient publisher = await PublisherClient.CreateAsync(topicName);

        var pubsubMessage = new PubsubMessage
        {
            // The data is any arbitrary ByteString. Here, we're using text.
            Data = ByteString.CopyFromUtf8(messageText),
            // The attributes provide metadata in a string-to-string dictionary.
            Attributes =
            {
                { "year", "2020" },
                { "author", "unknown" }
            }
        };
        string message = await publisher.PublishAsync(pubsubMessage);
        Console.WriteLine($"Published message {message}");
        // PublisherClient instance should be shutdown after use.
        // The TimeSpan specifies for how long to attempt to publish locally queued messages.
        await publisher.ShutdownAsync(TimeSpan.FromSeconds(15));
    }
}

המשך

import (
	"context"
	"fmt"
	"io"

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

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

	// client.Publisher can be passed a topic ID (e.g. "my-topic") or
	// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
	// If a topic ID is provided, the project ID from the client is used.
	// Reuse this publisher for all publish calls to send messages in batches.
	publisher := client.Publisher(topicID)
	result := publisher.Publish(ctx, &pubsub.Message{
		Data: []byte("Hello world!"),
		Attributes: map[string]string{
			"origin":   "golang",
			"username": "gcp",
		},
	})
	// 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("Get: %w", err)
	}
	fmt.Fprintf(w, "Published message with custom attributes; msg ID: %v\n", id)
	return nil
}

Java


import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.common.collect.ImmutableMap;
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 PublishWithCustomAttributesExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";

    publishWithCustomAttributesExample(projectId, topicId);
  }

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

    try {
      // Create a publisher instance with default settings bound to the topic
      publisher = Publisher.newBuilder(topicName).build();

      String message = "first message";
      ByteString data = ByteString.copyFromUtf8(message);
      PubsubMessage pubsubMessage =
          PubsubMessage.newBuilder()
              .setData(data)
              .putAllAttributes(ImmutableMap.of("year", "2020", "author", "unknown"))
              .build();

      // Once published, returns a server-assigned message id (unique within the topic)
      ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
      String messageId = messageIdFuture.get();
      System.out.println("Published a message with custom attributes: " + messageId);

    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const data = JSON.stringify({foo: 'bar'});

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function publishMessageWithCustomAttributes(topicNameOrId, data) {
  // Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
  const dataBuffer = Buffer.from(data);

  // Add two custom attributes, origin and username, to the message
  const customAttributes = {
    origin: 'nodejs-sample',
    username: 'gcp',
  };

  // Cache topic objects (publishers) and reuse them.
  const topic = pubSubClient.topic(topicNameOrId);

  const messageId = await topic.publishMessage({
    data: dataBuffer,
    attributes: customAttributes,
  });
  console.log(`Message ${messageId} published.`);
}

Python

from google.cloud import pubsub_v1

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

for n in range(1, 10):
    data_str = f"Message number {n}"
    # Data must be a bytestring
    data = data_str.encode("utf-8")
    # Add two attributes, origin and username, to the message
    future = publisher.publish(
        topic_path, data, origin="python-sample", username="gcp"
    )
    print(future.result())

print(f"Published messages with custom attributes to {topic_path}.")

Ruby

# topic_id = "your-topic-id"

pubsub = Google::Cloud::PubSub.new
publisher = pubsub.publisher topic_id

# Add two attributes, origin and username, to the message
publisher.publish_async "This is a test message.",
                        origin:   "ruby-sample",
                        username: "gcp" do |result|
  raise "Failed to publish the message." unless result.succeeded?
  puts "Message with custom attributes published asynchronously."
end

# Stop the async_publisher to send all queued messages immediately.
publisher.async_publisher.stop.wait!

Rust

use google_cloud_pubsub::client::Publisher;
use google_cloud_pubsub::model::Message;

pub async fn sample(project_id: &str, topic_id: &str) -> anyhow::Result<()> {
    let topic_name = format!("projects/{project_id}/topics/{topic_id}");
    let publisher = Publisher::builder(topic_name).build().await?;

    let message = Message::new()
        .set_data("Hello, World!")
        .set_attributes([("origin", "rust-sample"), ("username", "gcp")]);
    let message_id = publisher.publish(message).await?;

    println!("published message with custom attributes; message ID: {message_id}");
    Ok(())
}

שימוש במפתחות סדר לפרסום הודעה

כדי לקבל הודעות לפי הסדר בלקוחות המנויים, צריך להגדיר את לקוחות בעלי התוכן הדיגיטלי כך שיפרסמו הודעות עם מפתחות סדר.

כדי להבין את הקונספט של מפתחות סדר, קראו את המאמר הודעות סדר.

ריכזנו כאן רשימה של נקודות מרכזיות שחשוב לדעת על העברת הודעות לפי סדר בלקוחות של בעלי תוכן דיגיטלי:

הזמנה בלקוח בעל בעל תוכן דיגיטלי יחיד: כשלקוח בעל תוכן דיגיטלי יחיד מפרסם הודעות עם אותו מפתח הזמנה באותו אזור, לקוח המנוי מקבל את ההודעות האלה בדיוק בסדר שבו הן פורסמו. לדוגמה, אם לקוח מפרסם מפרסם את ההודעות 1, 2 ו-3 עם מפתח הסדר A, לקוח המנוי מקבל אותן בסדר 1, 2, 3.
הזמנה בין כמה לקוחות של בעלי תוכן דיגיטלי: סדר ההודעות שמתקבלות על ידי לקוחות מנויים תואם לסדר שבו הן פורסמו באותו אזור, גם כשכמה לקוחות של בעלי תוכן דיגיטלי משתמשים באותו מפתח הזמנה. עם זאת, ללקוחות של בעל התוכן הדיגיטלי אין מידע על ההזמנה הזו.

לדוגמה, אם לקוחות X ו-Y מפרסמים הודעות עם מפתח הסדר A, וההודעה של X מתקבלת ב-Pub/Sub לפני ההודעה של Y, אז כל הלקוחות הרשומים מקבלים את ההודעה של X לפני ההודעה של Y. אם נדרש סדר הודעות קפדני בין לקוחות שונים של בעלי תוכן דיגיטלי, הלקוחות האלה צריכים להטמיע מנגנון תיאום נוסף כדי להבטיח שהם לא יפרסמו בו-זמנית הודעות עם אותו מפתח סדר. לדוגמה, אפשר להשתמש בשירות נעילה כדי לשמור על הבעלות על מפתח הזמנה בזמן הפרסום.
הזמנה בין אזורים: ההתחייבות למסירה לפי סדר חלה רק כשפעולות הפרסום של מפתח הזמנה נמצאות באותו אזור. אם אפליקציית הפרסום מפרסמת הודעות עם אותו מפתח סדר לאזורים שונים, אי אפשר לאכוף את הסדר בין הפרסומים האלה. המנויים יכולים להתחבר לכל אזור, וההתחייבות לסדר עדיין נשמרת.

כשמריצים את האפליקציה ב- Google Cloud, כברירת מחדל היא מתחברת לנקודת הקצה של Pub/Sub באותו אזור. לכן, אם תפעילו את האפליקציה באזור אחד ב-Google Cloud , בדרך כלל תהיה אינטראקציה עם אזור אחד.

כשמפעילים את אפליקציית הפרסום מחוץ ל-Google Cloud או בכמה אזורים, אפשר להבטיח שהיא תתחבר לאזור אחד באמצעות נקודת קצה (endpoint) לפי מיקום כשמגדירים את לקוח Pub/Sub. כל נקודות הקצה של המיקומים ב-Pub/Sub מפנות לאזורים בודדים. למידע נוסף על נקודות קצה (endpoints) לפי מיקום, קראו את המאמר נקודות קצה (endpoints) ב-Pub/Sub. רשימה של כל נקודות הקצה למיקום ב-Pub/Sub
כשלים בפרסום: אם הפרסום עם מפתח סדר נכשל, ההודעות בתור של אותו מפתח סדר בבעל התוכן הדיגיטלי נכשלות, כולל בקשות פרסום עתידיות של מפתח הסדר הזה. במקרים כאלה, צריך להמשיך את הפרסום עם מפתחות סדר. במאמר ניסיון חוזר של בקשות עם מפתחות הזמנה תוכלו לקרוא דוגמה לחידוש פעולת הפרסום.

אפשר לפרסם הודעות עם מפתחות סידור באמצעות מסוף Google Cloud , Google Cloud CLI, ‏Pub/Sub API או ספריות הלקוח.

המסוף

כדי לפרסם הודעה עם מאפיינים:

נכנסים לדף Topics במסוף Google Cloud .

כניסה לדף Pub/Sub topics
לוחצים על הנושא שבו רוצים לפרסם הודעות.
בדף הפרטים של הנושא, לוחצים על הודעות.
לוחצים על פרסום ההודעה.
בשדה גוף ההודעה, מזינים את נתוני ההודעה.
בשדה Message ordering, מזינים מפתח סדר.
לוחצים על פרסום.

gcloud

כדי לפרסם הודעה עם מפתח סדר, משתמשים בפקודה gcloud pubsub topics publish ובדגל --ordering-key:

gcloud pubsub topics publish TOPIC_ID \
  --message=MESSAGE_DATA \
  --ordering-key=ORDERING_KEY

מחליפים את מה שכתוב בשדות הבאים:

‫TOPIC_ID: מזהה הנושא
‫MESSAGE_DATA: מחרוזת עם נתוני ההודעה
‫ORDERING_KEY: מחרוזת עם מפתח הזמנה

REST

כדי לפרסם הודעה עם מפתח סדר, שולחים בקשת POST כמו זו:

POST  https://pubsub.googleapis.com/v1/projects/PROJECT_ID/topics/TOPIC_ID:publish
Content-Type: application/json
Authorization: Bearer $(gcloud auth application-default print-access-token)

מחליפים את מה שכתוב בשדות הבאים:

‫PROJECT_ID: מזהה הפרויקט עם הנושא
‫TOPIC_ID: מזהה הנושא

מציינים את השדות הבאים בגוף הבקשה:

{
  "messages": [
    {
      "attributes": {
        "KEY": "VALUE",
        ...
      },
      "data": "MESSAGE_DATA",
      "ordering_key": "ORDERING_KEY",
    }
  ]
}

מחליפים את מה שכתוב בשדות הבאים:

‫KEY: המפתח של מאפיין ההודעה
‫VALUE: הערך של המפתח של מאפיין ההודעה
‫MESSAGE_DATA: מחרוזת בקידוד base64 עם נתוני ההודעה
‫ORDERING_KEY: מחרוזת עם מפתח הזמנה

ההודעה חייבת להכיל שדה נתונים לא ריק או לפחות מאפיין אחד.

אם הבקשה מצליחה, התגובה היא אובייקט JSON עם מזהה ההודעה. הדוגמה הבאה היא תשובה עם מזהה הודעה:

{
  "messageIds": [
    "19916711285",
  ]
}

C++‎

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::StatusOr;
[](pubsub::Publisher publisher) {
  struct SampleData {
    std::string ordering_key;
    std::string data;
  } data[] = {
      {"key1", "message1"}, {"key2", "message2"}, {"key1", "message3"},
      {"key1", "message4"}, {"key1", "message5"},
  };
  std::vector<future<void>> done;
  for (auto& datum : data) {
    auto message_id =
        publisher.Publish(pubsub::MessageBuilder{}
                              .SetData("Hello World! [" + datum.data + "]")
                              .SetOrderingKey(datum.ordering_key)
                              .Build());
    std::string ack_id = datum.ordering_key + "#" + datum.data;
    done.push_back(message_id.then([ack_id](future<StatusOr<std::string>> f) {
      auto id = f.get();
      if (!id) throw std::move(id).status();
      std::cout << "Message " << ack_id << " published with id=" << *id
                << "\n";
    }));
  }
  publisher.Flush();
  // Block until all the messages are published (optional)
  for (auto& f : done) f.get();
}

C#‎


using Google.Cloud.PubSub.V1;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public class PublishOrderedMessagesAsyncSample
{
    public async Task<int> PublishOrderedMessagesAsync(string projectId, string topicId, IEnumerable<(string, string)> keysAndMessages)
    {
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);

        var customSettings = new PublisherClient.Settings
        {
            EnableMessageOrdering = true
        };

        PublisherClient publisher = await new PublisherClientBuilder
        {
            TopicName = topicName,
            // Sending messages to the same region ensures they are received in order even when multiple publishers are used.
            Endpoint = "us-east1-pubsub.googleapis.com:443",
            Settings = customSettings
        }.BuildAsync();

        int publishedMessageCount = 0;
        var publishTasks = keysAndMessages.Select(async keyAndMessage =>
        {
            try
            {
                string message = await publisher.PublishAsync(keyAndMessage.Item1, keyAndMessage.Item2);
                Console.WriteLine($"Published message {message}");
                Interlocked.Increment(ref publishedMessageCount);
            }
            catch (Exception exception)
            {
                Console.WriteLine($"An error occurred when publishing message {keyAndMessage.Item2}: {exception.Message}");
            }
        });
        await Task.WhenAll(publishTasks);
        // PublisherClient instance should be shutdown after use.
        // The TimeSpan specifies for how long to attempt to publish locally queued messages.
        await publisher.ShutdownAsync(TimeSpan.FromSeconds(15));
        return publishedMessageCount;
    }
}

המשך

import (
	"context"
	"fmt"
	"io"
	"sync"
	"sync/atomic"

	"cloud.google.com/go/pubsub/v2"
	"google.golang.org/api/option"
)

func publishWithOrderingKey(w io.Writer, projectID, topicID string) {
	// projectID := "my-project-id"
	// topicID := "my-topic"
	ctx := context.Background()

	// Pub/Sub's ordered delivery guarantee only applies when publishes for an ordering key are in the same region.
	// For list of locational endpoints for Pub/Sub, see https://cloud.google.com/pubsub/docs/reference/service_apis_overview#list_of_locational_endpoints
	client, err := pubsub.NewClient(ctx, projectID,
		option.WithEndpoint("us-east1-pubsub.googleapis.com:443"))
	if err != nil {
		fmt.Fprintf(w, "pubsub.NewClient: %v", err)
		return
	}
	defer client.Close()

	var wg sync.WaitGroup
	var totalErrors uint64

	// client.Publisher can be passed a topic ID (e.g. "my-topic") or
	// a fully qualified name (e.g. "projects/my-project/topics/my-topic").
	// If a topic ID is provided, the project ID from the client is used.
	// Reuse this publisher for all publish calls to send messages in batches.
	publisher := client.Publisher(topicID)
	publisher.EnableMessageOrdering = true

	messages := []struct {
		message     string
		orderingKey string
	}{
		{
			message:     "message1",
			orderingKey: "key1",
		},
		{
			message:     "message2",
			orderingKey: "key2",
		},
		{
			message:     "message3",
			orderingKey: "key1",
		},
		{
			message:     "message4",
			orderingKey: "key2",
		},
	}

	for _, m := range messages {
		result := publisher.Publish(ctx, &pubsub.Message{
			Data:        []byte(m.message),
			OrderingKey: m.orderingKey,
		})

		wg.Add(1)
		go func(res *pubsub.PublishResult) {
			defer wg.Done()
			// The Get method blocks until a server-generated ID or
			// an error is returned for the published message.
			_, err := res.Get(ctx)
			if err != nil {
				// Error handling code can be added here.
				fmt.Printf("Failed to publish: %s\n", err)
				atomic.AddUint64(&totalErrors, 1)
				return
			}
		}(result)
	}

	wg.Wait()

	if totalErrors > 0 {
		fmt.Fprintf(w, "%d of 4 messages did not publish successfully", totalErrors)
		return
	}

	fmt.Fprint(w, "Published 4 messages with ordering keys successfully\n")
}

Java

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.rpc.ApiException;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class PublishWithOrderingKeys {
  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";

    publishWithOrderingKeysExample(projectId, topicId);
  }

  public static void publishWithOrderingKeysExample(String projectId, String topicId)
      throws IOException, InterruptedException {
    TopicName topicName = TopicName.of(projectId, topicId);
    // Create a publisher and set message ordering to true.
    Publisher publisher =
        Publisher.newBuilder(topicName)
            // Sending messages to the same region ensures they are received in order
            // even when multiple publishers are used.
            .setEndpoint("us-east1-pubsub.googleapis.com:443")
            .setEnableMessageOrdering(true)
            .build();

    try {
      Map<String, String> messages = new LinkedHashMap<String, String>();
      messages.put("message1", "key1");
      messages.put("message2", "key2");
      messages.put("message3", "key1");
      messages.put("message4", "key2");

      for (Map.Entry<String, String> entry : messages.entrySet()) {
        ByteString data = ByteString.copyFromUtf8(entry.getKey());
        PubsubMessage pubsubMessage =
            PubsubMessage.newBuilder().setData(data).setOrderingKey(entry.getValue()).build();
        ApiFuture<String> future = publisher.publish(pubsubMessage);

        // Add an asynchronous callback to handle publish success / failure.
        ApiFutures.addCallback(
            future,
            new ApiFutureCallback<String>() {

              @Override
              public void onFailure(Throwable throwable) {
                if (throwable instanceof ApiException) {
                  ApiException apiException = ((ApiException) throwable);
                  // Details on the API exception.
                  System.out.println(apiException.getStatusCode().getCode());
                  System.out.println(apiException.isRetryable());
                }
                System.out.println("Error publishing message : " + pubsubMessage.getData());
              }

              @Override
              public void onSuccess(String messageId) {
                // Once published, returns server-assigned message ids (unique within the topic).
                System.out.println(pubsubMessage.getData() + " : " + messageId);
              }
            },
            MoreExecutors.directExecutor());
      }
    } finally {
      // When finished with the publisher, shutdown to free up resources.
      publisher.shutdown();
      publisher.awaitTermination(1, TimeUnit.MINUTES);
    }
  }
}

Node.js

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const data = JSON.stringify({foo: 'bar'});
// const orderingKey = 'key1';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub({
  // Sending messages to the same region ensures they are received in order
  // even when multiple publishers are used.
  apiEndpoint: 'us-east1-pubsub.googleapis.com:443',
});

async function publishOrderedMessage(topicNameOrId, data, orderingKey) {
  // Publishes the message as a string, e.g. "Hello, world!" or JSON.stringify(someObject)
  const dataBuffer = Buffer.from(data);

  // Be sure to set an ordering key that matches other messages
  // you want to receive in order, relative to each other.
  const message = {
    data: dataBuffer,
    orderingKey: orderingKey,
  };

  // Cache topic objects (publishers) and reuse them.
  //
  // Pub/Sub's ordered delivery guarantee only applies when publishes for an ordering
  // key are in the same region. For list of locational endpoints for Pub/Sub, see:
  // https://cloud.google.com/pubsub/docs/reference/service_apis_overview#list_of_locational_endpoints
  const publishOptions = {
    messageOrdering: true,
  };
  const topic = pubSubClient.topic(topicNameOrId, publishOptions);

  // Publishes the message
  const messageId = await topic.publishMessage(message);

  console.log(`Message ${messageId} published.`);

  return messageId;
}

Python

from google.cloud import pubsub_v1

# TODO(developer): Choose an existing topic.
# project_id = "your-project-id"
# topic_id = "your-topic-id"

publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True)
# Sending messages to the same region ensures they are received in order
# even when multiple publishers are used.
client_options = {"api_endpoint": "us-east1-pubsub.googleapis.com:443"}
publisher = pubsub_v1.PublisherClient(
    publisher_options=publisher_options, client_options=client_options
)
# The `topic_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/topics/{topic_id}`
topic_path = publisher.topic_path(project_id, topic_id)

for message in [
    ("message1", "key1"),
    ("message2", "key2"),
    ("message3", "key1"),
    ("message4", "key2"),
]:
    # Data must be a bytestring
    data = message[0].encode("utf-8")
    ordering_key = message[1]
    # When you publish a message, the client returns a future.
    future = publisher.publish(topic_path, data=data, ordering_key=ordering_key)
    print(future.result())

print(f"Published messages with ordering keys to {topic_path}.")

Ruby

# topic_id = "your-topic-id"

pubsub = Google::Cloud::PubSub.new
# Start sending messages in one request once the size of all queued messages
# reaches 1 MB or the number of queued messages reaches 20
publisher = pubsub.publisher topic_id, async: {
  max_bytes:    1_000_000,
  max_messages: 20
}

publisher.enable_message_ordering!
10.times do |i|
  publisher.publish_async "This is message ##{i}.",
                          ordering_key: "ordering-key"
end

# Stop the async_publisher to send all queued messages immediately.
publisher.async_publisher.stop!
puts "Messages published with ordering key."

Rust

use google_cloud_pubsub::client::Publisher;
use google_cloud_pubsub::model::Message;

pub async fn sample(project: &str, topic_id: &str) -> anyhow::Result<()> {
    let publisher = Publisher::builder(format!("projects/{project}/topics/{topic_id}"))
        // Pub/Sub's ordered delivery guarantee only applies when publishes for
        // an ordering key are in the same region.
        .with_endpoint("https://us-east1-pubsub.googleapis.com")
        .build()
        .await?;

    let publish_futures = [
        ("message1", "key1"),
        ("message2", "key2"),
        ("message3", "key1"),
        ("message4", "key2"),
    ]
    .map(|(data, ordering_key)| {
        publisher.publish(
            Message::new()
                .set_data(data.as_bytes())
                .set_ordering_key(ordering_key),
        )
    });

    for publish_future in publish_futures {
        let message_id = publish_future
            .await
            .inspect_err(|e| eprintln!("error publishing message: {e:?}"))?;
        println!("published message with ID: {message_id}");
    }

    Ok(())
}

מעקב אחרי בעל אפליקציה

ב-Cloud Monitoring יש כמה מדדים למעקב אחרי נושאים.

כדי לעקוב אחרי נושא ולשמור על תקינות של בעל תוכן דיגיטלי, אפשר לעיין במאמר שמירה על תקינות של בעל תוכן דיגיטלי.

המאמרים הבאים

כדי להגביל את המיקומים שבהם Pub/Sub מאחסן נתוני הודעות, קראו את המאמר הגבלת המיקומים של משאבי Pub/Sub.
כדי לפרסם הודעות עם סכימה, קראו את הסקירה הכללית על סכימות.
כדי ללמוד איך להגדיר אפשרויות מסירה מתקדמות, אפשר לעיין במאמרים הבאים:

פרסום הודעות בנושאים קל לארגן דפים בעזרת אוספים אפשר לשמור ולסווג תוכן על סמך ההעדפות שלך.

לפני שמתחילים

התפקידים הנדרשים

פורמט הודעה

פרסום הודעות

המסוף

gcloud

REST

C++‎

C#‎

המשך

Java

Node.js

Node.js

PHP

Python

Ruby

Rust

שימוש במאפיינים לפרסום הודעה

המסוף

gcloud

C++‎

C#‎

המשך

Java

Node.js

Python

Ruby

Rust

שימוש במפתחות סדר לפרסום הודעה

המסוף

gcloud

REST

C++‎

C#‎

המשך

Java

Node.js

Python

Ruby

Rust

מעקב אחרי בעל אפליקציה

המאמרים הבאים

פרסום הודעות בנושאים