手把手教你写Spring Boot Starter-编程学习网

之前写过关于 Apache Pulsar 的简单示例，用来了解如何使用 Pulsar 这个新生代的消息队列中间件，但是如果想要在项目中使用，还会欠缺很多，最明显的就是集成复杂，如果你用过其他消息中间件，比如 Kafka、RabbitMq，只需要简单的引入 jar，就可以通过注解+配置快速集成到项目中。

开始一个 Pulsar Starter

既然已经了解了 Apache Pulsar，又认识了 spring-boot-starter，今天不妨来看下如何写一个 pulsar-spring-boot-starter 模块。

目标

写一个完整的类似 kafka-spring-boot-starter（springboot 项目已经集成到 spring-boot-starter 中），需要考虑到很多 kafka 的特性，今天我们主要实现下面几个模板

在项目中够通过引入 jar 依赖快速集成
提供统一的配置入口
能够快速发送消息
能够基于注解实现消息的消费

定义结构

└── pulsar-starter
    ├── pulsar-spring-boot-starter
    ├── pulsar-spring-boot-autoconfigure
    ├── spring-pulsar
    ├── spring-pulsar-xx
    ├── spring-pulsar-sample
└── README.md

整个模块的结构如上其中pulsar-starter作为一个根模块，主要控制子模块依赖的其他 jar 的版本以及使用到的插件版本。类似于 Spring-Bom，这样我们在后续升级时，就可以解决各个第三方 jar 的可能存在版本冲突导致的问题。

pulsar-spring-boot-starter

该模块作为外部项目集成的直接引用 jar，可以认为是 pulsar-spring-boot-starter 组件的入口，里面不需要写任何代码，只需要引入需要的依赖（也就是下面的子模块）即可

pulsar-spring-boot-autoconfigure

该模块主要定义了 spring.factories 以及 AutoConfigure、Properties。也就是自动配置的核心（配置项+Bean 配置）

spring-pulsar

该模块是核心模块，主要的实现都在这里

spring-pulsar-xx

扩展模块，可以对 spring-pulsar 做更细化的划分

spring-pulsar-sample

starter 的使用示例项目

实现

上面我们说到实现目标，现在看下各个模块应该包含什么内容，以及怎么实现我们的目标

入口 pulsar-spring-boot-starter

上面说到 starter 主要是引入整个模块基础的依赖即可，里面不用写代码。

<dependencies>
    <dependency>
        <groupId>com.suclgroupId>
        <artifactId>spring-pulsarartifactId>
        <version>${project.version}version>
    dependency>

    <dependency>
        <groupId>com.suclgroupId>
        <artifactId>pulsar-spring-boot-autoconfigureartifactId>
        <version>${project.version}version>
    dependency>
dependencies>

pulsar-spring-boot-autoconfigure

添加 spring-boot 基础的配置

<dependencies>
     <dependency>
         <groupId>org.springframework.bootgroupId>
         <artifactId>spring-bootartifactId>
     dependency>

     <dependency>
         <groupId>org.springframework.bootgroupId>
         <artifactId>spring-boot-starter-loggingartifactId>
     dependency>

     <dependency>
         <groupId>org.springframework.bootgroupId>
         <artifactId>spring-boot-configuration-processorartifactId>
         <optional>trueoptional>
     dependency>
dependencies>

定义自动配置类PulsarAutoConfiguration：

引入Properties，基于EnableConfigurationProperties与spring-boot-configuration-processor解析 Properties 生成对应spring-configuration-metadata.json文件，这样编写 application.yml 配置时就可以自动提示配置项的属性和值了。
构建一些必须的 Bean，如 PulsarClient、ConsumerFactory、ConsumerFactory 等
Import 配置 PulsarAnnotationDrivenConfiguration，这个主要是一些额外的配置，用来支持后面的功能

@Configuration
@EnableConfigurationProperties({PulsarProperties.class})
@Import({PulsarAnnotationDrivenConfiguration.class})
public class PulsarAutoConfiguration {

    private final PulsarProperties properties;

    public PulsarAutoConfiguration(PulsarProperties properties) {
        this.properties = properties;
    }

    @Bean(destroyMethod = "close")
    public PulsarClient pulsarClient() {
        ClientBuilder clientBuilder = new ClientBuilderImpl(properties);
        return clientBuilder.build();
    }

    @Bean
    @ConditionalOnMissingBean(ConsumerFactory.class)
    public ConsumerFactory pulsarConsumerFactory() {
        return new DefaultPulsarConsumerFactory(pulsarClient(), properties.getConsumer().buildProperties());
    }

    @Bean
    @ConditionalOnMissingBean(ProducerFactory.class)
    public ProducerFactory pulsarProducerFactory() {
        return new DefaultPulsarProducerFactory(pulsarClient(), properties.getProducer().buildProperties());
    }

}

配置 spring.factory

在目录src/main/resources/META-INF下创建spring.factories，内容如下：

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  com.sucl.pulsar.autoconfigure.PulsarAutoConfiguration

spring-pulsar

添加 pulsar-client 相关的依赖

<dependencies>
     <dependency>
         <groupId>org.apache.pulsargroupId>
         <artifactId>pulsar-clientartifactId>
     dependency>

     <dependency>
         <groupId>org.springframework.bootgroupId>
         <artifactId>spring-boot-autoconfigureartifactId>
     dependency>

     <dependency>
         <groupId>org.springframeworkgroupId>
         <artifactId>spring-messagingartifactId>
     dependency>
