学会Pulsar Consumer的使用方式

学会Pulsar Consumer的使用方式

目录1、使用前准备2、PulsarClient3、Producer4、Consumer4.1 第一次使用：4.2 第二次使用：4http://.3 第三次使用：4.4 第四次使用：4.5 第五次使用：重试机制源码分析4.6 第六次使用

1、使用前准备

引入依赖：

org.apache.pulsar

pulsar-client

2.6.1

2、PulsarClient

在尝试使用Producer和Consumer前，我们先讲一下Pulsar客户端，因为不管是Producer还是Consumer，都是依靠PulsarClient来创建的：

/**

* Pulsar工具类

* @author winfun

**/

public class PulsarUtils {

/**

* 根据serviceUrl创建PulsarClient

* @param serviceUrl 服务地址

* @return 客户端

* @throws PulsarClientException 异常

*/

public static PulsarClient createPulsarClient(String serviceUrl) throws PulsarClientException {

return PulsarClient.builder()

.serviceUrl(serviceUrl)

.build();

}

}

我们这里简单使用，只借用ServiceUrl创建客户端，其实还有很多比较重要的参数，下面稍微列举一下：

ioThreads：Set the number of threads to be used for handling connections to brokers (default: 1 thread)

listenerThreads：Set the number of threads to be used for message listeners (default: 1 thread). 一条线程默认只为一个消费者服务

enableTcpNoDelay：No-delay features make sure packets are sent out on the network as soon as possible

…

3、Producer

Producer这里我们也先简单使用，只负责往指定Topic发送消息，其他功能不用，例如异步发送、延时发送等

/**

* 初次使用Pulsar生产者，无任何封装

* @author winfun

**/

public class FirstProducerDemo {

public static void main(String[] args) throws PulsarClientException {

PulsarClient client = PulsarClient.builder()

.serviceUrl("pulsar://127.0.0.1:6650")

.build();

ProducerBuilder productBuilder = client.newProducer(Schema.STRING).topic("winfun/study/test-topic")

.blockIfQueueFull(Boolean.TRUE).batchingMaxMessages(100).enableBatching(Boolean.TRUE).sendTimeout(3, TimeUnit.SECONDS);

Producer producer = productBuilder.create();

for (int i = 0; i < 100; i++) {

producer.send("hello"+i);;

}

producer.close();

}

}

4、Consumer

下面我们将比较详细地介绍消费者的使用方式，因为这里能拓展的东西稍微多一点，下面开始使用旅程。

4.1 第一次使用：

我们利用PulsarClient创建Consumer；接着在死循环中利用Consumer#receive方法接收消息然后进行消费。

/**

* 初次使用Pulsar消费者，无任何封装

* @author winfun

**/

@Slf4j

public class FirstConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

PulsarClient client = PulsarClient.builder()

.serviceUrl("pulsar://127.0.0.1:6650")

.build();

/**

* The subscribe method will auto subscribe the consumer to the specified topic and subscription.

* One way to make the consumer listen on the topic is to set up a while loop.

* In this example loop, the consumer listens for messages, prints the contents of any received message, and then acknowledges that the message has been processed.

* If the processing logic fails, you can use negative acknowledgement to redeliver the message later.

*/

Consumer consumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-topic")

.subscriptionName("my-subscription")

.ackTimeout(10, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Exclusive)

http://.subscribe();

// 死循环接收

while (true){

Message message = consumer.receive();

String msgContent = message.getValue();

log.info("接收到消息: {}",msgContent);

consumer.acknowledge(message);

}

}

}

4.2 第二次使用：

上面我们可以看到，我们是利用死循环来保证及时消费，但是这样会导致主线程；所以下面我们可以使用Pulsar提供的MessageListener，即监听器，当消息来了，会回调监听器指定的方法，从而避免阻塞主线程。

/**

* 使用MessageListener，避免死循环代码&阻塞主线程

* @author winfun

**/

@Slf4j

public class SecondConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

PulsarClient client = PulsarUtils.createPulsarClient("pulsar://127.0.0.1:6650");

/**

* If you don't want to block your main thread and rather listen constantly for new messages, consider using a MessageListener.

*

*/

Consumer consumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-topic")

.subscriptionName("my-subscription")

.ackTimeout(10, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Exclusive)

