java - 脉冲星消息监听器的多线程
问题描述
我对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()
?
解决方案
该ClientBuilder#listenerThreads
方法允许配置内部线程池的大小,该内部线程池将在所有人之间共享和使用,Consumers
或者Readers
将从该客户端创建。
Pulsar Client 将保证MessageListener
单个消费者的 a 将始终由同一个线程调用,即提供的MessageListener
不需要是线程安全的。因此,是的,最好使用MessageListener
每个Consumer
.
请注意,这也可以确保订购。
所以基本上,如果你只使用一个,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();
}
}
推荐阅读
- javascript - Javascript 函数 .show() 不使用 HTML 代码 - 在页面搜索上
- kendo-ui - 剑道 UI 下拉列表不显示正确的值
- ansible - Ansible yaml 锚点和 jinja2 模板
- windows - AzureAD 加入的计算机无法访问 Windows DNS 上的 DNS 主机 (A) 记录
- javascript - 自定义 onChange 事件
- docker - 如何使用 CLion 在 Docker 容器中编码/运行程序?
- entity-framework-core - Entity Framework Core Insert 不遵守默认值
- reactjs - 创建可用于 Vue 和 React 应用程序的动态页眉和页脚的最佳方法是什么?
- elasticsearch - serilog 解构 - 混合类型
- python - 递归函数:用python递归