Using Pulsar as a message queue

Message queues are essential components of many large-scale data architectures. If every single work object that passes through your system absolutely must be processed in spite of the slowness or downright failure of this or that system component, there’s a good chance that you’ll need a message queue to step in and ensure that unprocessed data is retained—-with correct ordering—-until the required actions are taken.

Pulsar is a great choice for a message queue because:

You can use the same Pulsar installation to act as a real-time message bus and as a message queue if you wish (or just one or the other). You can set aside some topics for real-time purposes and other topics for message queue purposes (or use specific namespaces for either purpose if you wish).

Client configuration changes

To use a Pulsar topic as a message queue, you should distribute the receiver load on that topic across several consumers (the optimal number of consumers will depend on the load). Each consumer must:

  • Establish a shared subscription and use the same subscription name as the other consumers (otherwise the subscription is not shared and the consumers can’t act as a processing ensemble)

  • If you’d like to have tight control over message dispatching across consumers, set the consumers’ receiver queue size very low (potentially even to 0 if necessary). Each Pulsar consumer has a receiver queue that determines how many messages the consumer will attempt to fetch at a time. A receiver queue of 1000 (the default), for example, means that the consumer will attempt to process 1000 messages from the topic’s backlog upon connection. Setting the receiver queue to zero essentially means ensuring that each consumer is only doing one thing at a time.

    The downside to restricting the receiver queue size of consumers is that that limits the potential throughput of those consumers and cannot be used with partitioned topics. Whether the performance/control trade-off is worthwhile will depend on your use case.

Java clients

Here’s an example Java consumer configuration that uses a shared subscription:

  2. import org.apache.pulsar.client.api.Consumer;
  3. import org.apache.pulsar.client.api.PulsarClient;
  4. import org.apache.pulsar.client.api.SubscriptionType;
  6. String SERVICE_URL = "pulsar://localhost:6650";
  7. String TOPIC = "persistent://public/default/mq-topic-1";
  8. String subscription = "sub-1";
  10. PulsarClient client = PulsarClient.builder()
  11.         .serviceUrl(SERVICE_URL)
  12.         .build();
  14. Consumer consumer = client.newConsumer()
  15.         .topic(TOPIC)
  16.         .subscriptionName(subscription)
  17.         .subscriptionType(SubscriptionType.Shared)
  18.         // If you'd like to restrict the receiver queue size
  19.         .receiverQueueSize(10)
  20.         .subscribe();

Python clients

Here’s an example Python consumer configuration that uses a shared subscription:

  2. from pulsar import Client, ConsumerType
  4. SERVICE_URL = "pulsar://localhost:6650"
  5. TOPIC = "persistent://public/default/mq-topic-1"
  6. SUBSCRIPTION = "sub-1"
  8. client = Client(SERVICE_URL)
  9. consumer = client.subscribe(
  10.     TOPIC,
  11.     SUBSCRIPTION,
  12.     # If you'd like to restrict the receiver queue size
  13.     receiver_queue_size=10,
  14.     consumer_type=ConsumerType.Shared)

C++ clients

Here’s an example C++ consumer configuration that uses a shared subscription:

  2. #include <pulsar/Client.h>
  4. std::string serviceUrl = "pulsar://localhost:6650";
  5. std::string topic = "persistent://public/defaultmq-topic-1";
  6. std::string subscription = "sub-1";
  8. Client client(serviceUrl);
  10. ConsumerConfiguration consumerConfig;
  11. consumerConfig.setConsumerType(ConsumerType.ConsumerShared);
  12. // If you'd like to restrict the receiver queue size
  13. consumerConfig.setReceiverQueueSize(10);
  15. Consumer consumer;
  17. Result result = client.subscribe(topic, subscription, consumerConfig, consumer);

Go clients

Here is an example of a Go consumer configuration that uses a shared subscription:

  2. import "github.com/apache/pulsar-client-go/pulsar"
  4. client, err := pulsar.NewClient(pulsar.ClientOptions{
  5.     URL: "pulsar://localhost:6650",
  6. })
  7. if err != nil {
  8.     log.Fatal(err)
  9. }
  10. consumer, err := client.Subscribe(pulsar.ConsumerOptions{
  11.     Topic:             "persistent://public/default/mq-topic-1",
  12.     SubscriptionName:  "sub-1",
  13.     Type:              pulsar.Shared,
  14.     ReceiverQueueSize: 10, // If you'd like to restrict the receiver queue size
  15. })
  16. if err != nil {
  17.     log.Fatal(err)
  18. }