基于Docker-SpringBoot3.4.0集成Apache Pulsar4.0.1实现消息发布和订阅

itbeien2025年3月3日大约 5 分钟

基于Docker-SpringBoot3.4.0集成Apache Pulsar4.0.1实现消息发布和订阅

贝恩聊架构-SpringBoot3专栏内容定位主要为SpringBoot3整合各大框架和中间件，通过输出入门技术文章和基于源代码的技术案例，让读者基于本系列技术文章和源代码快速掌握SpringBoot3和各大开源框架、中间件的整合应用。给技术开发者节约环境搭建和框架整合所耗费的时间，开发人员都知道在框架整合的过程中会遇到各种技术问题和环境问题，所以参考已有搭建好的框架及功能能大大提高学习和工作效率。

Apache Pulsar 是一个一体化的云原生消息传递和流处理平台。消息既可以单独被消费和确认，也可以作为流进行消费，且延迟低于 10 毫秒。本篇文章主要讲解Apache Pulsar4.0.1安装流程和SpringBoot整合Apache Pulsar实现消息发送和订阅，由于Apache Pulsar不支持直接在windows安装运行，我们今天采用Docker方式在Windows上进行安装。

完整代码在文章最后，如果觉得本篇文章对你有用，记得点赞、关注、收藏哦。你的支持是我持续更新的动力!

SpringBoot3专栏软件环境

JDK17.0.12
SpringBoot3.4.0
Apache Pulsar4.0.1
Docker Desktop4.36.0 for Windows
IDEA2024.2.0.2
Win10专业版

我们先看本篇文章对应的项目结构，请看下图

1 安装Docker Desktop for Windows

1.1 下载地址

https://desktop.docker.com/win/main/amd64/Docker Desktop Installer.exe?utm_source=docker&utm_medium=webreferral&utm_campaign=docs-driven-download-win-amd64

1.2 启用windows功能

安装Docker Desktop for Windows之前要确保以下功能打开，开启之后需要重启电脑

虚拟机平台
适用于Linux的Windows子系统
Hyper-V

另外要确保已经开启虚拟化功能，见下图

1.3 注册并登录docker

1.4 基于docker安装Apache Pulsar

docker run -it  -p 6650:6650  -p 8080:8080  --mount source=pulsardata,target=/pulsar/data  --mount source=pulsarconf,target=/pulsar/conf  apachepulsar/pulsar:4.0.1  bin/pulsar standalone

2 项目搭建流程

2.1 pom.xml依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>cn.itbeien</groupId>
        <artifactId>springboot3-labs-master</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <artifactId>springboot-apachepulsar</artifactId>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-pulsar</artifactId>
        </dependency>
    </dependencies>

</project>

2.2 配置文件

server.port=5005
spring.pulsar.client.service-url=pulsar://localhost:6650

2.3 生产者代码

package cn.itbeien.pulsar.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.stereotype.Component;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@Component
@Slf4j
public class ProducerService {
    private final PulsarTemplate<String> pulsarTemplate;

    public ProducerService(PulsarTemplate<String> pulsarTemplate) {
        this.pulsarTemplate = pulsarTemplate;
    }

    public void someMethod() {
        log.info("生产者者发送消息...");
        this.pulsarTemplate.send("itbeienTopic", "Hello itbeien!");
    }
}

2.4 消费者代码

package cn.itbeien.pulsar.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.stereotype.Component;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@Component
@Slf4j
public class ConsumerService {

    @PulsarListener(topics = "itbeienTopic")
    public void processMessage(String content) {
        log.info("消费者接收到的消息:{}",content);
    }
}

2.5 单元测试

package cn.itbeien.pulsar.test;


import cn.itbeien.pulsar.service.ProducerService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.PulsarTemplate;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Copyright© 2024 itbeien
 */
@SpringBootTest
public class PulsarApp {
    @Autowired
    private ProducerService producerService;
    @Test
   public void producer() throws Exception{
        producerService.someMethod();

        Thread.sleep(10000);
   }

}