无需SpringBoot自动配置，运行时编程注册Pulsar监听器

要实现运行时动态注册Pulsar监听器，你可以借助Spring Pulsar提供的PulsarListenerEndpointRegistry和SimplePulsarListenerEndpoint组件完成，完全不需要依赖静态的@PulsarListener注解。以下是具体实现步骤和示例：

1. 确保依赖到位 #

如果是Spring Boot项目，先引入Spring Pulsar的starter依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-pulsar</artifactId>
</dependency>

2. 编写动态注册逻辑 #

创建一个服务类，封装监听器的注册、注销逻辑，直接通过代码配置监听器的所有参数：

import org.apache.pulsar.client.api.SubscriptionType;
import org.springframework.pulsar.listener.ConsumerAwareMessageListener;
import org.springframework.pulsar.listener.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.listener.SimplePulsarListenerEndpoint;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.stereotype.Service;

@Service
public class DynamicPulsarListenerService {

    private final PulsarListenerEndpointRegistry endpointRegistry;
    private final PulsarConsumerFactory<String> consumerFactory;

    // 构造注入核心组件
    public DynamicPulsarListenerService(PulsarListenerEndpointRegistry endpointRegistry,
                                        PulsarConsumerFactory<String> consumerFactory) {
        this.endpointRegistry = endpointRegistry;
        this.consumerFactory = consumerFactory;
    }

    /**
     * 动态注册Pulsar监听器
     * @param listenerId 监听器唯一ID（用于后续注销）
     * @param topic 订阅的Topic
     * @param subscriptionName 订阅名称
     */
    public void registerListener(String listenerId, String topic, String subscriptionName) {
        // 创建可配置的监听器端点
        SimplePulsarListenerEndpoint<String> endpoint = new SimplePulsarListenerEndpoint<>();
        endpoint.setId(listenerId);
        endpoint.setTopics(topic);
        endpoint.setSubscriptionName(subscriptionName);
        // 可选：配置订阅类型（Shared/Failover/Exclusive等）
        endpoint.setSubscriptionType(SubscriptionType.Shared);

        // 设置消息消费逻辑
        endpoint.setMessageListener((ConsumerAwareMessageListener<String>) (message, consumer) -> {
            // 这里替换成你的业务处理逻辑
            System.out.println("动态监听器收到消息：" + message.getValue());
            // 如果是手动确认模式，需要手动ack消息
            // consumer.acknowledge(message);
        });

        // 注册并启动监听器容器
        endpointRegistry.registerListenerContainer(endpoint, consumerFactory, true);
    }

    /**
     * 根据ID注销监听器
     * @param listenerId 监听器唯一ID
     */
    public void unregisterListener(String listenerId) {
        var container = endpointRegistry.getListenerContainer(listenerId);
        if (container != null) {
            container.stop();
            endpointRegistry.unregisterListenerContainer(listenerId);
        }
    }
}

3. 关键组件说明 #

• PulsarListenerEndpointRegistry：负责管理所有Pulsar监听器容器，提供注册、注销、启停的全生命周期管理能力
• SimplePulsarListenerEndpoint：可动态配置的监听器端点，支持设置topic、订阅名、订阅类型、消息处理器等所有核心参数
• ConsumerAwareMessageListener：允许消费时获取到Pulsar Consumer实例，方便手动确认消息、调整消费策略等操作

4. 注意事项 #

• 监听器ID必须唯一，否则会覆盖已存在的同ID监听器
• 如果需要手动确认消息，需在PulsarConsumerFactory配置中设置acknowledgmentMode为MANUAL
• 可以根据业务配置动态调整批量消费、重试策略等参数，只需在SimplePulsarListenerEndpoint上设置对应属性
• 注销监听器时务必先停止容器再注销，避免资源泄漏

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