脉冲星消息监听器的多线程

Question

我对java message listeners和apache pulsar还很陌生。假设我维护了一个这样的监听器，

private MessageListener<byte[]> generateListener() {
MessageListener<byte[]> listener = (consumer, respMsg) -> {
String respM = new String(respMsg.getValue(), StandardCharsets.UTF_8);
logger.info(respM);
consumer.acknowledgeAsync(respMsg);
};
return listener;
}

像这样的消费者实例，

Consumer<byte[]> c = consumerBuilder.messageListener(generateListener()).topic(topicName).subscriptionName("Consumer-" + i).subscribeAsync().get();

我想知道的是，这个监听器将如何处理多个传入消息？是否会像JMS侦听器那样，在单独的线程中处理每条消息？如果是，那么我如何配置要使用的线程数？是通过使用ClientBuilder.listenerThreads()属性吗？

当维护多个消费者时，是否需要维护每个消费者各自的多个侦听器对象，即类似这样的东西-

consumerBuilder.clone().messageListener(generateListener()).topic(topicName).subscriptionName("Consumer-" + i).subscribeAsync()？

Answer 1

ClientBuilder#listenerThreads方法允许配置内部线程池的大小，该线程池将在将从该客户端创建的所有Consumers或Readers中共享和使用。

Pulsar客户端将向您保证，单个使用者的MessageListener将始终由同一线程调用，即所提供的MessageListener不需要是线程安全的。因此，是的，最好每个Consumer使用一个专用的MessageListener对象。

请注意，这也确保了订购。

因此，基本上，如果您只使用一个Consumer，那么您可以将listenerThreads保持为1(这是默认值(。

下面是一个可以用来观察行为的完整示例：

public class PulsarConsumerListenerExample {
public static void main(String[] args) throws PulsarClientException {
int numListenerThread = 2;
PulsarClient client = PulsarClient
.builder()
.serviceUrl("pulsar://localhost:6650")
.listenerThreads(numListenerThread)
.build();
final List<Consumer<?>> consumers = new ArrayList<>();
for (int i = 0; i < numListenerThread; i++) {
consumers.add(createConsumerWithLister(client, "my-topic", "my-subscription", "C" + i));
}
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
for (Consumer<?> consumer : consumers) {
try {
consumer.close();
} catch (PulsarClientException e) {
e.printStackTrace();
}
}
}));
}
private static Consumer<String> createConsumerWithLister(final PulsarClient client,
final String topic,
final String subscription,
final String consumerName) throws PulsarClientException {
return client.newConsumer(Schema.STRING)
.topic(topic)
.consumerName(consumerName)
.subscriptionName(subscription)
.subscriptionMode(SubscriptionMode.Durable)
.subscriptionType(SubscriptionType.Failover)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.messageListener((MessageListener<String>) (consumer, msg) -> {
System.out.printf(
"[%s/%s]Message received: key=%s, value=%s, topic=%s, id=%s%n",
consumerName,
Thread.currentThread().getName(),
msg.getKey(),
msg.getValue(),
msg.getTopicName(),
msg.getMessageId().toString());
consumer.acknowledgeAsync(msg);
})
.subscribe();
}
}