消息队列篇--原理篇--Pulsar（Namespace，BookKeeper，类似Kafka甚至更好的消息队列）

news/2025/1/22 20:59:22/

Apache Pulusar是一个分布式、多租户、高性能的发布/订阅（Pub/Sub）消息系统，最初由Yahoo开发并开源。它结合了Kafka和传统消息队列的优点，提供高吞吐量、低延迟、强一致性和可扩展的消息传递能力，适用于大规模分布式系统的实时数据处理和异步通信。
Pulsar的架构设计结合了消息队列和流处理的特点，既可以作为传统消息队列使用，也可以作为流处理平台支持实时数据处理。

主要特点：

  • 分布式架构：Pulsar采用分层架构，将消息存储与代理服务分离，提供了更好的水平扩展能力和故障隔离。
  • 多租户支持：Pulsar支持多租户部署，不同租户可以共享同一集群，同时保证资源隔离和安全性。
  • 持久化和一致性：Pulsar支持消息的持久化存储，并通过BookKeeper提供强一致性保障。
  • 灵活的消息模型：Pulsar支持多种消息传递模式，包括Pub/Sub、P2P和Key_Shared订阅模式。
  • 多语言支持：Pulsar提供了多种编程语言的客户端库，如Java、Python、Go、C++等。
  • 丰富的生态：Pulsar拥有活跃的社区和丰富的生态系统，支持与其他工具和服务集成，如Kafka Connect、Flink、Spark等。

1、核心概念

（1）、命名空间（Namespace）

命名空间是Pulsar中的一个逻辑单元，用于组织和管理主题（Topic）。每个命名空间可以包含多个主题，并且可以为不同的命名空间设置不同的配置，例如保留策略、订阅类型等。命名空间通常用于实现多租户隔离。

（2）、主题（Topic）

主题是Pulsar中的消息通道，生产者（Producer）将消息发送到主题，消费者（Consumer）从主题中消费消息。

Pulsar支持两种类型的主题：

  • 持久化主题（Persistent Topic）：消息会被持久化存储，确保即使在broker故障的情况下也不会丢失。
  • 非持久化主题（Non-Persistent Topic）：消息不会被持久化存储，适用于对延迟敏感但对可靠性要求较低的场景。

（3）、订阅（Subscription）

订阅是消费者与主题之间的绑定关系。Pulsar支持多种订阅类型，每种订阅类型决定了消息的分发方式：

  • 独占订阅（Exclusive Subscription）：只有一个消费者可以订阅该主题，其他消费者无法订阅。
  • 共享订阅（Shared Subscription）：多个消费者可以订阅同一个主题，消息会被轮询分发给不同的消费者。
  • 故障转移订阅（Failover Subscription）：多个消费者可以订阅同一个主题，但只有一个是活跃的消费者，其他消费者作为备用。当活跃消费者失败时，备用消费者会接管消息消费。
  • Key_Shared 订阅：基于消息的key进行分区，确保相同key的消息总是被分发给同一个消费者。

（4）、消息（Message）

消息是Pulsar中的基本数据单位，由生产者发送到主题。

每个消息可以包含以下属性：

  • 消息体（Payload）：消息的实际内容，可以是任意二进制数据。
  • 消息ID（Message ID）：唯一标识每条消息的ID，用于确认消息的消费状态。
  • 属性（Properties）：用户可以为消息添加自定义的键值对属性，方便后续处理。
  • 时间戳（Timestamp）：消息的创建时间或发送时间。

（5）、分区（Partition）

Pulsar支持主题分区，即将一个主题划分为多个分区，每个分区可以独立地处理消息。分区可以提高主题的吞吐量和并发性，特别是在高负载场景下。Pulsar会自动将消息均匀分布到不同的分区中。

（6）、Broker

Broker是Pulsar的核心组件之一，负责接收生产者的消息并将其分发给消费者。注意，Broker不直接存储消息，而是将消息委托给BookKeeper进行持久化存储。Broker负责管理主题、订阅和消费者的连接，并处理消息的路由和分发。

（7）、BookKeeper

BookKeeper是Pulsar的持久化存储层，负责将消息持久化到磁盘。BookKeeper采用分布式日志存储机制，提供了高可用性和强一致性保障。每个消息会被写入多个BookKeeper节点，确保即使部分节点故障也不会丢失数据。

（8）、ZooKeeper

ZooKeeper是Pulsar的元数据管理组件，用于存储集群的配置信息、主题和命名空间的元数据、以及Broker和BookKeeper的状态信息。ZooKeeper提供了分布式协调服务，确保Pulsar集群的一致性和可靠性。

2、架构设计

Pulsar的架构设计采用了分层结构，将消息存储与代理服务分离，使得系统更加模块化和可扩展。

结构示例图：
在这里插入图片描述

