使用 pulsarMQ 进行收发消息

最近在做对接 pulsarMQ 进行收发消息的任务。目的是在 A 服务中，生产消息并发送到 pulsarMQ，在 B 服务中消费消息，然后根据消息内容做一些业务处理。于是，我便对腾讯云的 tdmq (pulsar 版) 进行了学习。

用浏览器打开 腾讯云tdmq-pulsar

查看 tcp协议，java sdk。
按照官方文档，依次找到自己的 serviceUrl，token，topic （注意：tcp java sdk方式，topic需要以 persistent:// 开头）

maven项目导入依赖

     org.apache.pulsar
     pulsar-client
     2.7.2

创建pulsarMQ配置类

@Configuration
public class PulsarMQConfig {

    @Bean
    @ConditionalOnProperty(value = "demoProducer.enable", havingValue = "1")
    public DemoProducer demoProducer() {
        return new DemoProducer();
    }

    @Bean
    @ConditionalOnProperty(value = "demoConsumer.enable", havingValue = "1")
    public DemoConsumer demoConsumer() {
        return new DemoConsumer();
    }
}

创建生产者

public class DemoProducer {
    
    private static final Logger log = LoggerFactory.getLogger(DemoProducer.class);

    
    @Value("${pulsar.service-url}")
    private String serviceUrl;

    
    @Value("${pulsar.token-auth-value}")
    private String token;

    @Value("${pulsar.topic}")
    private String topic;

    private PulsarClient pulsarClient;

    private Producer producer;

    @PostConstruct
    public void init() throws PulsarClientException {
        try {
            pulsarClient = PulsarClient.builder()
                    // 服务接入地址
                    .serviceUrl(serviceUrl)
                    // 授权角色密钥
                    .authentication(AuthenticationFactory.token(token))
                    .build();
            log.info(">> pulsar client created.");
            log.info("topic: {}", topic);
            
            producer = pulsarClient.newProducer(Schema.JSON(DemoMessage.class))
                    .topic(topic)
                    .blockIfQueueFull(Boolean.TRUE)
                    .create();
            log.info("create producer success");
        } catch (PulsarClientException e) {
            log.error("create producer error", e);
            close();
            throw e;
        }
    }

    public void asyncSendMessage(DemoMessage msg, BiConsumer action) {
        // 异步发送消息
        CompletableFuture completableFuture = producer.newMessage().value(msg).sendAsync();
        // 通过异步回调得知消息发送成功与否
        completableFuture.whenComplete(action);
    }


    @PreDestroy
    public void close() {
        try {
            if (Objects.nonNull(producer)) {
                producer.close();
            }
            if (Objects.nonNull(pulsarClient)) {
                pulsarClient.close();
            }
            log.info("close producer done.");
        } catch (PulsarClientException exception) {
            log.error("close producer error", exception);
        }
    }
}

创建消费者

public class DemoConsumer {
    private static final Logger logger = LoggerFactory.getLogger(DemoConsumer.class);

    
    @Value("${pulsar.service-url}")
    private String serviceUrl;

    
    @Value("${pulsar.token-auth-value}")
    private String token;

    @Value("${pulsar.topic}")
    private String topic;

    @Value("sub")
    private String subscribeName;

    private PulsarClient pulsarClient;

    private Consumer consumer;

    private ExecutorService executor = new ThreadPoolExecutor(
            10,
            10,
            60,
            TimeUnit.MINUTES,
            new LinkedBlockingDeque<>(100),
            new ThreadFactoryBuilder().setNameFormat("DemoConsumer-thread-pool-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    @PostConstruct
    public void init() throws PulsarClientException {
        try {
            pulsarClient = PulsarClient.builder()
                    // 服务接入地址
                    .serviceUrl(serviceUrl)
                    // 授权角色密钥
                    .authentication(AuthenticationFactory.token(token))
                    .ioThreads(10)
                    .listenerThreads(10)
                    .build();
            logger.info("serviceUrl: {}", serviceUrl);
            logger.info(">> pulsar client created.");

            logger.info("topic: {},subscribeName: {}", topic, subscribeName);
            consumer = pulsarClient.newConsumer(Schema.JSON(DemoMessage.class))
                    .topic(topic)
                    .subscriptionName(subscribeName)
                    // 10分钟没ack，重传
                    .ackTimeout(10, TimeUnit.MINUTES)
                    // 共享模式
                    .subscriptionType(SubscriptionType.Shared)
                    // 配置从最早开始消费，否则可能会消费不到历史消息
                    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                    .messageListener((consumer, message) -> {
                        executor.execute(() -> {
                            try {
                                process(message);
                            } catch (PulsarClientException e) {
                                logger.error("pulsar client error", e);
                            }
                        });
                    })
                    .subscribe();
            logger.info("create consumer success");
        } catch (PulsarClientException exception) {
            logger.error("create consumer error", exception);
            close();
            throw exception;
        }
    }

    private void process(Message message) throws PulsarClientException {
    // do something  
    }

    @PreDestroy
    public void close() {
        try {
            if (executor != null) {
                executor.shutdownNow();
            }
            if (Objects.nonNull(consumer)) {
                consumer.close();
            }
            if (pulsarClient != null) {
                pulsarClient.close();
            }
            logger.info("close consumer done.");
        } catch (PulsarClientException exception) {
            logger.error("close consumer error", exception);
        }
    }
}

• 腾讯云消息队列Pulsar版-javaSDK
• Pulsar-java-client官方文档