Apache Pulsar共享订阅模式下消费者取消订阅报错问题咨询

首先明确：共享订阅模式是支持取消订阅操作的，你遇到的错误并不是因为不支持，而是操作时机或者方式不符合客户端的要求。

这个错误提示"Unconnected or shared consumer attempting to unsubscribe"，通常是下面几种情况导致的，对应的解决方式如下：

1. 确保消费者处于活跃连接状态 #

取消订阅操作必须在消费者已经成功连接集群并处于活跃状态时执行。如果你的消费者还没完成subscribe()的初始化流程，或者已经因为网络问题、异常断开了连接，此时调用unsubscribe()就会触发这个错误。

建议在执行取消订阅前，可以通过consumer.isConnected()方法（不同语言客户端方法名可能略有不同）确认连接状态，确保消费者是正常连接的。

2. 注意消费者生命周期的操作顺序 #

很多人会犯的一个错误是：先调用了consumer.close()，之后又去调用unsubscribe()。其实Pulsar客户端的close()方法内部已经自动包含了取消订阅的逻辑，关闭后消费者会进入断开状态，此时再手动取消订阅肯定会报错。

正确的流程应该是二选一：

• 先手动调用unsubscribe()，再调用close()关闭消费者；
• 直接调用close()，不需要额外执行取消订阅。

举个Java客户端的示例（其他语言逻辑类似）：

// 初始化共享订阅的消费者
Consumer<byte[]> consumer = client.newConsumer()
    .topic("your-topic")
    .subscriptionName("shared-sub-name")
    .subscriptionType(SubscriptionType.Shared)
    .subscribe();

// 业务处理逻辑...

// 方式一：先取消订阅再关闭
consumer.unsubscribe();
consumer.close();

// 方式二：直接关闭，自动处理取消订阅
// consumer.close();

3. 多线程多消费者场景的注意事项 #

因为你是多线程多消费者的架构，要注意每个消费者实例的操作线程安全：

• 不要在多个线程同时对同一个消费者实例执行取消订阅或关闭操作，这会导致状态混乱；
• 建议为每个消费者实例单独绑定生命周期管理逻辑，比如在创建消费者的线程中负责执行取消订阅和关闭操作，或者用同步锁保证操作的原子性。

4. 排查客户端版本问题 #

如果以上操作都没问题，但还是报错，可能是你使用的Pulsar客户端版本存在bug。某些较旧的版本在共享订阅的取消订阅逻辑上有缺陷，建议升级到最新的稳定版本（比如2.10+或更高的LTS版本）再试。

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