Spring Boot 与 Apache Pulsar 集成构建高性能消息系统实践应用案例

<dependency>
    <groupId>org.apache.pulsar</groupId>
    <artifactId>pulsar-client</artifactId>
    <version>3.0.0</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

spring:
  pulsar:
    client:
      service-url: pulsar://localhost:6650
    admin:
      service-url: http://localhost:8080

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.CompletableFuture;
@Service
public class PulsarProducerService {
    private PulsarClient client;
    private Producer<String> producer;
    @PostConstruct
    public void init() throws Exception {
        client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();
        producer = client.newProducer(Schema.STRING)
                .topic("persistent://public/default/my-topic")
                .create();
    }
    public void sendMessage(String message) throws Exception {
        producer.send(message);
    }
    public CompletableFuture<String> sendAsyncMessage(String message) {
        return producer.sendAsync(message);
    }
    @PreDestroy
    public void close() throws Exception {
        if (producer != null) {
            producer.close();
        }
        if (client != null) {
            client.close();
        }
    }
}

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.TimeUnit;
@Service
public class PulsarConsumerService {
    private PulsarClient client;
    private Consumer<String> consumer;
    @PostConstruct
    public void init() throws Exception {
        client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();
        consumer = client.newConsumer(Schema.STRING)
                .topic("persistent://public/default/my-topic")
                .subscriptionName("my-subscription")
                .subscriptionType(SubscriptionType.Exclusive)
                .messageListener((consumer, msg) -> {
                    try {
                        System.out.println("Received message: " + new String(msg.getData()));
                        consumer.acknowledge(msg);
                    } catch (Exception e) {
                        consumer.negativeAcknowledge(msg);
                    }
                })
                .subscribe();
    }
    @PreDestroy
    public void close() throws Exception {
        if (consumer != null) {
            consumer.close();
        }
        if (client != null) {
            client.close();
        }
    }
}

producer = client.newProducer(Schema.STRING)
        .topic("persistent://public/default/my-partitioned-topic")
        .create();
// 发送消息到指定分区
producer.newMessage()
        .value("Hello Pulsar")
        .key("key1") // 基于key分区
        .send();

producer = client.newProducer(Schema.STRING)
        .topic("persistent://public/default/my-topic")
        .batchingEnabled(true)
        .batchingMaxMessages(1000)
        .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
        .create();

// 开启事务
Transaction txn = client.newTransaction()
        .withTransactionTimeout(1, TimeUnit.MINUTES)
        .build()
        .get();
// 在事务中发送消息
producer.newMessage(txn)
        .value("Hello Transaction")
        .send();
// 提交事务
txn.commit().get();

consumer = client.newConsumer(Schema.STRING)
        .topic("persistent://public/default/my-topic")
        .subscriptionName("my-subscription")
        .deadLetterPolicy(DeadLetterPolicy.builder()
                .maxRedeliverCount(10)
                .deadLetterTopic("persistent://public/default/my-dlq")
                .build())
        .subscribe();