消息队列篇--原理篇--Pulsar（Namespace，BookKeeper，雷同Kafka以致更好的消息队列）

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）、生产者示例

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

复制代码

（2）、消耗者示例

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

复制代码

6、Pulsar总结

Apache Pulsar是一个功能强大、架构机动的消息体系，特殊恰当大规模分布式体系的及时数据处理惩罚和异步通讯。它的分层架构、多租户支持、恒久化和划一性保障、机动的消息模子等特点，使其在性能、可靠性和可扩展性方面体现精彩。Pulsar还拥有丰富的生态体系，支持与其他工具和服务集成，实用于多种应用场景。
乘风破浪会偶尔，直挂云帆济沧海！！！