Getting started with Pulsar middleware

Articles are included in my Github selection, welcome to Star!

Brief introduction

Pulsar is a server-to-server messaging system with multi-tenancy, high performance, and more. Originally developed by Yahoo, it is currently managed by the Apache Software Foundation. It is the top project of Apache Software Foundation, is the next generation of cloud-native distributed message flow platform, set messaging, storage, lightweight functional computing as one, using computing and storage separation architecture design, support multi-tenant, persistent storage, multi-room cross-regional data replication, with strong consistency, high throughput, low latency and high scalability and other stream data storage features, is regarded as the cloud native era real-time message flow transmission, storage and computing best solution.

characteristic

• A single instance of Pulsar natively supports multiple clusters, seamlessly replicating messages across clusters across the machine room.
• Extremely low release and end-to-end latency.
• Seamlessly scales to over a million topics.
• Easy-to-use client API with support for Java, Go, Python, and C++.
• Supports multiple topic subscription models (exclusive subscription, shared subscription, failover subscription).
• Message delivery is guaranteed through the persistent message store mechanism provided by Apache BookKeeper.
• Pulsar IO, a serverless connector framework based on Pulsar Functions, makes it easier to move data in and out of Apache Pulsar.
• Stream-native data processing is implemented by Pulsar Functions, a lightweight serverless computing framework.
• Tiered storage offloads data from hot storage to cold/long-term storage (e.g., S3, GCS) when the data is stale.

Architecture

This is the architecture diagram on the official website, involving several components, here is a brief explanation:

Broker: The broker is responsible for the transmission of messages, the management of topics and load balancing, the broker is not responsible for the storage of messages, is a stateless component.

Bookie: Responsible for the persistence of messages, using the Apache BookKeeper component, a distributed WAL system.

Producer: Producer, who encapsulates messages and sends them to Broker synchronously or asynchronously.

Consumer: Consumers, who consume messages by subscribing to Topic, and confirm. The Reader role is also defined in Pulsar, which is also a consumer, except that it can get messages from a specified position without acknowledgment.

Zookeeper: Metadata storage, responsible for the configuration management of the cluster, including tenants, namespaces, etc., and for consistency coordination.

Four subscription models

When introducing the Pulsar feature, it is said that there are multiple subscription modes supported, a total of four, namely exclusive subscription, shared subscription, failover subscription, and key (key_shared) sharing.

Exclusive

Exclusive mode: Only one consumer can initiate and consume data at a time; the same consumer is identified by SubscriptionName), which is less applicable.

Shared

There can be N consumers running at the same time, and the message is delivered to each consumer in round-robin polling; when a consumer goes down without an ack, the message is delivered to other consumers. This consumption pattern can increase consumption capacity, but the message cannot be orderly.

Failover

Failover mode: On the basis of exclusive mode, multiple consumers can be started at the same time, once a consumer is hung up, the rest can be quickly toppled, but only one consumer can be consumed; some scenarios are available.

KeyShared

Based on the sharing mode; it is equivalent to grouping messages in the same topic, and messages in the same group can only be consumed by the same consumer in an orderly manner.

Download and install

I have version 2.9.1 of pulsar installed here, and the link address is as follows:

https://www.apache.org/dyn/mirrors/mirrors.cgi?action=download&filename=pulsar/pulsar-2.9.1/apache-pulsar-2.9.1-bin.tar.gz

After the download is complete, upload to the Linux server and unzip it using the command:

tar -zxvf apache-pulsar-2.9.0-bin.tar.gz

For stand-alone versions, use the command background to start:

./bin/pulsar-daemon start standalone

To terminate a command running in the background:

./bin/pulsar-daemon stop standalone

SpringBoot integration

Once booted on the Linux server, it's time to use the Java client to get to the steps, starting with the introduction of Maven dependencies:

<dependency>
    <groupId>io.github.majusko</groupId>
    <artifactId>pulsar-java-spring-boot-starter</artifactId>
    <version>1.1.0</version>
</dependency>
           

application.yml configuration file plus configuration:

#pulsar的服务地址
pulsar:
  service-url: pulsar://192.168.0.105:6650
           

Add a configuration class PulsarConfig:

@Configuration
public class PulsarConfig {

    @Bean
    public ProducerFactory producerFactory() {
        return new ProducerFactory().addProducer("testTopic", String.class);
    }
}
           

Add a constant class that records the topic name:

/**
 * Pulsar中间件的topic名称
 *
 * @author yehongzhi
 * @date 2022/4/9 5:57 PM
 */
public class TopicName {

    private TopicName(){}
    /**
     * 测试用的topic
     */
    public static final String TEST_TOPIC = "testTopic";
}
           

Add message producer PulsarProducer class:

/**
 * Pulsar生产者
 *
 * @author yehongzhi
 * @date 2022/4/9 5:23 PM
 */
@Component
public class PulsarProducer<T> {

    @Resource
    private PulsarTemplate<T> template;

    /**
     * 发送消息
     */
    public void send(String topic, T message) {
        try {
            template.send(topic, message);
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }
}
           

Add a consumer with the name "testTopic":

/**
 * topic名称为"testTopic"对应的消费者
 *
 * @author yehongzhi
 * @date 2022/4/9 6:00 PM
 */
@Component
public class TestTopicPulsarConsumer {

    private static final Logger log = LoggerFactory.getLogger(TestTopicPulsarConsumer.class);

    //SubscriptionType.Shared，表示共享模式
    @PulsarConsumer(topic = TopicName.TEST_TOPIC,
                    subscriptionType = SubscriptionType.Shared,
                    clazz = String.class)
    public void consume(String message) {
        log.info("PulsarRealConsumer content:{}", message);
    }

}
           

Finally add a PulsarController test to send a message:

@RestController
@RequestMapping("/pulsar")
public class PulsarController {

    @Resource
    private PulsarProducer<String> pulsarProducer;

    @PostMapping(value = "/sendMessage")
    public CommonResponse<String> sendMessage(@RequestParam(name = "message") String message) {
        pulsarProducer.send(TopicName.TEST_TOPIC, message);
        return CommonResponse.success("done");
    }
}
           

CommonResponse public responder:

public class CommonResponse<T> {

    private String code;

    private Boolean success;

    private T data;

    public static <T>CommonResponse<T> success(T t){
        return new CommonResponse<>("200",true,t);
    }

    public CommonResponse(String code, Boolean success, T data) {
        this.code = code;
        this.success = success;
        this.data = data;
    }
    //getter、setter方法
}
           

Start the project and then test with postman:

summary

The above is a simple introduction to pulsar middleware, which introduces the features, architecture, subscription model, and a small example of integrating SpringBoot. Finally, make a comparison with other mainstream middleware for your reference:

Thank you all for reading, and hope this article is helpful to you.

If you think it is useful, just like it, your like is the biggest motivation for my creation

I'm a programmer who strives to be remembered. We'll see you next time!!!

Ability is limited, if there is any mistake or inappropriate, please criticize and correct, learn and communicate together!