里程碑 版本 功能点 作者 完成 1.0.0 支持pulsartemplate发送消息&支持自定义注解实例化consumer监听消息 howinfun ✅ 1.1.0 支持动态开启/关闭consumer消费线程池、支持自定义配置consuemr消费线程池参数 howinfun ✅ 1.2.0 支持spring容器停止时，释放pulsar所有相关资源 howinfun todo 1.3.0 支持多pulsar数据源 howinfun todo 一、背景
pulsar 作为新生代云原生消息队列，越来越得到开发者的热爱;而我们现在基本上的项目都是基于 springboot 上开发的，但是我们可发现，到目前为止都没有比较大众和成熟的关于 pulsar 的 starter，所以我们需要自己整一个，最终防止常规使用 pulsar api 时产生大量的重复代码。
二、设计思路
由于是第一版的设计，所以咱们是从简单开始，不会最初就设计得很复杂，尽量保留 pulsar api 原生的功能。
2. 1. pulsarclient
我们都知道，不管是 producer 还是 consumer，都是由 pulsarclient 创建的。
对了，pulsarclient 能够依据业务需要自定义很多参数，但是第一版的设计只会支持比较常用的参数。
我们这个组件支持下面功能点
支持 pulsarclient 参数配置外部化，参数可配置在 applicatin.properties 中。 支持 applicatin.properties 提供配置提示信息。 读取外部配置文件，根据参数实例化 pulsarclient，并注入到 ioc 容器中。 2. 2. producer
producer是发送消息的组件。
这里我们提供一个模版类，能够依据需求创建对应的 producer 实例。 支持将 topicproducer 关系缓存起来，防止重复创建 producer 实例。 支持同步/异步发送消息。 2. 3. consumer
consumer是消费消息的组件。
这里我们提供一个抽象类，开发者只需要集成此实现类并实现 doreceive 方法即可，即消费消息的逻辑方法。 接着还提供一个自定义注解，自定义注解支持自定义 consmuer 配置，例如topic、tenant、namespace等。 实现类加入上述自定义注解后，组件将会自动识别并且生成对应的 consumer 实例。 支持同步/线程池异步消费。 三、使用例子 3. 1. 引入依赖io.github.howinfunwinfun-pulsar-spring-boot-starter1.1.0 3. 2. 加入配置 pulsar.service-url=pulsar://127.0.0.1:6650pulsar.tenant=winfunpulsar.namespace=studypulsar.operation-timeout=30pulsar.io-threads=10pulsar.listener-threads=10 3. 3. 发送消息 /***发送消息*@author:winfun**/@restcontroller@requestmapping(msg)publicclassmessagecontroller{@autowiredprivatepulsartemplatepulsartemplate;@autowiredprivatepulsarpropertiespulsarproperties;/****往指定topic发送消息*@authorwinfun*@paramtopictopic*@parammsgmsg*@return{@linkstring}**/@getmapping(/{topic}/{msg})publicstringsend(@pathvariable(topic)stringtopic,@pathvariable(msg)stringmsg)throwsexception{this.pulsartemplate.createbuilder().persistent(boolean.true).tenant(this.pulsarproperties.gettenant()).namespace(this.pulsarproperties.getnamespace()).topic(topic).send(msg);returnsuccess;}} 3. 4. 消费消息 /***@author:winfun*@date:2021/8/208:13下午**/@slf4j@pulsarlistener(topics={test-topic2},threadpool=@threadpool(corethreads=2,maxcorethreads=3,threadpoolname=test-thread-pool))publicclassconsumerlistenerextendsbasemessagelistener{/***消费消息*@paramconsumer消费者*@parammsg消息*/@overrideprotectedvoiddoreceived(consumerconsumer,messagemsg){log.info(成功消费消息{},msg.getvalue());try{consumer.acknowledge(msg);}catch(pulsarclientexceptione){e.printstacktrace();}}/****是否开启异步消费*@return{@linkboolean}**/@overridepublicbooleanenableasync(){returnboolean.true;}} 四、源码
源码就不放在这里分析了，大家可到github上看看，如果有什么代码上面的建议或意见，欢迎大家提mr。