Java操作Pulsar消息队列教程

有志者，事竟成！如果你在学习文章，那么本文《Java操作Pulsar消息队列指南》，就很适合你！文章讲解的知识点主要包括，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

在Java中操作Pulsar消息队列的核心是使用其官方Java客户端库，通过引入依赖并构建PulsarClient实例连接集群，再创建Producer发送消息或Consumer接收消息。1. 引入Maven依赖配置以获取Pulsar客户端库；2. 初始化PulsarClient实例，作为所有操作的基础；3. 创建Producer并通过同步或异步方式发送消息，推荐开启批量发送和使用sendAsync()提升性能；4. 创建Consumer订阅主题并接收消息，支持多种订阅模式如Shared、Exclusive等；5. 在消费过程中合理使用消息确认机制（acknowledge/NACK）和异常处理策略，确保消息可靠处理；6. 通过设置Message Key实现消息的有序性或路由控制；7. 合理管理客户端及生产者生命周期，避免频繁创建销毁造成资源浪费。Pulsar与传统MQ相比具有计算存储分离、多租户支持、统一消息模型等优势，适用于高可用、跨区域部署等多种场景。

在Java中操作Pulsar消息队列，核心在于利用其官方提供的Java客户端库。这套库设计得相当直观，你需要做的就是引入相应的依赖，然后通过构建PulsarClient实例来连接到Pulsar集群。一旦客户端建立，你就可以创建Producer来发送消息，或者创建Consumer来订阅并接收消息。整个过程围绕着客户端、生产者和消费者这三大核心组件展开，它们提供了丰富且灵活的API来满足各种消息传递场景的需求。

解决方案

要在Java中开发Pulsar客户端，首先你需要将Apache Pulsar Java客户端库添加到你的项目依赖中。如果你使用Maven，可以这样配置：

<dependency>
    <groupId>org.apache.pulsar</groupId>
    <artifactId>pulsar-client</artifactId>
    <version>2.11.0</version> <!-- 请替换为Pulsar集群兼容的最新稳定版本 -->
</dependency>

接着，你可以按照以下步骤来操作Pulsar：

1. 初始化Pulsar客户端

这是所有操作的起点。PulsarClient是线程安全的，通常一个应用只需要一个实例。

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

public class PulsarClientExample {

    private static final String SERVICE_URL = "pulsar://localhost:6650"; // 或 pulsar+ssl://your-broker-url:6651

    public static void main(String[] args) {
        PulsarClient client = null;
        try {
            client = PulsarClient.builder()
                    .serviceUrl(SERVICE_URL)
                    .build();
            System.out.println("Pulsar client initialized successfully.");

            // 可以在这里调用发送和接收消息的方法
            // sendMessage(client);
            // receiveMessage(client);

        } catch (PulsarClientException e) {
            System.err.println("Failed to initialize Pulsar client: " + e.getMessage());
            e.printStackTrace();
        } finally {
            if (client != null) {
                try {
                    client.close(); // 关闭客户端，释放资源
                } catch (PulsarClientException e) {
                    System.err.println("Failed to close Pulsar client: " + e.getMessage());
                }
            }
        }
    }
}

2. 发送消息（Producer）

创建Producer实例来向特定主题（Topic）发送消息。你可以同步发送，也可以异步发送。异步发送在生产环境中更为常见，因为它不会阻塞主线程，能带来更高的吞吐量。

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.MessageId;
import java.util.concurrent.TimeUnit;

public class MessageSender {

    private static final String TOPIC_NAME = "persistent://public/default/my-topic";

