透過並行控制程序處理更多訊息

重點： 並行控制可讓您設定 Pub/Sub 用戶端程式庫用於提取訊息的執行緒或串流數量。執行緒越多，就能同時處理更多訊息。

並行控制是 Pub/Sub 高階用戶端程式庫的可用功能。使用低階程式庫時，您也可以實作自己的並行控制。

並行控制支援取決於用戶端程式庫的程式設計語言。對於支援平行執行緒的語言實作 (例如 C++、Go 和 Java)，用戶端程式庫會預設選擇執行緒數量。

這個選項可能不適合您的應用程式。舉例來說，如果訂閱端應用程式無法處理大量收到的訊息，且並未受 CPU 限制，就必須增加執行緒數量。如果是 CPU 密集型訊息處理作業，減少執行緒數量可能較為合適。

本頁說明並行控制的概念，以及如何為訂閱端用戶端設定這項功能。如要設定發布商用戶端以進行並行控制，請參閱「並行控制」。

並行控制設定

並行控制變數的預設值和變數名稱可能因用戶端程式庫而異。詳情請參閱「API 參考文件」。舉例來說，在 Java 用戶端程式庫中，設定並行控制的方法為 setParallelPullCount()、setExecutorProvider()、setSystemExecutorProvider() 和 setChannelProvider()。

• setParallelPullCount() 可讓您決定要開啟的串流數量。如果訂閱端用戶端可處理的資料量超過單一串流傳送的資料量 (10 MBps)，您可以開啟更多串流。

• setExecutorProvider() 可讓您自訂用於處理訊息的執行器供應商。舉例來說，您可以將執行器供應商變更為傳回單一共用執行器的供應商，並在多個訂閱端之間使用有限數量的執行緒。這項設定有助於限制建立的執行緒數量。用於並行控制的執行緒總數取決於用戶端程式庫中傳遞的執行器供應商，以及平行提取計數。

• setSystemExecutorProvider() 可讓您自訂用於租約管理的執行工具供應器。通常您不會設定這個值，除非想在 setExecutorProvider 和 setSystemExecutorProvider 中使用相同的執行器供應商。舉例來說，如果您有多個低輸送量的訂閱項目，可以使用同一個執行器提供者。使用相同值可限制用戶端中的執行緒數量。

• setChannelProvider() 可讓您自訂用於開啟 Pub/Sub 連線的管道供應商。通常您不會設定這個值，除非想在多個訂閱端之間使用相同管道。在過多用戶端重複使用管道，可能會導致 GOAWAY 或 ENHANCE_YOUR_CALM 錯誤。如果在應用程式記錄或 Cloud Logs 中看到這些錯誤，請建立更多管道。

並行控制的程式碼範例

C++

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 C++ 設定說明操作。詳情請參閱 Pub/Sub C++ API 參考文件。

namespace pubsub = ::google::cloud::pubsub;
using ::google::cloud::future;
using ::google::cloud::GrpcBackgroundThreadPoolSizeOption;
using ::google::cloud::Options;
using ::google::cloud::StatusOr;
auto sample = [](std::string project_id, std::string subscription_id) {
  // Create a subscriber with 16 threads handling I/O work, by default the
  // library creates `std::thread::hardware_concurrency()` threads.
  auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(
      pubsub::Subscription(std::move(project_id), std::move(subscription_id)),
      Options{}
          .set<pubsub::MaxConcurrencyOption>(8)
          .set<GrpcBackgroundThreadPoolSizeOption>(16)));

  // Create a subscription where up to 8 messages are handled concurrently. By
  // default the library uses `std::thread::hardware_concurrency()` as the
  // maximum number of concurrent callbacks.
  auto session = subscriber.Subscribe(
      [](pubsub::Message const& m, pubsub::AckHandler h) {
        // This handler executes in the I/O threads, applications could use,
        // std::async(), a thread-pool, or any other mechanism to transfer the
        // execution to other threads.
        std::cout << "Received message " << m << "\n";
        std::move(h).ack();
        PleaseIgnoreThisSimplifiesTestingTheSamples();
      });
  return std::make_pair(subscriber, std::move(session));
};

Go

下列範例使用 Go Pub/Sub 用戶端程式庫的主要版本 (v2)。如果您仍在使用第 1 版程式庫，請參閱第 2 版遷移指南。如要查看第 1 版程式碼範例清單，請參閱 已淘汰的程式碼範例。

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 Go 設定說明操作。詳情請參閱 Pub/Sub Go API 參考文件。

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

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

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

	// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
	// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
	// If a subscription ID is provided, the project ID from the client is used.
	sub := client.Subscriber(subID)
	// NumGoroutines determines the number of streams sub.Receive will spawn to pull
	// messages. It is recommended to set this to 1, unless your throughput
	// is greater than 10 MB/s, as even having 1 stream can still result in
	// messages being handled asynchronously.
	sub.ReceiveSettings.NumGoroutines = 1
	// MaxOutstandingMessages limits the number of concurrent handlers of messages.
	// In this case, up to 8 unacked messages can be handled concurrently.
	sub.ReceiveSettings.MaxOutstandingMessages = 8

	// Receive messages for 10 seconds, which simplifies testing.
	// Comment this out in production, since `Receive` should
	// be used as a long running operation.
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	var received int32

	// Receive blocks until the context is cancelled or an error occurs.
	err = sub.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
		atomic.AddInt32(&received, 1)
		msg.Ack()
	})
	if err != nil {
		return fmt.Errorf("sub.Receive returned error: %w", err)
	}
	fmt.Fprintf(w, "Received %d messages\n", received)

	return nil
}

