再見RocketMQ！全新一代消息中間件，帶可視化管理，文檔賊友好！

最近很火的消息中間件Pulsar，本想學習下，發現網上很多都是介紹性能和對比Kafka的文章，實踐的文章很少！于是對着官方文檔實踐了一波，寫下了這篇文章，估計是國内第一篇Pulsar實戰文章了，希望對大家有所幫助！

Pulsar簡介

Pulsar是一個用于服務端到服務端的消息中間件，具有多租戶、高性能等優勢。Pulsar最初由Yahoo開發，目前由Apache軟體基金會管理。Pulsar采用

釋出-訂閱

的設計模式，Producer釋出消息到Topic，Consumer訂閱Topic、處理Topic中的消息。

Pulsar具有如下特性：

Pulsar的單個執行個體原生支援叢集。
極低的釋出延遲和端到端延遲。
可無縫擴充到超過一百萬個Topic。
簡單易用的用戶端API，支援Java、Go、Python和C++。
支援多種Topic訂閱模式（獨占訂閱、共享訂閱、故障轉移訂閱）。
通過Apache BookKeeper提供的持久化消息存儲機制保證消息傳遞。

Pulsar安裝

使用Docker安裝Pulsar是最簡單的，這次我們使用Docker來安裝。

首先下載下傳Pulsar的Docker鏡像；

docker pull apachepulsar/pulsar:2.7.1

複制

下載下傳完成後運作Pulsar容器，http協定通路使用 8080 端口，pulsar協定（Java、Python等用戶端）通路使用 6650 端口。

docker run --name pulsar \
-p 6650:6650 \
-p 8080:8080 \
--mount source=pulsardata,target=/pulsar/data \
--mount source=pulsarconf,target=/pulsar/conf \
-d apachepulsar/pulsar:2.7.1 \
bin/pulsar standalone

複制

Pulsar可視化

Pulsar Manager 是官方提供的可視化工具，可以對多個Pulsar進行可視化管理，雖然功能不多，但也基本夠用了，支援Docker部署。

下載下傳 pulsar-manager 的Docker鏡像；

docker pull apachepulsar/pulsar-manager:v0.2.0

複制

下載下傳完成後運作 pulsar-manager 容器，從 9527 端口可以通路Web頁面；

docker run -it --name pulsar-manager\
    -p 9527:9527 -p 7750:7750 \
    -e SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties \
    -d apachepulsar/pulsar-manager:v0.2.0

複制

運作成功後，我們剛開始無法通路，需要建立管理者賬号，這裡建立賬号為 admin:apachepulsar ：

CSRF_TOKEN=$(curl http://localhost:7750/pulsar-manager/csrf-token)
curl \
    -H "X-XSRF-TOKEN: $CSRF_TOKEN" \
    -H "Cookie: XSRF-TOKEN=$CSRF_TOKEN;" \
    -H 'Content-Type: application/json' \
    -X PUT http://localhost:7750/pulsar-manager/users/superuser \
    -d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "[email protected]"}'

複制

建立成功後，通過登入頁面進行登入，通路位址：http://192.168.5.78:9527

登入成功後我們需要先配置一個環境，就是将需要管理的Pulsar服務配置上去，配置的 Service URL 為：http://192.168.5.78:8080

可以檢視Tenant清單；

可以檢視Topic清單和管理Topic；

還可以檢視Topic的詳細資訊。

Pulsar結合SpringBoot使用

Pulsar結合SpringBoot使用也是非常簡單的，我們可以使用Pulsar官方的Java SDK，也可以使用第三方的SpringBoot Starter。這裡使用Starter，非常簡單！

首先在 pom.xml 中添加Pulsar相關依賴；

<!--SpringBoot整合Pulsar-->
<dependency>
    <groupId>io.github.majusko</groupId>
    <artifactId>pulsar-java-spring-boot-starter</artifactId>
    <version>1.0.4</version>
</dependency>

複制

然後在 application.yml 中添加Pulsar的 Service URL 配置；

pulsar:
  service-url: pulsar://192.168.5.78:6650

複制

再添加Pulsar的Java配置，聲明兩個Topic，并确定好發送的消息類型；

/**
 * Pulsar配置類
 * Created by macro on 2021/5/21.
 */
@Configuration
public class PulsarConfig {
    @Bean
    public ProducerFactory producerFactory() {
        return new ProducerFactory()
                .addProducer("bootTopic", MessageDto.class)
                .addProducer("stringTopic", String.class);
    }
}

複制

建立Pulsar生産者，往Topic中發送消息，這裡可以發現Pulsar是支援直接發送消息對象的；

/**
 * Pulsar消息生産者
 * Created by macro on 2021/5/19.
 */
@Component
public class PulsarProducer {
    @Autowired
    private PulsarTemplate<MessageDto> template;

    public void send(MessageDto message){
        try {
            template.send("bootTopic",message);
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }
}

複制

建立Pulsar消費者，從Topic中擷取并消費消息，也是可以直接擷取到消息對象的；

/**
 * Pulsar消息消費者
 * Created by macro on 2021/5/19.
 */
@Slf4j
@Component
public class PulsarRealConsumer {

    @PulsarConsumer(topic="bootTopic", clazz= MessageDto.class)
    public void consume(MessageDto message) {
        log.info("PulsarRealConsumer consume id:{},content:{}",message.getId(),message.getContent());
    }

}

複制

添加測試接口，調用生産者發送消息；

/**
 * Pulsar功能測試
 * Created by macro on 2021/5/19.
 */
@Api(tags = "PulsarController", description = "Pulsar功能測試")
@Controller
@RequestMapping("/pulsar")
public class PulsarController {

    @Autowired
    private PulsarProducer pulsarProducer;

    @ApiOperation("發送消息")
    @RequestMapping(value = "/sendMessage", method = RequestMethod.POST)
    @ResponseBody
    public CommonResult sendMessage(@RequestBody MessageDto message) {
        pulsarProducer.send(message);
        return CommonResult.success(null);
    }
}

複制

在Swagger中調用接口進行測試；

調用成功後，控制台将輸入如下資訊，表示消息已經被成功接收并消費了。

2021-05-21 16:25:07.756  INFO 11472 --- [al-listener-3-1] c.m.m.tiny.component.PulsarRealConsumer  : PulsarRealConsumer consume id:1,content:SpringBoot Message!

複制

總結

上次寫了一篇《吊炸天的 Kafka 圖形化工具 Eagle，必須推薦給你！》介紹了Kafka的基本使用，這裡和Pulsar做個對比。Pulsar對Docker支援無疑是更好的，官方文檔也更全。對比下圖形化工具

Pulsar Manager

和

Kafka Eagle

，Pulsar的圖形化工具感覺有點簡陋。介于目前雅虎、騰訊、360等網際網路大廠都在使用Pulsar，Pulsar的性能和穩定性應該是很不錯的！

參考資料

Pulsar的官方文檔很全，樣式也不錯，基本照着文檔來一遍就能入門了。

Pulsar官方文檔：https://pulsar.apache.org/docs/en/standalone-docker/
SpringBoot Starter官方文檔：https://github.com/majusko/pulsar-java-spring-boot-starter

項目源碼位址

https://github.com/macrozheng/mall-learning/tree/master/mall-tiny-pulsar