    public static void sendMessage(PulsarClient client) throws PulsarClientException {
        Producer<byte[]> producer = null;
        try {
            producer = client.newProducer()
                    .topic(TOPIC_NAME)
                    .producerName("my-java-producer")
                    .enableBatching(true) // 开启批量发送
                    .batchingMaxMessages(1000) // 批量消息最大数量
                    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) // 批量发送延迟
                    .sendTimeout(30, TimeUnit.SECONDS) // 发送超时
                    .blockIfQueueFull(true) // 如果发送队列满了，则阻塞
                    .create();

            // 同步发送
            MessageId msgId = producer.send("Hello Pulsar Sync!".getBytes());
            System.out.println("Sent message synchronously with ID: " + msgId);

            // 异步发送
            producer.sendAsync("Hello Pulsar Async!".getBytes()).thenAccept(id -> {
                System.out.println("Sent message asynchronously with ID: " + id);
            }).exceptionally(ex -> {
                System.err.println("Failed to send message asynchronously: " + ex.getMessage());
                return null;
            });

            // 发送带Key的消息，用于有序消费或路由
            producer.newMessage()
                    .key("my-message-key-1")
                    .value("Keyed message content".getBytes())
                    .sendAsync().thenAccept(id -> {
                        System.out.println("Sent keyed message with ID: " + id);
                    });

            // 等待异步消息发送完成，生产环境通常不需要这样等待，而是通过回调处理
            Thread.sleep(1000); // 简单等待一下，确保异步消息有机会发送

        } catch (Exception e) {
            System.err.println("Failed to send message: " + e.getMessage());
            e.printStackTrace();
        } finally {
            if (producer != null) {
                try {
                    producer.close();
                } catch (PulsarClientException e) {
                    System.err.println("Failed to close producer: " + e.getMessage());
                }
            }
        }
    }
}

3. 接收消息（Consumer）

创建Consumer实例来订阅特定主题，并从中接收消息。Pulsar支持多种订阅模式，以适应不同的消费需求。

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.SubscriptionType;
import java.util.concurrent.TimeUnit;

public class MessageReceiver {

    private static final String TOPIC_NAME = "persistent://public/default/my-topic";
    private static final String SUBSCRIPTION_NAME = "my-java-subscription";

    public static void receiveMessage(PulsarClient client) throws PulsarClientException {
        Consumer<byte[]> consumer = null;
        try {
            consumer = client.newConsumer()
                    .topic(TOPIC_NAME)
                    .subscriptionName(SUBSCRIPTION_NAME)
                    .subscriptionType(SubscriptionType.Shared) // 共享订阅模式
                    .messageListener((cons, msg) -> { // 异步消息监听器
                        try {
                            System.out.println("Received message: " + new String(msg.getData()) +
                                    ", ID: " + msg.getMessageId() + ", Key: " + msg.getKey());
                            cons.acknowledge(msg); // 确认消息，表示已成功处理
                        } catch (Exception e) {
                            System.err.println("Error processing message: " + e.getMessage());
                            cons.negativeAcknowledge(msg); // 负确认，消息会被重新投递
                        }
                    })
                    .subscribe();

            System.out.println("Consumer subscribed to topic " + TOPIC_NAME + " with subscription " + SUBSCRIPTION_NAME);

            // 保持主线程运行，以便消费者可以持续接收消息
            // 生产环境通常是守护线程或由框架管理
            Thread.sleep(Long.MAX_VALUE); // 简单地让程序一直运行

        } catch (Exception e) {
            System.err.println("Failed to receive message: " + e.getMessage());
            e.printStackTrace();
        } finally {
            if (consumer != null) {
                try {
                    consumer.close();
                } catch (PulsarClientException e) {
                    System.err.println("Failed to close consumer: " + e.getMessage());
                }
            }
        }
    }
}

Pulsar客户端与传统MQ有何不同？

说实话，第一次接触Pulsar的时候，我个人觉得它和Kafka、RabbitMQ这些“老牌”MQ在概念上挺像的，都是生产者、消费者、主题那一套。但深入了解后，你会发现Pulsar在架构设计上走了一条完全不一样的路，这直接影响了它的客户端使用方式和能提供的特性。

最核心的区别在于Pulsar将计算和存储分离了。Broker负责处理消息的路由和分发，而消息的实际存储则交给了BookKeeper集群。这种分离带来的好处是显而易见的：扩容伸缩变得异常灵活，你可以独立地扩展计算能力（Broker）和存储能力（BookKeeper），互不影响。这在传统MQ中，Broker往往既负责路由又负责存储，扩容时可能会遇到瓶颈。

再者，Pulsar天生支持多租户（Multi-tenancy）和地理复制（Geo-replication）。这意味着你可以在一个Pulsar集群上轻松地为不同的团队或应用创建独立的命名空间（Namespace），彼此隔离，互不干扰。而且，消息可以轻松地在不同数据中心之间进行复制，这对于构建高可用、跨区域的应用来说简直是福音。想想看，在传统MQ里要实现这些，往往需要额外的组件或者复杂的配置，Pulsar直接就给你集成好了。

