tencent cloud

//从配置文件中获取 serviceURL 接入地址、Token 密钥、Topic 全名和 Subscription 名称（均可从控制台复制）
@Value("${tdmq.serviceUrl}")
private String serviceUrl;
@Value("${tdmq.token}")
private String token;
@Value("${tdmq.topic}")
private String topic;

//声明1个 Client 对象、producer 对象
private PulsarClient pulsarClient;
private Producer<String> producer;

//在一段初始化程序中创建好客户端和生产者对象
public void init() throws Exception {
    pulsarClient = PulsarClient.builder()
            .serviceUrl(serviceUrl)
            .authentication(AuthenticationFactory.token(token))
            .build();
    producer = pulsarClient.newProducer(Schema.STRING)
            .topic(topic)
            .create();
}

//在实际生产消息的业务逻辑中直接引用，注意 Producer 通过范式声明的 Schema 类型要和传入对象匹配
public void onProduce(Producer<String> producer){
    //添加业务逻辑
    String msg = "my-message";//模拟从业务逻辑拿到消息
    try {
        //TDMQ Pulsar 版默认开启 Schema 校验, 消息对象一定需要和 producer 声明的 Schema 类型匹配
        MessageId messageId = producer.newMessage()
              .key("msgKey")
                  .value(msg)
                  .send();
        System.out.println("delivered msg " + msgId + ", value:" + value);
    } catch (PulsarClientException e) {
          System.out.println("delivered msg failed, value:" + value);
          e.printStackTrace();
    }
}

public void onProduceAsync(Producer<String> producer){
    //添加业务逻辑
    String msg = "my-asnyc-message";//模拟从业务逻辑拿到消息
    //异步发送消息，无线程阻塞，提升发送速率
    CompletableFuture<MessageId> messageIdFuture = producer.newMessage()
          .key("msgKey")
              .value(msg)
              .sendAsync();
    //通过异步回调得知消息发送成功与否
    messageIdFuture.whenComplete(((messageId, throwable) -> {
        if( null != throwable ) {
            System.out.println("delivery failed, value: " + msg );
            //此处可以添加延时重试的逻辑
        } else {
            System.out.println("delivered msg " + messageId + ", value:" + msg);
        }
    }));
}

public void destroy(){
    if (producer != null) {
        producer.close();
    }
    if (pulsarClient != null) {
        pulsarClient.close();
    }
}

import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service
public class ConsumerService implements Runnable {

    //从配置文件中获取 serviceURL 接入地址、Token 密钥、Topic 全名和 Subscription 名称（均可从控制台复制）
    @Value("${tdmq.serviceUrl}")
    private String serviceUrl;
    @Value("${tdmq.token}")
    private String token;
    @Value("${tdmq.topic}")
    private String topic;
    @Value("${tdmq.subscription}")
    private String subscription;

    private volatile boolean start = false;
    private PulsarClient pulsarClient;
    private Consumer<String> consumer;
    private static final int corePoolSize = 10;
    private static final int maximumPoolSize = 10;

    private ExecutorService executor;
    private static final Logger logger = LoggerFactory.getLogger(ConsumerService.class);

    @PostConstruct
    public void init() throws Exception {
        pulsarClient = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(AuthenticationFactory.token(token))
                .build();
        consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                //.subscriptionType(SubscriptionType.Shared)
                .subscriptionName(subscription)
                .subscribe();
        executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100),
                new ThreadPoolExecutor.AbortPolicy());
        start = true;
    }

    @PreDestroy
    public void destroy() throws Exception {
        start = false;
        if (consumer != null) {
            consumer.close();
        }
        if (pulsarClient != null) {
            pulsarClient.close();
        }
        if (executor != null) {
            executor.shutdownNow();
        }
    }

    @Override
    public void run() {
        logger.info("tdmq consumer started...");
        for (int i = 0; i < maximumPoolSize; i++) {
            executor.submit(() -> {
                while (start) {
                    try {
                        Message<String> message = consumer.receive();
                        if (message == null) {
                            continue;
                        }
                        onConsumer(message);
                    } catch (Exception e) {
                        logger.warn("tdmq consumer business error", e);
                    }
                }
            });
        }
        logger.info("tdmq consumer stopped...");
    }

    /**
     * 这里写消费业务逻辑
     *
     * @param message
     * @return return true: 消息ack  return false: 消息nack
     * @throws Exception 消息nack
     */
    private void onConsumer(Message<String> message) {
        //业务逻辑,延时类操作
        try {
            System.out.println(Thread.currentThread().getName() + " - message receive: " + message.getValue());
            Thread.sleep(1000);//模拟业务逻辑处理
            consumer.acknowledge(message);
            logger.info(Thread.currentThread().getName() + " - message processing succeed：" + message.getValue());
        } catch (Exception exception) {
            consumer.negativeAcknowledge(message);
            logger.error(Thread.currentThread().getName() + " - message processing failed：" + message.getValue());
        }
    }
}