下一代MQ中间件，不来了解下？

哈喽，大家好，我是指北君。

创新互联公司专注于察哈尔右翼前企业网站建设,响应式网站建设,商城开发。察哈尔右翼前网站建设公司,为察哈尔右翼前等地区提供建站服务。全流程按需求定制网站，专业设计，全程项目跟踪，创新互联公司专业和态度为您提供的服务

最近项目中准备使用消息中间件Apache Pulsar，借着机会先做个简单了解吧。

Apache Pulsar

Apache Pulsar是Apache软件基金会顶级项目，是下一代云原生分布式消息流平台。

Pulsar 作为下一代云原生分布式消息流平台，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐以及低延时的高可扩展流数据存储特性， 内置诸多其他系统商业版本才有的特性，是云原生时代解决实时消息流数据传输、存储和计算的最佳解决方案。

Pulsar简介

  • 系统架构

  • 功能特色
    租户和命名空间（namespace）是 Pulsar 支持多租户的两个核心概念。在租户级别，Pulsar 为特定的租户预留合适的存储空间、应用授权与认证机制。在命名空间级别，Pulsar 有一系列的配置策略（policy），包括存储配额、流控、消息过期策略和命名空间之间的隔离策略。
    Pulsar 做了队列模型和流模型的统一，在 Topic 级别只需保存一份数据，同一份数据可多次消费。以流式、队列等方式计算不同的订阅模型大大提升了灵活度。
    Pulsar 使用计算与存储分离的云原生架构，数据从 Broker 搬离，存在共享存储内部。上层是无状态 Broker，复制消息分发和服务；下层是持久化的存储层 Bookie 集群。Pulsar 存储是分片的，这种构架可以避免扩容时受限制，实现数据的独立扩展和快速恢复。
    Pulsar 原生支持跨地域复制，因此 Pulsar 可以跨不同地理位置的数据中心复制数据。当数据中心中断或网络分区时，在多个数据中心存有消息副本尤为重要，提高可用性。
    Pulsar Functions 是基于 Pulsar 的轻量级流处理方式。Pulsar Functions 直接部署在 broker 节点上（或作为 Kubernetes 集群中的容器）。通过 Pulsar Functions，Pulsar 可以直接解决许多流处理任务，简化操作。​
  • 支持客户端

Java 客户端

C++ 客户端

.Net/C# 客户端

Go 客户端

NodeJS 客户端

Ruby 客户端

Pulsar安装与部署

目前Pulsar不支持Window，下面通过Docker进行安装，可以参考官网​​https://pulsar.apache.org/docs/next/getting-started-docker/​​

同时可以安装Pulsar Manager，具体操作可以参考官方文档 ​​https://pulsar.apache.org/docs/next/administration-pulsar-manager/​​

其中Pulsar Manager 是一个网页式可视化管理与监测工具，支持多环境下的动态配置。可用于管理和监测租户、命名空间、topic、订阅、broker、集群等。

  1. window环境使用docker推荐使用Docker Desktop，和linux一样可以通过docker命令管理镜像、部署容器等操作。

打开并启动Docker Desktop后，在终端执行命令执行

_> docker search pulsar

可以查询到pulsar相关的镜像

  1. 镜像下载

这里我们选择分别下载红框的两个镜像，执行命令

_> docker pull apachepulsar/pulsar _> docker pull apachepulsar/pulsar-manager

  1. 启动
  • 启动Pulsar
docker run -it -p 6650:6650 -p 8080:8080 \
      --mount source=pulsardata,target=/pulsar/data \
      --mount source=pulsarconf,target=/pulsar/conf \
      apachepulsar/pulsar bin/pulsar standalone
  • 启动Pulsar Manager
docker run --name pulsar-manager -dit \
      -p 9527:9527 -p 7750:7750 \
      -e SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties \
      apachepulsar/pulsar-manager

添加用户：

for /f "tokens=1" %A in ('curl http://localhost:7750/pulsar-manager/csrf-token') do set CSRF_TOKEN=%A
curl -X PUT "X-XSRF-TOKEN: %CSRF_TOKEN%"   -H "Cookie: XSRF-TOKEN=%CSRF_TOKEN%;" 
  -H "Content-Type: application/json" -d "{\"name\": \"admin\", \"password\": \"123456\", \"description\": \"super user admin\", \"email\": \"admin@test.com\"}" 
  "http://localhost:7750/pulsar-manager/users/superuser"

访问：

http://localhost:9527/ 
用户名密码：admin/123456

配置environments：

这里需要保证Pulsar Manager应用服务能够访问到Pulsar应用，由于都是通过Docker部署，配置Service URL需要使用网络IP，不要用localhost。