.messageListener((MessageListener) (consumer1, msg) -> {

/**

* 当接收到一个新的消息，就会回调 MessageListener的receive方法。

* 消息将会保证按顺序投放到单个消费者的同一个线程，因此可以保证顺序消费

* 除非应用程序或broker崩溃，否则只会为每条消息调用此方法一次

* 应用程序负责调用消费者的确认方法来确认消息已经被消费

* 应用程序负责处理消费消息时可能出现的异常

*/

log.info("接收到消息:{}",msg.getValue());

try {

consumer1.acknowledge(msg);

} catch (PulsarClientException e) {

e.printStackTrace();

}

}).subscribe();

}

}

4.3 第三次使用：

上面利用监听器来解决死循环代码和阻塞主线程问题；但是我们可以发现，每次消费都是单线程，当一个消息消费完才能进行下一个消息的消费，这样会导致消费效率非常的低；

如果如果追求高吞吐量，不在乎消息消费的顺序性，那么我们可以接入线程池；一有消息来就丢进线程池中，这样不但可以支持异步消费，还能保证消费的效率非常的高。

/**

* MessageListener 内使用线程池进行异步消费

* @author winfun

**/

@Slf4j

public class ThirdConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

Executor executor = new ThreadPoolExecutor(

10,

10,

10,

TimeUnit.SECONDS,

new ArrayBlockingQueue<>(100)

);

PulsarClient client = PulsarUtils.createPulsarClient("pulsar://127.0.0.1:6650");

/**

* If you don't want to block your main thread and rather listen constantly for new messages, consider using a MessageListener.

*

*/

Consumer consumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-topic")

.subscriptionName("my-subscription")

.ackTimeout(10, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Exclusive)

.messageListener((MessageListener) (consumer1, msg) -> {

/**

* MessageListener还是保证了接收的顺序性

* 但是利用线程池进行异步消费后不能保证消费顺序性

*/

executor.execute(() -> handleMsg(consumer1, msg));

}).subscribe();

}

/**

* 线程池异步处理

* @param consumer 消费者

* @param msg 消息

*/

public static void handleMsg(Consumer consumer, Message msg){

ThreadUtil.sleep(RandomUtil.randomInt(3),TimeUnit.SECONDS);

log.info("接收到消息:{}",msg.getValue());

try {

consumer.acknowledge(msg);

} catch (PulsarClientException e) {

e.printStackTrace();

}

}

}

4.4 第四次使用：

我们可以发现，在上面的三个例子中，如果在调用Consumer#acknowledge方法前，因为代码问题导致抛异常了，我们是没有做处理的，那么会导致消费者会一直重试没有被确认的消息。

那么我们此时需要接入Pulsar提供的死信队列：当Consumer消费消息时抛异常，并达到一定的重试次数，则将消息丢入死信队列；但需要注意的是，单独使用死信队列，Consumer的订阅类型需要是 Shared/Key_Shared；否则不会生效。

/**

* 超过最大重试次数，进入死信队列

* @author: winfun

**/

@Slf4j

public class FourthConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

/**

* 如果指定了死信队列策略，但是没指定死信队列

* 死信队列：String.format("%s-%s-DLQ", topic, this.subscription)

* 这里的this.subscription为上面指定的 subscriptionName。

*

* 一般在生产环境，会将pulsar的自动创建topic功能给关闭掉，所以上线前，记得先提工单创建指定的死信队列。

*

* 重点信息：

* 如果是单单使用死信队列，subscriptionType为 Shared/Key_Shared，否则死信队列不生效。

*/

PulsarClient client = PulsarUtils.createPulsarClient("pulsar://127.0.0.1:6650");

Consumer consumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-topic")

.subscriptionName("my-subscription")

.receiverQueueSize(100)

.ackTimeout(1, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Key_Shared)

.negativeAckRedeliveryDelay(1,TimeUnit.SECONDS)

.deadLetterPolicy(DeadLetterPolicy.builder()

//可以指定最大重试次数，最大重试三次后，进入到死信队列

.maxRedeliverCount(3)

//可以指定死信队列

.deadLetterTopic("winfun/study/test-topic-dlq3")

.build())