Pulsar的主要组件及其作用：

  • Broker：负责接收生产者的消息并将其分发给消费者。Broker不直接存储消息，而是将消息委托给BookKeeper进行持久化存储。Broker还负责管理主题、订阅和消费者的连接。

  • BookKeeper：即上图BK Client。负责将消息持久化到磁盘，提供高可用性和强一致性保障。BookKeeper采用分布式日志存储机制，确保消息的安全性和可靠性。

  • Bookie：Bookie是BookKeeper的存储节点组成，持久化地存储消息。BookKeeper采用分布式日志存储的方式，将消息以日志的形式存储在多个Bookie节点上。这种设计确保了消息的可靠性和持久性，即使在节点故障的情况下也能保证消息不丢失。

  • ZooKeeper：负责存储集群的元数据，包括主题、命名空间、Broker和BookKeeper的状态信息。ZooKeeper提供了分布式协调服务，确保集群的一致性和可靠性。

  • Proxy（可选）：Pulsar提供了一个可选的代理层（Proxy），允许客户端通过HTTP或WebSocket协议与Pulsar集群进行通信。Proxy可以简化客户端的连接管理，并提供跨区域访问的能力。

  • Function：Pulsar提供了一个轻量级的流处理框架（Pulsar Functions），允许用户编写简单的流处理逻辑并将其部署到Pulsar集群中。Pulsar Functions可以用于实时数据处理、事件驱动计算等场景。

  • SQL：Pulsar提供了一个SQL查询引擎（Pulsar SQL），允许用户通过SQL语句查询Pulsar中的消息数据。Pulsar SQL可以用于数据分析、监控和告警等场景。

3、特性与优势

（1）、高吞吐量和低延迟

Pulsar采用了分层架构，将消息存储与代理服务分离，使得系统能够同时具备高吞吐量和低延迟。Broker负责处理消息的路由和分发，而BookKeeper负责持久化存储，两者相互协作，确保消息的高效传递。

（2）、多租户支持

Pulsar支持多租户部署，不同租户可以共享同一集群，同时保证资源隔离和安全性。每个租户可以拥有自己的命名空间，并可以根据需要设置不同的配置，例如保留策略、订阅类型等。
即：类似Nacos的命名空间，实现配置，服务等隔离。

（3）、持久化和一致性

Pulsar支持消息的持久化存储，并通过BookKeeper提供强一致性保障。每个消息会被写入多个Bookie节点，确保即使部分节点故障也不会丢失数据。Pulsar还支持事务和幂等性，确保消息的可靠传递。

（4）、灵活的消息模型

Pulsar支持多种消息传递模式，包括Pub/Sub、P2P和Key_Shared订阅模式。用户可以根据实际需求选择合适的订阅类型，满足不同的业务场景。Pulsar还支持消息的重播、回溯和跳过等功能，方便用户进行调试和故障排查。

（5）、多语言支持

Pulsar提供了多种编程语言的客户端库，包括Java、Python、Go、C++等。用户可以根据自己的技术栈选择合适的客户端库，快速集成Pulsar到应用程序中。

（6）、丰富的生态

Pulsar拥有活跃的社区和丰富的生态系统，支持与其他工具和服务集成。例如，Pulsar可以与Kafka Connect、Flink、Spark等工具集成，实现数据的实时处理和分析。Pulsar还提供了Pulsar Functions和Pulsar SQL等功能，进一步扩展了其应用场景。

4、应用场景

（1）、实时数据处理

Pulsar的高吞吐量和低延迟特性使其非常适合用于实时数据处理场景。例如，电商网站可以使用Pulsar来处理订单、支付、库存等实时数据，确保数据的及时性和准确性。

（2）、物联网（IoT）

Pulsar的分布式架构和多租户支持使其非常适合用于物联网场景。物联网设备可以将传感器数据发送到Pulsar，Pulsar可以将这些数据分发给不同的消费者进行处理。Pulsar还支持消息的重播和回溯功能，方便用户进行历史数据分析。

（3）、微服务架构

Pulsar可以作为微服务之间的消息总线，实现服务间的异步通信。微服务可以通过Pulsar发送和接收消息，避免阻塞主线程，提高系统的响应速度和稳定性。

（4）、日志收集和监控

Pulsar可以用于日志收集和监控场景，将应用的日志数据发送到Pulsar，Pulsar可以将这些数据分发给不同的消费者进行处理。Pulsar还支持消息的持久化存储，确保日志数据不会丢失。

（5）、事件驱动架构

Pulsar支持事件驱动架构，用户可以将事件发送到Pulsar，Pulsar可以将这些事件分发给不同的消费者进行处理。Pulsar还支持消息的重播和回溯功能，方便用户进行事件的回放和调试。

5、代码示例

