基于腾讯云tdmq消息队列封装SpringBootStarter(一)

基于腾讯云tdmq消息队列封装SpringBootStarter(一)

杰子学编程 104 2022-05-31

基于腾讯云tdmq消息队列封装SpringBootStarter(一)

一、环境准备

1.1 注册腾讯云TDMQ

创建tdmq集群

创建完成后记录下集群ID(clusterId);

1.2 创建命名空间

创建好集群后,在命名空间中新建命名空间,命名空间名称可以根据实际业务场景进行区分,比如这里创建可以根据测试环境、预发布环境、生产环境等进行区分创建。

新建命名空间

1.3、创建好命名空间后,新建个topic主题。

创建topic

以上信息创建好后,我们在集群中可以看到集群的访问地址,如下:

查看接入地址

在创建tdmq集群时我们需要申请外网访问,这个需要找腾讯的客服开通。

至此,我们开发的基础环境已经准备完成。

二、编写生产者、消费者代码

2.1、创建工程

在idea中新建个工程,工程名称为spring-boot-starter-tdmq

创建工程

工程名称和包路径可以根据实际情况进行自定义。

2.2、引入相关依赖包

<!-- spring-boot-autoconfigure -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<!--tdmq 核心依赖 2.9.1 -->
<dependency>
    <groupId>org.apache.pulsar</groupId>
    <artifactId>pulsar-client</artifactId>
    <version>${pulsar.version}</version>
</dependency>

这里使用的是腾讯云tdmq-pulsar版,这里需要引入pulsar-client.

2.3、创建生产者

首先我们在项目中创建一个config的包路径,新建一个tdmq的配置类TdmqProperties

@Data
@ConfigurationProperties(prefix = "tdmq")
public class TdmqProperties {
    /**
     * 启用
     */
    private boolean enable = false;
    /**
     * 服务地址
     */
    private String serviceUrl;
    /**
     * token
     */
    private String token;
    /**
     * 集群ID
     */
    private String clusterId;
    /**
     * 命名空间ID
     */
    private String environmentId;
}

这里主要配置下tdmq集群相关的信息。我们先将配置信息写在配置类中,后续优化我们在使用外部配置进行初始化操作。

创建PulsarClient对象,生产者消费者都是基于PulsarClient对象进行创建。

这里我们通过SpringBoot的自动装配功能来装配PulsarClient

这里我们创建一个自动化配置类TdmqAutoConfiguration。内容如下:

@Data
@EnableConfigurationProperties({TdmqProperties.class})
public class TdmqAutoConfiguration {
    /**
     * Pulsar 客户端
     * 推荐一个进程一个实例
     *
     * @return {@link TdmqAutoConfiguration}
     */
    @Bean
    @ConditionalOnMissingBean(PulsarClient.class)
    @ConditionalOnProperty(name = "tdmq.enable", havingValue = "true")
    public PulsarClient pulsarClient(TdmqProperties mqProperties) throws PulsarClientException {
        return PulsarClient.builder()
                .serviceUrl(mqProperties.getServiceUrl())
                .authentication(AuthenticationFactory.token(mqProperties.getToken()))
                .build();
    }
}

这里增加了两个条件注解@ConditionalOnProperty@ConditionalOnMissingBean(PulsarClient.class)这里分别表示如果没有找到PulsarClient对象和配置文件中启用了tdmq功能,我们才实例化PulsarClient对象。

有了PulsarClient对象,我们可以继续编写生产者代码啦。

在工程中创建producer包路径,并在该路径下创建TdmqProucer.class,内容如下:

@Service
public class TdmqProucer {
    @Autowired
    private PulsarClient pulsarClient;

    /**
     * 发送消息
     *
     * @param message 消息内容
     * @return 消息ID
     * @throws PulsarClientException
     */
    public MessageId sendMsg(String message) throws PulsarClientException {
        Producer<byte[]> producer = pulsarClient.newProducer().topic("test").create();
        MessageId messageId = producer.newMessage().value(message.getBytes()).send();
        producer.close();
        return messageId;
    }
}

2.4、创建消息消费者

在工程中创建consumer包路径,并在该路径下创建TdmqConsumer类,内容如下:

/**
 * @Author julyWhj
 * @Description 消费者$
 * @Date 2022/1/2 10:13 上午
 **/
@Slf4j
@Service
public class TdmqConsumer {
    @Autowired
    private PulsarClient pulsarClient;
    private Consumer<byte[]> consumer;

    @PostConstruct
    public void initConsumer() throws PulsarClientException {
        log.info("MessageLoggingListener is start");
        consumer = pulsarClient.newConsumer().topic("自己的topic").subscriptionName("自己的subscriptionName").subscriptionType(SubscriptionType.Exclusive)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();
        Executors.newSingleThreadExecutor().submit(this::consumer);
    }

    /**
     * 销毁容器
     *
     * @throws PulsarClientException
     */
    @PreDestroy
    public void destroy() throws PulsarClientException {
        consumer.close();
        pulsarClient.close();
    }

    public void consumer() {
        while (true) {
            Message<byte[]> message = null;
            try {
                //使用while(true)无问题,该处会发生阻塞释放CPU资源
                message = consumer.receive();
                String json = new String(message.getData());
                if (StringUtils.isNotEmpty(json)) {
                    log.info("获取消息数据内容:{}", json);
                }
                consumer.acknowledge(message);
            } catch (PulsarClientException e) {
                log.error("数据消费失败", e);
            }
        }
    }
}

2.5、测试生产者和消费者

编写单元测试类

@Slf4j
@SpringBootTest
class SpringBootStarterTdmqApplicationTests {
    @Autowired
    private TdmqProucer proucer;

    @Test
    public void producer() throws PulsarClientException {
        MessageId messageId = proucer.sendMsg("自己的topic", "发送消息测试");
        log.info("send msg is success Id = {}", messageId);
    }
}

我们看下测试结果:

测试结果

最简单的生产消费已经完成了。接下来我们需要对代码进一步封装,这里我们参考amqp的方式,分别封装proucerTemplate和TdmqConsumer注解。实现消费者功能达到一下情况:

 @TdmqConsumer(topic = TopicConstant.MESSAGE_LOGGING_TOPIC, clazz = CreateMsgBean.class, subscriptionName = "subscriptionName")
    void consume(JddMessage<String> msg) {
        log.info("------------{}", JSONUtil.toJsonStr(msg));
    }

OK,基本使用先到这里,我们开始后续内容的封装优化。


# Java