Use a Java client

Create a producer

Once you’ve instantiated a PulsarClient object, you can create a Producer for a specific Pulsar topic.

  1. Producer<byte[]> producer = client.newProducer()
  2.         .topic("my-topic")
  3.         .create();
  5. // You can then send messages to the broker and topic you specified:
  6. producer.send("My message".getBytes());

By default, producers produce messages that consist of byte arrays. You can produce different types by specifying a message schema.

  1. Producer<String> stringProducer = client.newProducer(Schema.STRING)
  2.         .topic("my-topic")
  3.         .create();
  4. stringProducer.send("My message");

Make sure that you close your producers, consumers, and clients when you do not need them.

  1. producer.close();
  2. consumer.close();
  3. client.close();

Close operations can also be asynchronous:

  1. producer.closeAsync()
  2.    .thenRun(() -> System.out.println("Producer closed"))
  3.    .exceptionally((ex) -> {
  4.        System.err.println("Failed to close producer: " + ex);
  5.        return null;
  6.    });

Create a consumer

In Pulsar, consumers subscribe to topics and handle messages that producers publish to those topics. You can instantiate a new consumer by first instantiating a PulsarClient object and passing it a URL for a Pulsar broker (as above).

Once you’ve instantiated a PulsarClient object, you can create a Consumer by specifying a topic and a subscription.

  1. Consumer consumer = client.newConsumer()
  2.         .topic("my-topic")
  3.         .subscriptionName("my-subscription")
  4.         .subscribe();

The subscribe method will auto-subscribe the consumer to the specified topic and subscription. One way to make the consumer listen to the topic is to set up a while loop. In this example loop, the consumer listens for messages, prints the contents of any received message, and then acknowledges that the message has been processed. If the processing logic fails, you can use negative acknowledgment to redeliver the message later.

  1. while (true) {
  2.   // Wait for a message
  3.   Message msg = consumer.receive();
  5.   try {
  6.       // Do something with the message
  7.       System.out.println("Message received: " + new String(msg.getData()));
  9.       // Acknowledge the message
  10.       consumer.acknowledge(msg);
  11.   } catch (Exception e) {
  12.       // Message failed to process, redeliver later
  13.       consumer.negativeAcknowledge(msg);
  14.   }
  15. }

If you don’t want to block your main thread but constantly listen for new messages, consider using a MessageListener. The MessageListener uses a thread pool inside the client. You can set the number of threads for message listeners in the ClientBuilder. The MessageListener will use a thread pool inside the PulsarClient. You can set the number of threads to use for message listeners in the ClientBuilder.

  1. MessageListener myMessageListener = (consumer, msg) -> {
  2.   try {
  3.       System.out.println("Message received: " + new String(msg.getData()));
  4.       consumer.acknowledge(msg);
  5.   } catch (Exception e) {
  6.       consumer.negativeAcknowledge(msg);
  7.   }
  8. }
  10. Consumer consumer = client.newConsumer()
  11.      .topic("my-topic")
  12.      .subscriptionName("my-subscription")
  13.      .messageListener(myMessageListener)
  14.      .subscribe();

Create a reader

With the reader interface, Pulsar clients can “manually position” themselves within a topic and read all messages from a specified message onward. The Pulsar API for Java enables you to create Reader objects by specifying a topic and a MessageId.

The following is an example.

  1. byte[] msgIdBytes = // Some message ID byte array
  2. MessageId id = MessageId.fromByteArray(msgIdBytes);
  3. Reader reader = pulsarClient.newReader()
  4.         .topic(topic)
  5.         .startMessageId(id)
  6.         .create();
  8. while (true) {
  9.     Message message = reader.readNext();
  10.     // Process message
  11. }

In the example above, a Reader object is instantiated for a specific topic and message (by ID); the reader iterates over each message in the topic after the message is identified by msgIdBytes (how that value is obtained depends on the application).

The code sample above shows pointing the Reader object to a specific message (by ID), but you can also use MessageId.earliest to point to the earliest available message on the topic of MessageId.latest to point to the most recent available message.