.messageListener((MessageListener) (consumer1, msg) -> {

log.info("接收到队列「{}」消息:{}",msg.getTopicName(),msg.getValue());

if (msg.getValue().equals("hello3")) {

throw new RuntimeException("hello3消息消费失败！");

}else {

try {

consumer1.acknowledge(msg);

} catch (PulsarClientException e) {

e.printStackTrace();

}

}

}).subscribe();

}

}

4.5 第五次使用：

死信队列一般是不做消费的，我们会关注死信队列的情况，从而作出下一步的动作。

而且，一般做消息重试，我们不希望在原Topic中做重试，这样会影响原有消息的消费进度。

那么我们可以同时使用重试队列和死信队列。

当代码抛出异常时，我们可以捕获住，然后调用Consumer#reconsumeLater方法，将消息丢入重试队列；当消息重试指定次数后还无法正常完成消费，即会将消息丢入死信队列。

/**

* 重试队列

* @author winfun

**/

@Slf4j

public class FifthConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

PulsarClient client = PulsarUtils.createPulsarClient("pulsar://127.0.0.1:6650");

/**

* 注意点：

* 1、使用死信策略，但是没有指定重试topic和死信topic名称

* 死信队列：String.format("%s-%s-DLQ", topic, this.subscription)

* 重试队列：String.format("%s-%s-RETRY", topic, this.subscription)

* 这里的this.subscription为上面指定的 subscriptionName。

*

* 2、是否限制订阅类型

* 同时开启重试队列和死信队列，不限制subscriptionType只能为Shared/Key_Shared；

* 如果只是单独使用死信队列，需要限制subscriptionType为Shared

*

* 3、重试原理

* 如果使用重试队列，需要保证 enableRetry 是开启的，否则调用 reconsumeLater 方法时会抛异常：org.apache.pulsar.client.api.PulsarClientException: reconsumeLater method not support!

* 如果配置了重试队列，consumer会同时监听原topic和重试topic，consumer的实现类对应是：MultiTopicsConsumerImpl

* 如果消费消息时调用了 reconsumeLater 方法，会将此消息丢进重试topic

* 如果在重试topic重试maxRedeliverCount次后都无法正确ack消息，即将消息丢到死信队列。

* 死信队列需要另起Consumer进行监听消费。

*

* 4、直接抛异常

* 如果我们不是业务层面上调用 reconsumeLater 方法来进行重试，而是代码层面抛异常了，如果subscriptionType不为Shared/Key_Shared，消息无法进入重试队列和死信队列，是当前消费者无限在原topic进行消费。

* 而如果如果subscriptionType为Shared/Key_Shared，则消息进行maxRedeliverCount次消费后，会直接进入到死信队列，此时不会用到重试队列。

* 因此，重试队列是仅仅针对 reconsumeLater 方法的，而不针对异常的重试。

*/

Consumer consumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-retry-topic")

.subscriptionName("my-subscription")

.receiverQueueSize(100)

.ackTimeout(1, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Exclusive)

.negativeAckRedeliveryDelay(1,TimeUnit.SECONDS)

.enableRetry(true)

.deadLetterPolicy(DeadLetterPolicy.builder()

//可以指定最大重试次数，最大重试三次后，进入到死信队列

.maxRedeliverCount(3)

.retryLetterTopic("winfun/study/test-retry-topic-retry")

//可以指定死信队列

.deadLetterTopic("winfun/study/test-retry-topic-dlq")

.build())

.messageListener((MessageListener) (consumer1, msg) -> {

log.info("接收到队列「{}」消息:{}",msg.getTopicName(),msg.getValue());

if (msg.getValue().equals("hello3")) {

try {

consumer1.reconsumeLater(msg,1,TimeUnit.SECONDS);

} catch (PulsarClientException e) {

e.printStackTrace();

}

//throw new RuntimeException("hello3消息消费失败！");

}else {

try {

consumer1.acknowledge(msg);

} catch (PulsarClientException e) {

e.printStackTrace();

}

}

}).subscribe();

}

}

重试机制源码分析

关于重试机制，其实是比较有意思的，下面我们会简单分析一下源码。

1.判断是否开启重试机制，如果没有开启重试机制，则直接抛异常