还有一点是统一的消息模型。Pulsar既能像Kafka那样处理流数据（Streaming），也能像RabbitMQ那样处理队列消息（Queuing）。这意味着你的应用可以根据实际需求选择不同的订阅模式，比如共享订阅可以实现负载均衡的消费，而独占订阅则能保证消息的严格顺序。这种灵活性，在很多场景下，能省去不少麻烦。我记得以前为了同时满足流式处理和任务队列的需求，可能得部署两套不同的消息系统，Pulsar一个就搞定了。

在Java中实现Pulsar消息发送的最佳实践是什么？

在Java应用里用Pulsar发消息，可不是简单地调用个send()方法就完事儿。想要真正发挥Pulsar的高性能和可靠性，有些实践是必须要考虑的。

首先，PulsarClient和Producer实例的生命周期管理至关重要。PulsarClient是重量级对象，应该在应用启动时创建一次，并在整个应用生命周期内复用，通常作为单例。而Producer虽然可以每次发送消息时都创建，但更推荐的做法是也将其池化或者作为单例复用，因为创建Producer涉及到与Broker的连接建立和资源分配，频繁创建会带来不必要的开销。我见过不少新手项目，每次发消息都new Producer()，性能问题很快就暴露出来了。

其次，强烈推荐使用异步发送 (sendAsync())。同步发送会阻塞调用线程，直到消息被Broker确认，这在高并发场景下是性能杀手。sendAsync()返回一个CompletableFuture，你可以通过回调函数（thenAccept, exceptionally）来处理发送成功或失败的逻辑。这样，你的应用线程可以立即返回去处理其他任务，大大提升了吞吐量。

再有，消息的批量发送（Batching）也是提升性能的关键。Pulsar客户端默认是开启批量发送的，它会将短时间内发送的多条小消息聚合成一个大的批次再发送给Broker。这能有效减少网络IO和Broker的处理开销。你可以通过enableBatching(true)、batchingMaxMessages和batchingMaxPublishDelay等参数来调整批处理策略。合理配置这些参数，能让你的Pulsar发送性能上一个台阶。

最后，别忘了消息的键（Message Key）。如果你需要保证消息的顺序性，或者希望将特定类型的消息路由到同一个消费者（在共享订阅模式下），给消息设置一个有意义的key是很有用的。Pulsar会根据消息的key进行哈希，确保相同key的消息总是被发送到同一个分区（Partition），从而保证了有序性。

如何有效地消费Pulsar消息并处理异常？

消费Pulsar消息，并不仅仅是receive()然后处理那么简单。一个健壮的消费者，必须能够妥善处理各种异常情况，并确保消息的可靠性。

消息的确认（Acknowledgement）机制是核心。当你成功处理完一条消息后，必须调用consumer.acknowledge(message)来告诉Pulsar这条消息可以被安全地删除了。如果没有确认，Pulsar会认为这条消息没有被成功处理，并在一定时间后重新投递。Pulsar提供了两种确认方式：单条确认（acknowledge(MessageId)）和累积确认（acknowledgeCumulative(MessageId)）。累积确认会确认所有比指定消息ID更早的消息，这在处理有序流时非常有用，但在乱序处理时要小心。

异常处理是消费者逻辑中不可或缺的部分。如果你的业务逻辑在处理消息时抛出了异常，你不能简单地忽略它。正确的做法是调用consumer.negativeAcknowledge(message)（简称NACK）。NACK会告诉Pulsar这条消息处理失败了，Pulsar会在稍后重新投递这条消息。这对于临时性的错误（比如数据库连接中断）非常有用。对于那些无法处理的“坏消息”（比如数据格式错误），可以考虑将其发送到死信队列（Dead Letter Queue, DLQ）。Pulsar客户端支持配置DLQ，这样那些反复处理失败的消息就不会一直占用资源，而是被隔离起来供后续分析。

在订阅模式的选择上，Shared订阅模式非常适合需要负载均衡和高吞吐量的场景，多个消费者可以同时消费同一个主题的消息。而Exclusive或Failover模式则适用于需要严格消息顺序或主备高可用的场景。选择合适的订阅模式，直接影响你的消费逻辑和异常处理策略。

还有一点，消费者通常会通过消息监听器（messageListener）异步地接收消息。这意味着你的消息处理逻辑是在Pulsar客户端的内部线程池中执行的。如果你的处理逻辑非常耗时，可能会阻塞Pulsar的内部线程，影响其他消息的接收。在这种情况下，你可以考虑将消息放入一个内部队列，然后由你自己的线程池来异步处理这些消息，从而实现解耦和背压控制。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~