（1）、生产者示例

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.MessageId;public class PulsarProducerExample {public static void main(String[] args) throws Exception {// 1、创建Pulsar客户端try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build()) {// 2、创建生产者try (Producer<byte[]> producer = client.newProducer().topic("persistent://public/default/example-topic")    // 指定主题.create()) {// 3、发送消息for (int i = 0; i < 10; i++) {String message = "Hello, Pulsar! " + i;MessageId msgId = producer.send(message.getBytes());    // 发送消息System.out.println(" [x] Sent message: " + message + ", msgId: " + msgId);}}}}
}

（2）、消费者示例

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;public class PulsarConsumerExample {public static void main(String[] args) throws Exception {// 1、创建Pulsar客户端try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build()) {// 2、创建消费者try (Consumer<byte[]> consumer = client.newConsumer().topic("persistent://public/default/example-topic")   // 监听的主题.subscriptionName("example-subscription").subscriptionType(SubscriptionType.Shared).subscribe()) {// 3、接收和消费消息while (true) {       // 利用循环接收消息Message<byte[]> msg = consumer.receive();      // 具体接收消息try {System.out.println(" [x] Received message: " + new String(msg.getData()));consumer.acknowledge(msg);  // 4、确认消息已消费} catch (Exception e) {consumer.negativeAcknowledge(msg);  // 5、处理失败，重新投递}}}}}
}

6、Pulsar总结

Apache Pulsar是一个功能强大、架构灵活的消息系统，特别适合大规模分布式系统的实时数据处理和异步通信。它的分层架构、多租户支持、持久化和一致性保障、灵活的消息模型等特点，使其在性能、可靠性和可扩展性方面表现出色。Pulsar还拥有丰富的生态系统，支持与其他工具和服务集成，适用于多种应用场景。

乘风破浪会有时，直挂云帆济沧海！！！


http://www.ppmy.cn/news/1565309.html

相关文章

RabbitMQ--延迟队列

RabbitMQ--延迟队列

&#xff08;一&#xff09;延迟队列 1.概念 延迟队列是一种特殊的队列&#xff0c;消息被发送后&#xff0c;消费者并不会立刻拿到消息&#xff0c;而是等待一段时间后&#xff0c;消费者才可以从这个队列中拿到消息进行消费 2.应用场景 延迟队列的应用场景很多&#xff0c;…
阅读更多...
【opencv】第10章 角点检测

【opencv】第10章 角点检测

第10章 角点检测 10.1 Harris角点检测 10.1.1 兴趣点与角点 在图像处理和与计算机视觉领域&#xff0c;兴趣点(interest points),也被称作关键点 (key points)、特 征 点(feature points)。它被大量用于解决物体识别、图像识别、 图像匹配、视觉跟踪、三维重建等一系列的问题…
阅读更多...
三天急速通关Java基础知识：Day1 基本语法

三天急速通关Java基础知识：Day1 基本语法

三天急速通关JAVA基础知识&#xff1a;Day1 基本语法 0 文章说明1 关键字 Keywords2 注释 Comments2.1 单行注释2.2 多行注释2.3 文档注释 3 数据类型 Data Types3.1 基本数据类型3.2 引用数据类型 4 变量与常量 Variables and Constant5 运算符 Operators6 字符串 String7 输入…
阅读更多...
如何使用CRM数据分析优化销售和客户关系？

如何使用CRM数据分析优化销售和客户关系？

嘿&#xff0c;大家好&#xff01;你有没有想过为什么有些公司在市场上如鱼得水&#xff0c;而另一些却在苦苦挣扎&#xff1f;答案可能就藏在他们的销售策略和客户关系管理&#xff08;CRM&#xff09;系统里。今天我们要聊的就是如何通过有效的 CRM 数据分析来提升你的销售额…
阅读更多...
寒假刷题Day11

寒假刷题Day11

一、2516. 每种字符至少取 K 个 class Solution { public:int takeCharacters(string s, int k) {int cnt[3]{};for (char c : s){cnt [c - a];}if (cnt[0] < k || cnt[1] < k || cnt[2] < k) {return -1; // 字母个数不足 k}int mx 0, left 0;for(int right 0; r…
阅读更多...
机器学习有哪些应用场景

机器学习有哪些应用场景

机器学习的应用场景非常广泛&#xff0c;涵盖了各个行业和领域。以下是一些主要的应用场景&#xff1a; 一、图像和视觉识别 人脸识别&#xff1a;用于解锁手机、自动标记照片、监控安全系统&#xff0c;以及门禁系统、考勤系统、人脸识别防盗门、电子护照及身份证等。此外&a…
阅读更多...
音频入门（二）：音频数据增强

音频入门（二）：音频数据增强

本文介绍了一些常见的音频数据增强方法&#xff0c;并给出了代码实现。 目录 一、简介 二、代码 1. 安装必要的库 2. 代码 3. 各函数的介绍 4. 使用方法 参考&#xff1a; 一、简介 音频数据增强是机器学习和深度学习领域中用于改善模型性能和泛化能力的技术。 使用数据…
阅读更多...
25/1/21 算法笔记＜ROS2＞ 话题通信接口

25/1/21 算法笔记＜ROS2＞ 话题通信接口

在 ROS 2 中&#xff0c;通信接口是节点之间进行数据交换的核心机制。ROS 2 提供了多种通信接口&#xff0c;包括话题&#xff08;Topic&#xff09;、服务&#xff08;Service&#xff09;、动作&#xff08;Action&#xff09;和参数&#xff08;Parameter&#xff09;。每种…
阅读更多...
最新文章

Copyright @ 2022~2023