public void reconsumeLater(Message> message, long delayTime, TimeUnit unit) throws PulsarClientException {

// 如果没开启重试机制，直接抛异常

if (!this.conf.isRetryEnable()) {

throw new PulsarClientException("reconsumeLater method not support!");

} else {

try {

// 当然了，reconsumeLaterAsync里面也会判断是否开启重试机制

this.reconsumeLaterAsync(message, delayTime, unit).get();

} catch (Exception var7) {

Throwable t = var7.getCause();

if (t instanceof PulsarClientException) {

throw (PulsarClientException)t;

} else {

throw new PulsarClientException(t);

}

}

}

}

还有我们可以发现，pulsar很多方法是支持同步和异步的，而同步就是直接调用异步方法，再后调用get()方法进行同步阻塞等待即可。

2.调用 reconsumeLaterAsunc 方法，接着调用 get() 进行同步阻塞等待结果

public CompletableFuture reconsumeLaterAsync(Message> message, long delayTime, TimeUnit unit) {

if (!this.conf.isRetryEnable()) {

return FutureUtil.failedFuture(new PulsarClientException("reconsumeLater method not support!"));

} else {

try {

return this.doReconsumeLater(message, AckType.Individual, Collections.emptyMap(), delayTime, unit);

} catch (NullPointerException var6) {

return FutureUtil.failedFuture(new InvalidMessageException(var6.getMessage()));

}

}

}

3.调用 doReconsumeLater 方法

我们知道，在 Pulsar 的 Consumer 中，可以支持多 Topic 监听，而如果我们加入了重试机制，默认是同个 Consumer 同时监听原队列和重试队列，所以 Consumer 接口的实现此时为 MultiTopicsConsumerImpl，而不是 ComsumerImpl。

那我们看看 MultiConsumerImpl 的 doReconsumeLater 是如何进行重新消费的：

protected CompletableFuture doReconsumeLater(Message> message, AckType ackType, Map properties, long delayTime, TimeUnit unit) {

MessageId messageId = message.getMessageId();

Preconditions.checkArgument(messageId instanceof TopicMessageIdImpl);

TopicMessageIdImpl topicMessageId = (TopicMessageIdImpl)messageId;

if (this.getState() != State.Ready) {

return FutureUtil.failedFuture(new PulsarClientException("Consumer already closed"));

} else {

MessageId innerId;

if (ackType == AckType.Cumulative) {

Consumer individualConsumer = (Consumer)this.consumers.get(topicMessageId.getTopicPartitionName());

if (individualConsumer != null) {

innerId = topicMessageId.getInnerMessageId();

return individualConsumer.reconsumeLaterCumulativeAsync(message, delayTime, unit);

} else {

return FutureUtil.failedFuture(new NotConnectedException());

}

} else {

ConsumerImpl consumer = (ConsumerImpl)this.consumers.get(topicMessageId.getTopicPartitionName());

innerId = topicMessageId.getInnerMessageId();

return consumer.doReconsumeLater(message, ackType, properties, delayTime, unit).thenRun(() -> {

this.unAckedMessageTracker.remove(topicMessageId);

});

}

}

}

首先判断客户端是否为准备状态

接着判断 AckType 是累计的还是单独的，如果是累计的话，subscriptionType 一定要是 exclusive

不管是累计还是单独的，最后都是调用 ConsumerImpl 的 doReconsumerLater 方法

