如何在使用Apache Pulsar作为分布式任务队列时限制并发处理任务数？

完全可以实现，以下是几种实用的实现方式：

1. 客户端配置参数直接限制 #

Pulsar客户端提供了两个核心参数来控制单个消费者的未确认消息数量：

• maxPendingMessages：单个分区内，消费者允许持有的未确认消息上限
• maxPendingMessagesAcrossPartitions：该消费者跨所有分配到的分区，允许持有的未确认消息总上限

在创建Shared订阅的消费者时，直接配置这两个参数即可。比如Java客户端示例：

Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
    .topic("task-topic")
    .subscriptionName("shared-task-sub")
    .subscriptionType(SubscriptionType.Shared)
    .maxPendingMessages(10) // 单个分区最多10条未确认消息
    .maxPendingMessagesAcrossPartitions(50) // 该消费者全局最多50条未确认消息
    .subscribe();

如果是多消费者部署的场景，要控制整个订阅的总并发数，可以通过「消费者数量 × 单个消费者的maxPendingMessagesAcrossPartitions」来估算和配置，比如总并发要100，就部署5个消费者，每个设置maxPendingMessagesAcrossPartitions=20。

2. Broker端全局/命名空间级管控 #

如果需要全局统一限制，或者防止客户端配置超限，可以在Broker层面设置maxUnackedMessagesPerConsumer参数，这个配置优先级高于客户端参数。

可以通过命令行在命名空间级别设置：

bin/pulsar-admin namespaces set-max-unacked-messages-per-consumer your-namespace --value 20

该配置会限制命名空间下所有消费者的未确认消息数上限，超过后Broker会停止向消费者推送新消息，直到有消息被确认。

3. 自定义消费逻辑实现精细控制 #

如果需要动态调整并发数，或者结合业务逻辑做更灵活的控制，可以在消费者端维护一个计数器，手动控制消息接收的启停：

AtomicInteger inFlightTasks = new AtomicInteger(0);
final int MAX_CONCURRENCY = 15;

consumer.messageListener((consumer, msg) -> {
    // 检查当前并发数，达到上限则暂停消费
    if (inFlightTasks.incrementAndGet() > MAX_CONCURRENCY) {
        consumer.pause();
    }

    // 执行任务处理逻辑
    processTask(msg);

    // 确认消息后更新计数器，当低于阈值时恢复消费
    consumer.acknowledge(msg);
    if (inFlightTasks.decrementAndGet() <= MAX_CONCURRENCY * 0.8) {
        consumer.resume();
    }
});

这种方式需要注意设置合理的恢复阈值，避免频繁暂停/恢复导致的性能波动，同时要处理好异常场景（比如任务处理失败时的计数器回退）。

内容的提问来源于stack exchange，提问作者Marco Stahl