新聞中心
哈嘍,大家好,我是了不起。

創(chuàng)新互聯(lián)-專業(yè)網(wǎng)站定制、快速模板網(wǎng)站建設(shè)、高性價(jià)比徽縣網(wǎng)站開發(fā)、企業(yè)建站全套包干低至880元,成熟完善的模板庫,直接使用。一站式徽縣網(wǎng)站制作公司更省心,省錢,快速模板網(wǎng)站建設(shè)找我們,業(yè)務(wù)覆蓋徽縣地區(qū)。費(fèi)用合理售后完善,十載實(shí)體公司更值得信賴。
之前寫過關(guān)于 Apache Pulsar 的簡單示例,用來了解如何使用 Pulsar 這個(gè)新生代的消息隊(duì)列中間件,但是如果想要在項(xiàng)目中使用,還會欠缺很多,最明顯的就是 集成復(fù)雜,如果你用過其他消息中間件,比如 Kafka、RabbitMq,只需要簡單的引入 jar,就可以通過注解+配置快速集成到項(xiàng)目中。
開始一個(gè) Pulsar Starter
既然已經(jīng)了解了 Apache Pulsar,又認(rèn)識了 spring-boot-starter,今天不妨來看下如何寫一個(gè) pulsar-spring-boot-starter 模塊。
目標(biāo)
寫一個(gè)完整的類似 kafka-spring-boot-starter(springboot 項(xiàng)目已經(jīng)集成到 spring-boot-starter 中),需要考慮到很多 kafka 的特性, 今天我們主要實(shí)現(xiàn)下面幾個(gè)模板
- 在項(xiàng)目中夠通過引入 jar 依賴快速集成
- 提供統(tǒng)一的配置入口
- 能夠快速發(fā)送消息
- 能夠基于注解實(shí)現(xiàn)消息的消費(fèi)
定義結(jié)構(gòu)
└── pulsar-starter
├── pulsar-spring-boot-starter
├── pulsar-spring-boot-autoconfigure
├── spring-pulsar
├── spring-pulsar-xx
├── spring-pulsar-sample
└── README.md
整個(gè)模塊的結(jié)構(gòu)如上其中pulsar-starter作為一個(gè)根模塊,主要控制子模塊依賴的其他 jar 的版本以及使用到的插件版本。類似于 Spring-Bom,這樣我們在后續(xù)升級 時(shí),就可以解決各個(gè)第三方 jar 的可能存在版本沖突導(dǎo)致的問題。
- pulsar-spring-boot-starter
該模塊作為外部項(xiàng)目集成的直接引用 jar,可以認(rèn)為是 pulsar-spring-boot-starter 組件的入口,里面不需要寫任何代碼,只需要引入需要的依賴(也就是下面的子模塊)即可
- pulsar-spring-boot-autoconfigure
該模塊主要定義了 spring.factories 以及 AutoConfigure、Properties。也就是自動配置的核心(配置項(xiàng)+Bean 配置)
- spring-pulsar
該模塊是核心模塊,主要的實(shí)現(xiàn)都在這里
- spring-pulsar-xx
擴(kuò)展模塊,可以對 spring-pulsar 做更細(xì)化的劃分
- spring-pulsar-sample
starter 的使用示例項(xiàng)目
實(shí)現(xiàn)
上面我們說到實(shí)現(xiàn)目標(biāo),現(xiàn)在看下各個(gè)模塊應(yīng)該包含什么內(nèi)容,以及怎么實(shí)現(xiàn)我們的目標(biāo)
- 入口 pulsar-spring-boot-starter
上面說到 starter 主要是引入整個(gè)模塊基礎(chǔ)的依賴即可,里面不用寫代碼。
com.sucl
spring-pulsar
${project.version}
com.sucl
pulsar-spring-boot-autoconfigure
${project.version}
- pulsar-spring-boot-autoconfigure
- 添加 spring-boot 基礎(chǔ)的配置
org.springframework.boot
spring-boot
org.springframework.boot
spring-boot-starter-logging
org.springframework.boot
spring-boot-configuration-processor
true
- 定義自動配置類PulsarAutoConfiguration:
- 引入Properties,基于EnableConfigurationProperties與spring-boot-configuration-processor解析 Properties 生成對應(yīng)spring-configuration-metadata.json文件,這樣編寫 application.yml 配置時(shí)就可以自動提示配置項(xiàng)的屬性和值了。
- 構(gòu)建一些必須的 Bean,如 PulsarClient、ConsumerFactory、ConsumerFactory 等
- Import 配置 PulsarAnnotationDrivenConfiguration,這個(gè)主要是一些額外的配置,用來支持后面的功能
@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下創(chuàng)建spring.factories,內(nèi)容如下:
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.sucl.pulsar.autoconfigure.PulsarAutoConfiguration
- spring-pulsar
- 添加 pulsar-client 相關(guān)的依賴
org.apache.pulsar
pulsar-client
org.springframework.boot
spring-boot-autoconfigure
org.springframework
spring-messaging
- 定義 EnablePulsar,之前說到過,@Enable 注解主要是配合 AutoConfigure 來做功能加強(qiáng),沒有了自動配置,我們依然可以使用這些模塊的功能。這里做了一件事,向 Spring 容器注冊了兩個(gè) Bean
- PulsarListenerAnnotationBeanProcessor 在 Spring Bean 生命周期中解析注解自定義注解 PulsarListener、PulsarHandler,
- PulsarListenerEndpointRegistry 用來構(gòu)建 Consumer 執(zhí)行環(huán)境以及對 TOPIC 的監(jiān)聽、觸發(fā)消費(fèi)回調(diào)等等,可以說是最核心的 Bean
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Import({PulsarListenerConfigurationSelector.class})
public @interface EnablePulsar {
}
- 定義注解,參考 RabbitMq,主要針對需要關(guān)注的類與方法,分別對應(yīng)注解@PulsarListener、@PulsarHandler,通過這兩個(gè)注解配合可以讓我們監(jiān)聽到關(guān)注的 TOPIC, 當(dāng)有消息產(chǎn)生時(shí),觸發(fā)對應(yīng)的方法進(jìn)行消費(fèi)。
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface PulsarListener {
/**
*
* @return TOPIC 支持SPEL
*/
String[] topics() default {};
/**
*
* @return TAGS 支持SPEL
*/
String[] tags() default {};
}
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface PulsarHandler {
}
- 注解@PulsarListener 的處理流程比較復(fù)雜,這里用一張圖描述,或者可以通過下面 github 的源代碼查看具體實(shí)現(xiàn)
flow
- spring-pulsar-sample
按照下面的流程,你會發(fā)現(xiàn)通過簡單的幾行代碼就能夠?qū)崿F(xiàn)消息的生產(chǎn)與消費(fèi),并集成到項(xiàng)目中去。
- 簡單寫一個(gè) SpringBoot 項(xiàng)目,并添加 pulsar-spring-boot-starter
com.sucl
pulsar-spring-boot-starter
${project.version}
org.springframework.boot
spring-boot-starter-web
2.添加配置
cycads:
pulsar:
service-url: pulsar://localhost:6650
listener-topics: TOPIC_TEST
3.編寫對應(yīng)消費(fèi)代碼
@Slf4j
@Component
@PulsarListener(topics = "#{'${cycads.listener-topics}'.split(',')}")
public class PulsarDemoListener {
@PulsarHandler
public void onConsumer(Message message){
log.info(">>> 接收到消息:{}", message.getPayload());
}
}
- 向 Pulsar Broker 發(fā)送消息進(jìn)行測試
@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(">>>>>>> 消息發(fā)送完成:{}", messageId);
}
@Configuration
@PropertySource(value = "classpath:application-test.properties")
static class ContextConfig {
//
}
}
- 控制臺可以看到這樣的結(jié)果
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}]
知識點(diǎn)
- Pulsar Client
基于 pulsar-client 提供的 ConfigurationData 擴(kuò)展 Properties;了解 Pulsar Client 如何連接 Broker 并進(jìn)行消息消費(fèi),包括同步消費(fèi)、異步消費(fèi)等等
- spring.factories
實(shí)現(xiàn) starter 自動配置的關(guān)鍵,基于 SPI 完成配置的自動加載
- Spring Bean 生命周期
通過 Bean 生命周期相關(guān)擴(kuò)展實(shí)現(xiàn)注解的解析與容器的啟動,比如 BeanPostProcessor, BeanFactoryAware, SmartInitializingSingleton, InitializingBean, DisposableBean 等
- Spring Messaging
基于回調(diào)與 MethodHandler 實(shí)現(xiàn)消息體的封裝、參數(shù)解析以及方法調(diào)用;
源碼示例
??https://github.com/sucls/pulsar-starter.git??
結(jié)束語
如果你看過 spring-kafka 的源代碼,那么你會發(fā)現(xiàn)所有代碼基本都是仿造其實(shí)現(xiàn)。一方面能夠閱讀 kafka client 在 spring 具體如何實(shí)現(xiàn);同時(shí)通過編寫自己的 spring starter 模塊,學(xué)習(xí) 整個(gè) starter 的實(shí)現(xiàn)過程。
網(wǎng)站欄目:手把手教你寫SpringBootStarter
URL鏈接:http://fisionsoft.com.cn/article/ccoppjs.html


咨詢
建站咨詢