protected CompletableFuture doReconsumeLater(Message> message, AckType ackType, Map properties, long delayTime, TimeUnit unit) {

MessageId messageId = message.getMessageId();

if (messageId instanceof TopicMessageIdImpl) {

messageId = ((TopicMessageIdImpl)messageId).getInFLJoXpjnerMessageId();

}

Preconditions.checkArgument(messageId instanceof MessageIdImpl);

if (this.getState() != State.Ready && this.getState() != State.Connecting) {

this.stats.incrementNumAcksFailed();

PulsarClientException exception = new PulsarClientException("Consumer not ready. State: " + this.getState());

if (AckType.Individual.equals(ackType)) {

this.onAcknowledge(messageId, exception);

} else if (AckType.Cumulative.equals(ackType)) {

this.onAcknowledgeCumulative(messageId, exception);

}

return FutureUtil.failedFuture(exception);

} else {

if (delayTime < 0L) {

delayTime = 0L;

}

// 如果 retryLetterProducer 为null，则尝试创建

if (this.retryLetterProducer == null) {

try {

this.createProducerLock.writeLock().lock();

if (this.retryLetterProducer == null) {

this.retryLetterProducer = this.client.newProducer(this.schema).topic(this.deadLetterPolicy.getRetryLetterTopic()).enableBatching(false).blockIfQueueFull(false).create();

}

} catch (Exception var28) {

log.error("Create retry letter producer exception with topic: {}", this.deadLetterPolicy.getRetryLetterTopic(), var28);

} finally {

this.createProducerLock.writeLock().unlock();

}

}

// 如果 retryLetterProcuder 不为空，则尝试将消息丢进重试队列中

if (this.retryLetterProducer != null) {

try {

MessageImpl retryMessage = null;

String originMessageIdStr = null;

String originTopicNameStr = null;

if (message instanceof TopicMessageImpl) {

retryMessage = (MessageImpl)((TopicMessageImpl)message).getMessage();

originMessageIdStr = ((TopicMessageIdImpl)message.getMessageId()).getInnerMessageId().toString();

originTopicNameStr = ((TopicMessageIdImpl)message.getMessageId()).getTopicName();

} else if (message instanceof MessageImpl) {

retryMessage = (MessageImpl)message;

originMessageIdStr = ((MessageImpl)message).getMessageId().toString();

originTopicNameStr = ((MessageImpl)message).getTopicName();

}

SortedMap propertiesMap = new TreeMap();

int reconsumetimes = 1;

if (message.getProperties() != null) {

propertiesMap.putAll(message.getProperties());

}

// 如果包含 RECONSUMETIMES，则最递增

if (propertiesMap.containsKey("RECONSUMETIMES")) {

reconsumetimes = Integer.valueOf((String)propertiesMap.get("RECONSUMETIMES"));

++reconsumetimes;

// 否则先加入「原始队列」和「原始messageId」信息

} else {

propertiesMap.put("REAL_TOPIC", originTopicNameStr);

propertiesMap.put("ORIGIN_MESSAGE_IDY_TIME", originMessageIdStr);

}

// 加入重试次数信息

propertiesMap.put("RECONSUMETIMES", String.valueOf(reconsumetimes));

// 加入延时时间信息

propertiesMap.put("DELAY_TIME", String.valueOf(unit.toMillis(delayTime)));

TypedMessageBuilder typedMessageBuilderNew;

// 判断是否超过最大重试次数，如果还未超过，则重新投放到重试队列

if (reconsumetimes <= this.deadLetterPolicy.getMaxRedeliverCount()) {

typedMessageBuilderNew = this.retryLetterProducer.newMessage().value(retryMessage.getValue()).properties(propertiesMap);

if (delayTime > 0L) {

typedMessageBuilderNew.deliverAfter(delayTime, unit);

}

if (message.hasKey()) {

typedMessageBuilderNew.key(message.getKey());

}

// 发送延时消息

typedMessageBuilderNew.send();

// 确认当前消息

return this.doAcknowledge(messageId, ackType, properties, (TransactionImpl)null);

}

// 先忽略

this.processPossibleToDLQ((MessageIdImpl)messageId);

// 判断 deadLetterProducer 是否为null，如果为null，尝试创建

if (this.deadLetterProducer == null) {

try {

if (this.deadLetterProducer == null) {

this.createProducerLock.writeLock().lock();

this.deadLetterProducer = this.client.newProducer(this.schema).topic(this.deadLetterPolicy.getDeadLetterTopic()).blockIfQueueFull(false).create();

}

} catch (Exception var25) {

log.error("Create dead letter producer exception with topic: {}", this.deadLetterPolicy.getDeadLetterTopic(), var25);

} finally {

this.createProducerLock.writeLock().unlock();

}

}

// 如果 deadLetterProducer 不为null

if (this.deadLetterProducer != null) {

// 加入「原始队列」信息

propertiesMap.put("REAL_TOPIC", originTopicNameStr);

// 加入「原始MessageId」信息

propertiesMap.put("ORIGIN_MESSAGE_IDY_TIME", originMessageIdStr);

typedMessageBuilderNew = this.deadLetterProducer.newMessage().value(retryMessage.getValue()).properties(propertiesMap);

// 将消息内容发往死信队列中

typedMessageBuilderNew.send();

// 确认当前消息

return this.doAcknowledge(messageId, ackType, properties, (TransactionImpl)null);

}

} catch (Exception var27) {

log.error("Send to retry letter topic exception with topic: {}, messageId: {}", new Object[]{this.deadLetterProducer.getTopic(), messageId, var27});

Set messageIds = new HashSet();

messageIds.add(messageId);

this.unAckedMessageTracker.remove(messageId);

this.redeliverUnacknowledgedMessages(messageIds);

}

}

return CompletableFuture.completedFuture((Object)null);

}

}

