Apache Pulsar学习笔记06: Pulsar的Java客户端库
2021-10-27
上一节学习了Pulsar中的生产者、消费者、订阅和订阅类型。
当一个消费者连接到Pulsar时，会创建一个订阅(Subscription)，订阅有4种类型：独占模式、灾备模式、共享模式和基于key的共享模式。
订阅是命名好的配置规则，指导消息如何投递给消费者。生产者和消费者是连接到Pulsar的客户端，上节使用pulsar-client连接到Pulsar集群完成了消费者和生产者的测试。
真实场景中的生产者和消费者是我们开发的程序，从本节开始将学习如何使用各种编程语言开发Pulsar的生产者和消费者。
Pulsar为很多编程语言提供了简单易用的客户端API库，封装了客户端与Pulsar Broker之间的通信细节。Pulsar官方提供了Java, Go, Python, C++, Node.js, C#语言的客户端库，提供了Websocket的API。
除了官方正式发布的客户端库外，还有很多第三方的客户端可供选择，例如Rust, Scala, Haskell等。
本节将学习使用Pulsar的Java客户端库。可以使用Pulsar Java Client创建Pulsar的生产者、消费者、Message Reader，还能完成对Pulsar的管理任务。
本节基于java的构建工具gradle创建的java项目，并在junit单元测试代码中演示pulsar java client的使用。在项目中添加pulsar-client依赖到build.gradle文件中。
1implementation 'org.apache.pulsar:pulsar-client:2.8.1'
1.创建PulsarClient #
要使用Java客户端创建Pulsar的生产者或消费者，需要先创建一个PulsarClient对象连接到Pulsar Broker。 创建连接到Broker的PulsarClient对象需要用的pulsar协议的URL。
Pulsar使用自定义二进制协议在Producers/Consumers和Brokers之间进行通信。前面我们使用docker容器运行单机版pulsar时对外映射了两个端口，8080是pulsar的web端口，而6650则是broker提供给生产者和消费者连接的Pulsar自定义二进制协议TCP端口。
1docker ps
2CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
3e9cb8dd13c27 apachepulsar/pulsar:2.8.1 "bin/pulsar standalo…" 2 days ago Up 2 days 0.0.0.0:6650->6650/tcp, :::6650->6650/tcp, 0.0.0.0:8080->8080/tcp, :::8080->8080/tcp pulsar
Pulsar协议的URL示例如下:
1pulsar://localhost:6650
2
3# 多个broker
4pulsar://localhost:6550,localhost:6651,localhost:6652
5
6# 生产级别的pulsar集群
7pulsar://pulsar.us-west.example.com:6650
8
9# 使用TLS认证时的URL
10pulsar+ssl://pulsar.us-west.example.com:6651
Pulsar使用自定义二进制协议在Producers/Consumers和Brokers之间进行通信。Pulsar自定义的二进制协议是基于protobuf的，遵循渐进性学习的原则，这里不展开，具体看查看Pulsar文档中的https://pulsar.apache.org/docs/zh-CN/develop-binary-protocol/。
1@Test
2public void testCreatePulsarClient() throws PulsarClientException {
3 try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.13:6650").build()) {
4 System.out.println(client);
5 }
6}
从上面的测试代码可以看出，PulsarClient对象使用builder模式创建，builder的方法除了serviceUrl指定pulsar URL外，还有其他的方法用于配置认证等其他配置操作。
PulsarClient的创建为我们隐藏了客户端连接到Pulsar Broker的所有细节，例如自动重试等。
还有十分重要的一点就是: PulsarClient实例对象是线程安全，也就是说在一个项目中只需创建一个PulsarClient就可以创建和管理多个Producer和Consumer。
这也能够更好将其与Spring框架相整合。
2.使用PulsarClient创建生产者 #
接下来演示使用PulsarClient创建一个Producer，指定一个Topic，并发送消息到Topic。
1@Test
2public void testProducer() throws PulsarClientException {
3 try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.13:6650").build()) {
4 try (Producer<byte[]> producer = client.newProducer().topic("persistent://study/app1/topic-1").create()) {
5 producer.newMessage()
6 .key("msgKey1")
7 .value("hello".getBytes(StandardCharsets.UTF_8))
8 .property("p1", "v1")
9 .property("p2", "v2")
10 .send();
11 }
12 }
13}
上面的代码使用PulsarClient创建了一个Producer，创建Producer最简单的方式是设置一下Topic就行。接下来使用Producer创建了一个消息，指定了消息的Key，并为这个消息设置了两个属性，最后将消息发出。 消息的属性作为消息的元数据可以用来为消息添加一些有用的信息，例如消息是什么时间发送的，是由谁发送的等等。 消息的Key的指定就很有用了，在上一节学习订阅类型的时候，生产者可以使用基于消息Key的共享订阅模式，在这个模式下多个Consumer可以绑定在一个订阅上，同时具有相同相同key的消息被交付给相同的Consumer。 另外如果使用了分区Topic，生产者发送的消息会以消息Key做Hash进行路由，相同Key的消息会被发送到相同的分区。
3.使用PulsarClient创建消费者 #
使用PulsarClient通过指定主题Topic和订阅Subscription来创建Consumer。
1Consumer consumer = client.newConsumer()
2 .topic("my-topic")
3 .subscriptionName("my-subscription")
4 .subscribe();
上面的代码subscribe()方法将尝试使用指定的订阅名my-subscription将消费者连接到名称为my-topic的这个主题上:
- 如果订阅已经存在并绑定了其他的消费者且订阅类型是独占模式的，则subscribe方法会抛出异常。
- 如果是第一次使用的订阅名连接到主题，会自动创建订阅，每当创建新的订阅时，默认情况下它的位置被定位到Topic的末尾，即该订阅上的消费者从创建订阅后发布到Topic中的第一条消息开始消费。
- 如果通过一个之前已经存在的订阅连接到主题，则消费者将从订阅中最早未确认的消息开始消费。
3.1 同步接收消息 #
一个常用的消费模式是让消费者在while循环中监听Topic，下面的代码演示了消费者在while循环中持续监听Topic，打印接收到的消息内容，然后确认消息已被处理。如果处理逻辑失败发生了任何异常，将发送一个否定确认(negative ack)，在稍后的时间消息会重新发给消费者进行重试。
1@Test
2public void testConsumerWhile() throws PulsarClientException {
3 try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.13:6650").build()) {
4 Consumer<byte[]> consumer = client.newConsumer()
5 .topic("persistent://study/app1/topic-1")
6 .subscriptionName("sub-2")
7 .subscriptionType(SubscriptionType.Exclusive) // 订阅类型: 独占模式
8 .subscribe();
9
10 while (true) {
11 // 等待一个消息
12 Message<byte[]> msg = consumer.receive();
13 try {
14 // 处理消息
15 System.out.println("Message received: " + new String(msg.getData()));
16 // 处理完成发送确认ACK, 通知Broker消息可以被删除
17 consumer.acknowledge(msg);
18 } catch (Exception e) {
19 // 处理失败，发送否定确认(negative ack)，在稍后的时间消息会重新发给消费者进行重试
20 consumer.negativeAcknowledge(msg);
21 }
22 }
23 }
24}
3.2 异步接收消息 #
在while循环中监听Topic，并使用consumer.receive()接收方法是同步的方式。用来检索消息的receive()方法是阻塞方法，它会无限期地阻塞等待新消息到来。
这个模式在消息的数量较少且对消息从发布到消息之间的延迟不敏感的场景下可以使用。但同步不是一个好的方式，更好的方法是以异步的形式处理。
1CompletableFuture<Message> asyncMessage = consumer.receiveAsync();
异步接收操作返回包装在CompletableFuture中的Message。
3.3 使用org.apache.pulsar.client.api.MessageListener接收 #
可以使用MessageListener监听接收，在这种方式下，Pulsar Consumer会自动创建用于运行messagelistener实例的线程池。 使用MessageListener可以轻松实现多Consumer(多线程)共享订阅消费Topic中的消息。
1@Test
2public void testConsumerMessageListener() throws PulsarClientException, InterruptedException {
3 try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://192.168.2.13:6650").build()) {
4 ConsumerBuilder<byte[]> consumerBuilder = client.newConsumer()
5 .topic("persistent://study/app1/topic-1")
6 .subscriptionName("sub-4")
7 .subscriptionType(SubscriptionType.Shared) // 订阅类型: 共享模式
8 .messageListener((c, msg) -> {
9 try {
10 System.out.println(c.getConsumerName() + " received: " + new String(msg.getData()));
11 c.acknowledge(msg);
12 } catch (Exception e) {
13 c.negativeAcknowledge(msg);
14 }
15 });
16 for (int i =0;i<4;i++) {
17 consumerBuilder.consumerName("testConsumerMessageListener-" + i).subscribe();
18 }
19 Thread.sleep(TimeUnit.MINUTES.toMillis(1));
20 }
21}
3.4 批量接收 #
在创建Consumer时，可以设置Consumer的批量接收策略:
1Consumer consumer = client.newConsumer()
2 .topic("my-topic")
3 .subscriptionName("my-subscription")
4 .batchReceivePolicy(BatchReceivePolicy.builder()
5 .maxNumMessages(100)
6 .maxNumBytes(1024 * 1024)
7 .timeout(200, TimeUnit.MILLISECONDS)
8 .build())
9 .subscribe();
例如上面设置的批量接收策略为: 消息数量达到100条，消息的字节数达到1024K，等待超时达到200毫秒，三个条件满足任意一个即可。 默认批量接收策略是：
1BatchReceivePolicy.builder()
2 .maxNumMessage(-1)
3 .maxNumBytes(10 * 1024 * 1024)
4 .timeout(100, TimeUnit.MILLISECONDS)
5 .build();
根据批量接收策略，使用consumer的batchReceive()方法可以一次接收多条消息。
1Messages messages = consumer.batchReceive();
2for (Object message : messages) {
3 // do something
4}
5consumer.acknowledge(messages)
3.5 多主题订阅 #
消费者除了订阅单个主题外，还可以使用多主题订阅订阅多个主题。 要使用多主题订阅, 可以提供一个主题正则表达式(regex)或主题List。 如果通过regex选择主题, 则所有主题都必须位于同一Pulsar命名空间中。
1import org.apache.pulsar.client.api.Consumer;
2import org.apache.pulsar.client.api.PulsarClient;
3
4import java.util.Arrays;
5import java.util.List;
6import java.util.regex.Pattern;
7
8ConsumerBuilder consumerBuilder = pulsarClient.newConsumer()
9 .subscriptionName(subscription);
10
11// Subscribe to all topics in a namespace
12Pattern allTopicsInNamespace = Pattern.compile("public/default/.*");
13Consumer allTopicsConsumer = consumerBuilder
14 .topicsPattern(allTopicsInNamespace)
15 .subscribe();
16
17// Subscribe to a subsets of topics in a namespace, based on regex
18Pattern someTopicsInNamespace = Pattern.compile("public/default/foo.*");
19Consumer allTopicsConsumer = consumerBuilder
20 .topicsPattern(someTopicsInNamespace)
21 .subscribe();
在上面的示例中，消费者订阅了能够匹配主题名称正则模式的持久主题。
如果希望消费者订阅所有可以匹配主题名称模式的持久和非持久主题，需要将subscriptionTopicsMode设置为RegexSubscriptionMode.AllTopics。
1Pattern pattern = Pattern.compile("public/default/.*");
2pulsarClient.newConsumer()
3 .subscriptionName("my-sub")
4 .topicsPattern(pattern)
5 .subscriptionTopicsMode(RegexSubscriptionMode.AllTopics)
6 .subscribe();
还可以通过明确的主题列表订阅多个主题，主题列表中的主题可以跨命名空间。
1List<String> topics = Arrays.asList(
2 "topic-1",
3 "topic-2",
4 "topic-3"
5);
6
7Consumer multiTopicConsumer = consumerBuilder
8 .topics(topics)
9 .subscribe();
10
11// Alternatively:
12Consumer multiTopicConsumer = consumerBuilder
13 .topic(
14 "topic-1",
15 "topic-2",
16 "topic-3"
17 )
18 .subscribe();
3.6 死信策略 #
当消费者处理消息失败时，如果消费者给了Broker否定确认(negative ack)，或者Broker在预先设置的时间内没有收到确认ACK，Broker可以将消息重新发送给消息者。 这就相当于"重试"逻辑，如果重试成功，消息会被正常消费。但总会有重试无法成功的情况，因此不能无限的重试下去。
关于消费者处理消息时的异常处理有以下三种选择:
- 第一种方法是捕获任何异常，无论是否发生异常都简单地确认这些消息已成功处理，这实际上是忽略了处理失败的消息。这种方法只适用于业务上允许消息丢失的场景。
- 第二种方法是上面说的无限重试，消息处理成功发确认ACK，消息处理失败捕获异常时发送否定确认(negative ack)。这种方法可能会导致失败消息的无限重新传递循环，可能会引起消息堵塞，导致后边的消息无法被消费。
- 第三种方法是将有问题的消息路由到一个单独的主题，称为死信主题（Dead Letter Topic)。这样就能避免无限的重新传递循环引起消息堵塞，同时死信主题中保留的消息可以在后续进行进一步有程序自动处理或者人工检查和处理。
在创建消费者时可以为其设置死信策略，示例代码如下:
1Consumer consumer = client.newConsumer()
2 .topic("persistent://study/app1/topic-1")
3 .subscriptionName("sub5")
4 .deadLetterPolicy(DeadLetterPolicy.builder()
5 .maxRedeliverCount(10)
6 .deadLetterTopic("persistent://study/app1/dlt-1"”))
7 .subscribe();
在设置死信策略时可以设置消息的最大重发次数，也就是说消息处理失败时，重试的次数，如果超过设置的最大次数，则将会被发送到死信主题。
4.使用Pulsar Client创建消息Reader #
使用消息Reader可以由用户自己手动在Topic中定位，读取想要读取的消息。
1byte[] msgIdBytes = // Some message ID byte array
2MessageId id = MessageId.fromByteArray(msgIdBytes);
3Reader reader = pulsarClient.newReader()
4 .topic(topic)
5 .startMessageId(id) // .startMessageId(MessageId.earliest)
6 .create();
7
8while (true) {
9 Message message = reader.readNext();
10 // Process message
11}