Memproses lebih banyak pesan dengan kontrol konkurensi

Poin Penting: Kontrol serentak memungkinkan Anda mengonfigurasi jumlah thread atau stream yang digunakan library klien Pub/Sub untuk menarik pesan. Dengan lebih banyak thread, Anda dapat memproses lebih banyak pesan secara serentak.

Kontrol konkurensi adalah fitur yang tersedia di library klien tingkat tinggi Pub/Sub. Anda juga dapat menerapkan kontrol serentak sendiri saat menggunakan library level rendah.

Dukungan untuk kontrol serentak bergantung pada bahasa pemrograman library klien. Untuk implementasi bahasa yang mendukung thread paralel, seperti C++, Go, dan Java, library klien membuat pilihan default untuk jumlah thread.

Pilihan ini mungkin tidak optimal untuk aplikasi Anda. Misalnya, jika aplikasi pelanggan Anda tidak dapat mengimbangi volume pesan masuk dan tidak terikat CPU, Anda harus meningkatkan jumlah thread. Untuk operasi pemrosesan pesan yang menggunakan CPU secara intensif, mengurangi jumlah thread mungkin sesuai.

Konfigurasi kontrol serentak

Nilai default untuk variabel kontrol konkurensi dan nama variabel dapat berbeda di berbagai library klien. Misalnya, dalam library klien Java, metode untuk mengonfigurasi kontrol serentak adalah setSystemExecutorProvider(), setExecutorProvider(), dan setParallelPullCount().

• setParallelPullCount() memungkinkan Anda menentukan jumlah streaming yang akan dibuka. Anda dapat membuka lebih banyak streaming jika klien pelanggan Anda dapat menangani lebih banyak data daripada yang dikirim pada satu streaming dengan kecepatan 10 MBps.

• setExecutorProvider() memungkinkan Anda menyesuaikan penyedia eksekutor yang digunakan untuk memproses pesan. Misalnya, Anda dapat mengubah penyedia eksekutor menjadi penyedia yang menampilkan satu eksekutor bersama dengan jumlah thread terbatas di beberapa klien pelanggan. Konfigurasi ini membantu membatasi jumlah thread yang dibuat.

• setSystemExecutorProvider() memungkinkan Anda menyesuaikan penyedia eksekutor yang digunakan untuk pengelolaan sewa. Biasanya, Anda tidak mengonfigurasi nilai ini kecuali Anda ingin menggunakan penyedia eksekutor yang sama di setExecutorProvider dan setSystemExecutorProvider. Misalnya, Anda dapat menggunakan penyedia eksekutor yang sama jika memiliki sejumlah langganan dengan throughput rendah. Menggunakan nilai yang sama akan membatasi jumlah thread di klien.

Jumlah total thread yang digunakan untuk kontrol serentak bergantung pada penyedia eksekutor yang diteruskan dalam library klien dan jumlah pull paralel.

Contoh kode untuk kontrol konkurensi

C++

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan C++ di Panduan Memulai: Menggunakan Library Klien. Untuk informasi selengkapnya, lihat dokumentasi referensi 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Panduan Memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Pub/Sub Go API.

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

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

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()

	sub := client.Subscription(subID)
	// Must set ReceiveSettings.Synchronous to false (or leave as default) to enable
	// concurrency pulling of messages. Otherwise, NumGoroutines will be set to 1.
	sub.ReceiveSettings.Synchronous = false
	// NumGoroutines determines the number of goroutines sub.Receive will spawn to pull
	// messages.
	sub.ReceiveSettings.NumGoroutines = 16
	// MaxOutstandingMessages limits the number of concurrent handlers of messages.
	// In this case, up to 8 unacked messages can be handled concurrently.
	// Note, even in synchronous mode, messages pulled in a batch can still 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Panduan Memulai: Menggunakan Library Klien. Untuk informasi selengkapnya, lihat dokumentasi referensi API Pub/Sub Java.

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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Ruby di Panduan Memulai: Menggunakan Library Klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Pub/Sub Ruby API.

# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::Pubsub.new

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

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

Langkah selanjutnya

Baca opsi pengiriman lainnya yang dapat Anda konfigurasi untuk langganan:

• Menangani lonjakan sementara dengan kontrol alur

• Menangani kegagalan pesan dengan kebijakan percobaan ulang langganan

• Meneruskan pesan yang belum terkirim ke topik yang dihentikan pengirimannya

• Memutar ulang pesan yang sebelumnya dikonfirmasi atau menghapus permanen pesan

• Memperpanjang waktu ack dengan pengelolaan sewa

• Menerima pesan secara berurutan

• Memfilter pesan menurut atributnya

• Menerima dan mengonfirmasi pesan tepat satu kali