分析了一波，我们可以看到和上面代码的注释描述的基本一致。

4.6 第六次使用

上面我们提到，当Consumer指定了重试队列，Consumer会同时监听原Topic和重试Topic，那么如果我们想多个Consumer消费重试Topic时，需要将Consumer的订阅类型指定为 Shared/Key_Shared，让重试队列支持多Consumer监听消费，提升重试队列的消费效率。

/**

* 重试队列-Shared

* @author winfun

**/

@Slf4j

public class SixthConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

PulsarClient client = PulsarUtils.createPulsarClient("pulsar://127.0.0.1:6650");

/**

* 因为如果指定了重试策略，Consumer会同时监听「原队列」和「重试队列」

* 即如果我们想「重试队列」可以让多个 Consumer 监听，从而提高消费能力，那么 Consumer 需指定为 Shared 模式。

*/

Consumer consumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-retry-topic")

.subscriptionName("my-subscription")

.receiverQueueSize(100)

.ackTimeout(1, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Shared)

.negativeAckRedeliveryDelay(1,TimeUnit.SECONDS)

.enableRetry(true)

.deadLetterPolicy(DeadLetterPolicy.builder()

//可以指定最大重试次数，最大重试三次后，进入到死信队列

.maxRedeliverCount(3)

.retryLetterTopic("winfun/study/test-retry-topic-retry")

//可以指定死信队列

.deadLetterTopic("winfun/study/test-retry-topic-dlq")

.build())

.messageListener((MessageListener) (consumer1, msg) -> {

log.info("接收到队列「{}」消息:{}",msg.getTopicName(),msg.getValue());

if (msg.getValue().contains("1") || msg.getValue().contains("2") || msg.getValue().contains("3")) {

try {

consumer1.reconsumeLater(msg,1,TimeUnit.SECONDS);

} catch (PulsarClientException e) {

e.printStackTrace();

}

//throw new RuntimeException("hello3消息消费失败！");

}else {

try {

consumer1.acknowledge(msg);

} catch (PulsarClientException e) {

e.printStackTrace();

}

}

}).subscribe();

}

}

/**

* 监听重试队列-Shared订阅模式

* @author winfun

**/

@Slf4j

public class RetryConsumerDemo {

public static void main(String[] args) throws PulsarClientException {

PulsarClient client = PulsarUtils.createPulsarClient("pulsar://127.0.0.1:6650");

Consumer deadLetterConsumer = client.newConsumer(Schema.STRING)

.topic("winfun/study/test-retry-topic-retry")

.subscriptionName("my-subscription2")

.receiverQueueSize(100)

.ackTimeout(1, TimeUnit.SECONDS)

.subscriptionType(SubscriptionType.Shared)

.messageListener((MessageListener) (consumer1, msg) -> {

log.info("接收到队列「{}」消息:{}",msg.getTopicName(),msg.getValue());

try {

consumer1.acknowledge(msg);

} catch (PulsarClientException e) {

e.printStackTrace();

}

}).subscribe();

}

}

到此，我们已经将Consmuer的几种使用方式都尝试了一遍，可以说基本包含了常用的操作；但是我们可以发现，如果我们每次新建一个Consumer都需要写一堆同样的代码，那其实挺麻烦的，又不好看；并且，现在我们大部分项目都是基于 SpringBoot 来做的，而 SpringBoot 也没有一个比较大众的Starter。

所以接下来的计划就是，自己写一个编写一个关于Pulsar的SpringBoot Starter，这个组件不会特别复杂，但是会支持 Producer 和 Cousnmer 的自动配置，并且支持 Consumer 上面提到的几个点：MessageListener 监听、线程池异步并发消费、重试机制等。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。