Introducing Apache Pulsar and Its Integration with Spring Boot

Spring Boot is a mainstream micro‑service framework with a mature ecosystem, and this guide presents a quick‑start series for integrating common middleware, including Apache Pulsar, into Spring Boot applications.

Pulsar Introduction

Pulsar was created by Yahoo in 2013, donated to the Apache Foundation in 2016, and is now a top‑level Apache project used by companies such as Yahoo, Verizon, and Twitter for massive messaging workloads.

It claims to be faster and cheaper to run than Kafka while addressing many of Kafka's pain points.

Pulsar is flexible: it can act as a distributed log system like Kafka or as a simple message broker similar to RabbitMQ .

It offers multiple subscription types, delivery guarantees, and retention policies.

Features

Built‑in multi‑tenant support, allowing isolated usage of the same cluster by different teams with authentication, authorization, and quota controls.

Multi‑layer architecture: data is stored in a dedicated layer using Apache BookKeeper, separating brokers from storage, which improves scalability, rebalancing, maintenance, and availability.

Multi‑layer storage: old data can be automatically moved to external storage such as Amazon S3 while remaining transparent to clients.

Functions: lightweight, developer‑friendly APIs for stream processing without needing a separate engine.

Security: built‑in proxy, multi‑tenant security mechanisms, and pluggable authentication.

Fast rebalancing: partitions are split into small chunks, enabling rapid rebalancing.

Integration with many systems: easy integration with Kafka, RabbitMQ, etc.

Support for multiple programming languages including Go, Java, Scala, Node, and Python.

Why Choose Pulsar

While Kafka dominates large‑scale log processing, it lacks features needed in some scenarios such as topic‑level deletion and failover. Pulsar provides native high consistency, high‑availability storage via BookKeeper, a decoupled storage‑service architecture for easy scaling, multiple consumption modes, and multi‑region deployment.

Show Me the Code

External Dependency:

<dependency>
    <groupId>org.apache.pulsar</groupId>
    <artifactId>pulsar-client</artifactId>
    <version>2.4.0</version>
</dependency>

Configuration (application.yaml):

pulsar:
  service:
    url: pulsar://127.0.0.1:6650

Producer – Sending Messages:

Long id = Long.valueOf(new Random().nextInt(1000));
User user = User.builder().id(id).userName("TomGE").age(29).address("上海").build();
userPulsarMsgProducer.send(user);

The producer is initialized at application startup using @PostConstruct to create an org.apache.pulsar.client.api.Producer bean managed by Spring.

public void send(T msg) {
    String msgBody = JSON.toJSONString(msg);
    try {
        MessageId messageId = producer.send(msgBody.getBytes(StandardCharsets.UTF_8));
        log.info("pulsar msg send success, topic:{}, messageId:{}, msg:{}", getTopic(), messageId, msgBody);
    } catch (Throwable e) {
        log.error("pulsar msg send failed, topic:{}, msg:{}", getTopic(), msgBody);
    }
}

Producer – Delayed Messages: Useful for scenarios such as automatic order confirmation after a grace period.

producer.newMessage().deliverAfter(delay, unit)
        .value(msgBody.getBytes(StandardCharsets.UTF_8))
        .send();

Consumer – Receiving Messages:

@PostConstruct
void init() throws PulsarClientException {
    consumer = client.createConsumer(getTopic(), getSubscriptionName(), new DefaultJsonMsgListener());
}

class DefaultJsonMsgListener implements MessageListener
{
    @Override
    public void received(Consumer
consumer, Message
message) {
        if (message != null && message.getData() != null && message.getData().length != 0) {
            String msgBody = new String(message.getValue(), StandardCharsets.UTF_8);
            log.warn("topic:{} receive message:{}", getTopic(), msgBody);
            try {
                T msg = JSON.parseObject(msgBody, clazzT);
                handleMsg(msg);
            } catch (Exception e) {
                log.error("handle msg failed, topic:{}, message:{}", getTopic(), msgBody, e);
                return;
            }
            try {
                consumer.acknowledge(message);
            } catch (PulsarClientException e) {
                log.error("topic:{} ack failed", getTopic(), e);
            }
        }
    }
}

Demo Code Repository

https://github.com/aalansehaiyang/spring-boot-bulking
Module: spring-boot-bulking-pulsar

Additional recommended reading links are provided at the end of the original article.