Multithreading with pulsar message listeners

The ClientBuilder#listenerThreads method allows configuring the size of an internal thread-pool that will be shared and used across all Consumers or Readers that will be created from that client.

Pulsar Client will give you the guarantee that a MessageListener for a single consumer will always be invoked by the same thread, i.e that the provided MessageListener don’t need to be thread-safe. So, YES it is prefered to use a dedicated MessageListener object per Consumer.

Note that this also ensure ordering.

So basically, if you only use a single Consumer then you can keep the listenerThreads to 1 (that is the default).

Here is a complete example that can used to observe the behaviour :

public class PulsarConsumerListenerExample {

    public static void main(String[] args) throws PulsarClientException {

        int numListenerThread = 2;

        PulsarClient client = PulsarClient
                .builder()
                .serviceUrl("pulsar://localhost:6650")
                .listenerThreads(numListenerThread)
                .build();

        final List<Consumer<?>> consumers = new ArrayList<>();
        for (int i = 0; i < numListenerThread; i++) {
            consumers.add(createConsumerWithLister(client, "my-topic", "my-subscription", "C" + i));
        }

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            for (Consumer<?> consumer : consumers) {
                try {
                    consumer.close();
                } catch (PulsarClientException e) {
                    e.printStackTrace();
                }
            }
        }));
    }

    private static Consumer<String> createConsumerWithLister(final PulsarClient client,
                                                             final String topic,
                                                             final String subscription,
                                                             final String consumerName) throws PulsarClientException {
        return client.newConsumer(Schema.STRING)
            .topic(topic)
            .consumerName(consumerName)
            .subscriptionName(subscription)
            .subscriptionMode(SubscriptionMode.Durable)
            .subscriptionType(SubscriptionType.Failover)
            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
            .messageListener((MessageListener<String>) (consumer, msg) -> {
                System.out.printf(
                    "[%s/%s]Message received: key=%s, value=%s, topic=%s, id=%s%n",
                    consumerName,
                    Thread.currentThread().getName(),
                    msg.getKey(),
                    msg.getValue(),
                    msg.getTopicName(),
                    msg.getMessageId().toString());
                consumer.acknowledgeAsync(msg);
            })
            .subscribe();
    }
}

CLICK HERE to find out more related problems solutions.

Leave a Comment

Your email address will not be published.

Scroll to Top