dependencies>

定义 EnablePulsar，之前说到过，@Enable 注解主要是配合 AutoConfigure 来做功能加强，没有了自动配置，我们依然可以使用这些模块的功能。这里做了一件事，向 Spring 容器注册了两个 Bean

PulsarListenerAnnotationBeanProcessor 在 Spring Bean 生命周期中解析注解自定义注解 PulsarListener、PulsarHandler，
PulsarListenerEndpointRegistry 用来构建 Consumer 执行环境以及对 TOPIC 的监听、触发消费回调等等，可以说是最核心的 Bean

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Import({PulsarListenerConfigurationSelector.class})
public @interface EnablePulsar {

}

定义注解，参考 RabbitMq，主要针对需要关注的类与方法，分别对应注解@PulsarListener、@PulsarHandler，通过这两个注解配合可以让我们监听到关注的 TOPIC，当有消息产生时，触发对应的方法进行消费。

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface PulsarListener {

    
    String[] topics() default {};

    
    String[] tags() default {};
}

@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface PulsarHandler {

}

注解@PulsarListener 的处理流程比较复杂，这里用一张图描述，或者可以通过下面 github 的源代码查看具体实现

flow

spring-pulsar-sample

按照下面的流程，你会发现通过简单的几行代码就能够实现消息的生产与消费，并集成到项目中去。

简单写一个 SpringBoot 项目，并添加 pulsar-spring-boot-starter

<dependencies>
    <dependency>
        <groupId>com.suclgroupId>
        <artifactId>pulsar-spring-boot-starterartifactId>
        <version>${project.version}version>
    dependency>

    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-webartifactId>
    dependency>
dependencies>

2.添加配置

cycads:
  pulsar:
    service-url: pulsar://localhost:6650
  listener-topics: TOPIC_TEST

3.编写对应消费代码

@Slf4j
@Component
@PulsarListener(topics = "#{'${cycads.listener-topics}'.split(',')}")
public class PulsarDemoListener {

    @PulsarHandler
    public void onConsumer(Message message){
        log.info(">>> 接收到消息：{}", message.getPayload());
    }

}

向 Pulsar Broker 发送消息进行测试

@Slf4j
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = {ContextConfig.class})
@Import({PulsarAutoConfiguration.class})
public class ProducerTests {

    @Autowired
    private ProducerFactory producerFactory;

    @Test
    public void sendMessage() {
        Producer producer = producerFactory.createProducer("TOPIC_TEST");
        MessageId messageId = producer.send("this is a test message");
        log.info(">>>>>>> 消息发送完成：{}", messageId);
    }

    @Configuration
    @PropertySource(value = "classpath:application-test.properties")
    static class ContextConfig {
        //
    }
}

控制台可以看到这样的结果

2023-02-26 19:57:15.572  INFO 26520 --- [pulsar-01] c.s.p.s.listener.PulsarDemoListener : >>> 接收到消息：GenericMessage [payload=this is a test message, headers={id=f861488c-2afb-b2e7-21a1-f15e9759eec5, timestamp=1677412635571}]

知识点

Pulsar Client

基于 pulsar-client 提供的 ConfigurationData 扩展 Properties；了解 Pulsar Client 如何连接 Broker 并进行消息消费，包括同步消费、异步消费等等

spring.factories

实现 starter 自动配置的关键，基于 SPI 完成配置的自动加载

Spring Bean 生命周期

通过 Bean 生命周期相关扩展实现注解的解析与容器的启动，比如 BeanPostProcessor, BeanFactoryAware, SmartInitializingSingleton, InitializingBean, DisposableBean 等

Spring Messaging

基于回调与 MethodHandler 实现消息体的封装、参数解析以及方法调用；

源码示例

https://github.com/sucls/pulsar-starter.git

结束语

如果你看过 spring-kafka 的源代码，那么你会发现所有代码基本都是仿造其实现。一方面能够阅读 kafka client 在 spring 具体如何实现；同时通过编写自己的 spring starter 模块，学习整个 starter 的实现过程。

文章详情

手把手教你写Spring Boot Starter

开始一个 Pulsar Starter

目标

定义结构

实现

知识点

源码示例

结束语

软考中级精品资料免费领

相关文章

猜你喜欢

手把手教你写Spring Boot Starter

怎么手写一个Spring Boot Starter

手把手教你入门 Spring Boot + CAS 单点登录

手把手带你编写属于自己的 Starter

手把手教你写一个spring IOC容器的方法

手把手教你写--FTP(基于python)

手把手教你用Vue3写播放器

Webpack - 手把手教你写一个 loader / plugin

手把手写Spring框架

手把手教你怎么创建spring项目

花式玩 Spring Boot！过滤器竟有 N 种注册方式！手把手教你

手把手教你写Maven的archetype项目脚手架

零基础手把手教你写NFT抢购软

手把手带你开发Starter，点对点带你讲解原理

手把手教你写设计方案，你学明白了吗？

手把手教你玩Hugging Face

手把手教你使用JavaFX

手把手教你写一个Shell脚本部署你的服务

手把手教你从0手写一个力导向关系图

手把手教你写一个uniapp通用页面组件