管理界面：

Pulsar与SpringBoot集成

  • springboot version : 2.3.7.RELEASE
  • pulsar client: 2.10.2
  1. 通过Properties简单定义一些Broker相关的属性
@Data
@ConfigurationProperties(prefix = "pulsar")
public class PulsarProperties {
    
    private String cluster;
    
    private String namespace;

    private String serverUrl;

    private String token;
}
  1. 通过配置定义了一些常用的组件，比如生产、消费工厂
@Configuration
@EnableConfigurationProperties({PulsarProperties.class})
public class PulsarBootstrapConfiguration {

    private final PulsarProperties properties;

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

    @Bean(destroyMethod = "close")
    public PulsarClient pulsarClient() throws PulsarClientException {
        ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(properties.getServerUrl());
        return clientBuilder.build();
    }

    @Bean
    public PulsarProducerFactory pulsarProducerFactory() throws PulsarClientException {
        return new PulsarProducerFactory(pulsarClient(), properties);
    }

    @Bean
    public PulsarConsumerFactory pulsarConsumerFactory() throws PulsarClientException {
        return new PulsarConsumerFactory(pulsarClient(), properties);
    }

}
  1. 启动服务，在服务启动后，通过实现SmartInitializingSingleton接口，完成容器基本启动(不包含Lazy的Bean)后，开始对消费者Consumer监听
@Slf4j
@SpringBootApplication
public class PulsarApplication implements SmartInitializingSingleton {

    @Autowired
    private PulsarConsumerFactory consumerFactory;

    public static void main(String[] args) {
        SpringApplication.run(PulsarApplication.class,args);
    }

    @Override
    public void afterSingletonsInstantiated() {
        startConsumerListener();
    }

    private void startConsumerListener(){
        Consumer consumer = createConsumer();
        if( consumer != null ){
            while (!Thread.currentThread().isInterrupted()){
                CompletableFuture> completableFuture = consumer.receiveAsync();
                Message message = null;
                try {
                    message = completableFuture.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("错误",e);
                } catch (ExecutionException e) {
                    log.error("错误",e);
                }

                if( message!=null ){
                    try {
                        log.info(" 接收消息：{} ", message.getValue() );
                        consumer.acknowledge(message);
                    } catch (PulsarClientException e) {
                        consumer.negativeAcknowledge(message);
                        throw new RuntimeException(e);
                    }
                }
            }
        }
    }

    private Consumer createConsumer() {
        try {
            return consumerFactory.getConsumer(Constants.TOPIC_DEMO);
        } catch (PulsarClientException e) {
            log.error("创建consumer出错:{}", e.getMessage(),e);
        }
        return null;
    }
}
  1. 消息发送测试
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class PulsarBootTests {

    @Autowired
    private PulsarProducerFactory producerFactory;

    @Test
    public void sendMessage() throws PulsarClientException {
        Producer producer = producerFactory.getProducer(Constants.TOPIC_DEMO);

        producer.send(" 测试消息: " + new Date());

        producer.close();
    }

}
  1. 检查消息接收情况
2023-02-05 12:05:14.043  INFO 23472 --- [ulsar-timer-6-1] o.a.p.c.impl.ConsumerStatsRecorderImpl   : [TOPIC_DEMO] [sub-TOPIC_DEMO] [7c2b2] Prefetched messages: 0 --- Consume throughput received: 0.02 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.02 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0
2023-02-05 12:06:16.425  INFO 23472 --- [           main] com.sucl.pulsar.PulsarApplication        :  接收消息： 测试消息: Sun Feb 05 12:06:16 CST 2023

结束语

该篇主要通过官网对Apache Pulsar做了简单的了解与尝试，同时基于SpringBoot,以简单的示例代码实现了消息的发送与接收，其中各个组件仅仅使用了默认的配置，在生产环境需要根据Pulsar的特性以及官方API使其具有扩展性与易用性。

分享题目：下一代MQ中间件，不来了解下？
文章路径：http://www.shufengxianlan.com/qtweb/news23/125723.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

企业网站制作知识

同城分类信息

免备案虚拟主机空间　　 　移动服务器托管　　 　网站设计　　 　成都响应式网站　　 　成都VI设计　　 　成都电信光华数据中心　　 　APP定制　　 　广告吸塑字　　 　成都法律咨询　　 　成都木托盘　　 　免备案空间　　 　定制级高端网站建设　　 　成都西信服务器托管　　 　混凝土搅拌罐车　　 　ssl数字证书　　 　户外广告设计　　 　微信营销　　 　成都网站制作　　 　成都网站建设　　 　成都营销网站制作