Java

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 Java 設定說明操作。詳情請參閱 Pub/Sub Java API 參考文件。

import com.google.api.gax.core.ExecutorProvider;
import com.google.api.gax.core.InstantiatingExecutorProvider;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

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

    subscribeWithConcurrencyControlExample(projectId, subscriptionId);
  }

  public static void subscribeWithConcurrencyControlExample(
      String projectId, String subscriptionId) {
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          // Handle incoming message, then ack the received message.
          System.out.println("Id: " + message.getMessageId());
          System.out.println("Data: " + message.getData().toStringUtf8());
          consumer.ack();
        };

    Subscriber subscriber = null;
    try {
      // Provides an executor service for processing messages. The default `executorProvider` used
      // by the subscriber has a default thread count of 5.
      ExecutorProvider executorProvider =
          InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(4).build();

      // `setParallelPullCount` determines how many StreamingPull streams the subscriber will open
      // to receive message. It defaults to 1. `setExecutorProvider` configures an executor for the
      // subscriber to process messages. Here, the subscriber is configured to open 2 streams for
      // receiving messages, each stream creates a new executor with 4 threads to help process the
      // message callbacks. In total 2x4=8 threads are used for message processing.
      subscriber =
          Subscriber.newBuilder(subscriptionName, receiver)
              .setParallelPullCount(2)
              .setExecutorProvider(executorProvider)
              .build();

      // Start the subscriber.
      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (TimeoutException timeoutException) {
      // Shut down the subscriber after 30s. Stop receiving messages.
      subscriber.stopAsync();
    }
  }
}

小茹

下列範例使用 Ruby Pub/Sub 用戶端程式庫第 3 版。如果您仍在使用第 2 版程式庫，請參閱 第 3 版遷移指南。如要查看 Ruby 第 2 版程式碼範例清單，請參閱 已淘汰的程式碼範例。

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 Ruby 設定操作說明操作。詳情請參閱 Pub/Sub Ruby API 參考文件。

# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::PubSub.new
subscriber = pubsub.subscriber subscription_id

# Use 2 threads for streaming, 4 threads for executing callbacks and 2 threads
# for sending acknowledgements and/or delays
listener = subscriber.listen streams: 2, threads: {
  callback: 4,
  push:     2
} do |received_message|
  puts "Received message: #{received_message.data}"
  received_message.acknowledge!
end

listener.start
# Let the main thread sleep for 60 seconds so the thread for listening
# messages does not quit
sleep 60
listener.stop.wait!

Rust

在試用這個範例之前，請先按照「快速入門導覽課程：使用用戶端程式庫」中的 Rust 設定說明操作。詳情請參閱 Pub/Sub Rust API 參考文件。

use google_cloud_pubsub::client::Subscriber;
use google_cloud_pubsub::subscriber::MessageStream;
use std::time::Duration;

pub async fn sample(project_id: &str, subscription_id: &str) -> anyhow::Result<()> {
    // Assume an available concurrency of 2 for this sample.
    const NCPU: usize = 2;

    let subscription_name = format!("projects/{project_id}/subscriptions/{subscription_id}");
    let client = Subscriber::builder()
        // Configure the subscriber to use multiple gRPC channels. This lets the
        // client multiplex its open streams and its acknowledgement RPCs.
        .with_grpc_subchannel_count(NCPU)
        .build()
        .await?;
    let tasks: Vec<_> = (0..2 * NCPU)
        .map(|i| {
            // Pub/Sub caps the throughput of a single stream to 10 MB/s. To achieve
            // higher throughput, you should open multiple streams.
            let stream = client.subscribe(&subscription_name).build();
            tokio::spawn(subscribe_task(i, stream))
        })
        .collect();

    for t in tasks {
        t.await??;
    }
    println!("done listening for messages");
    Ok(())
}

async fn subscribe_task(index: usize, mut stream: MessageStream) -> anyhow::Result<()> {
    // Terminate the example after 10 seconds.
    let shutdown_token = stream.shutdown_token();
    let shutdown = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(10)).await;
        shutdown_token.shutdown().await;
    });

    println!("listening for messages on stream {index}...");

    while let Some((m, h)) = stream.next().await.transpose()? {
        println!("received message: {m:?}");
        h.ack();
    }
    shutdown.await?;

    Ok(())
}

後續步驟

如要瞭解可為訂閱項目設定的其他運送選項，請參閱下列文章：

• 使用流量控制處理暫時性尖峰流量

• 使用訂閱重試政策處理訊息傳送失敗問題

• 將未送達的訊息轉送至無效信件主題

• 重播先前已確認的訊息或清除訊息

• 透過租約管理延長確認時間

• 依序接收訊息

• 依訊息屬性篩選

• 接收並確認訊息一次