ClientBuilder#listenerThreads
方法允许配置一个内部线程池的大小,该线程池将在从该客户端创建的所有Consumers
或Readers
之间共享并使用。
Pulsar Client将为单个消费者的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();
}
}