Pulsar

  Pulsar 是新一代云原生分布式消息引擎，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户隔离、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延迟及高可扩展性等流数据存储特性，是一个pub-sub (发布-订阅)模型的消息队列系统。可以很好的服务大数据生态，及轻松构建基于实时数据的分析和应用。

一、主要特性

• 支持多集群，多机房跨区域消息复制，异地多活，可跨机房在集群间无缝地完成消息复制
• 采用计算和存储分离架构，支持水平动态扩容，可以支持百万级别Topic数量的扩展，同时还能一直保持良好的性能。(Kafka计算存储未分离，扩容需要占用很多系统资源在节点间拷贝数据，而Plusar完全不用)
• 租户隔离，Pulsar 通过租户和命名空间这两个关键概念支持多租户，在topic的 URL 中附加命名空间达到了租户隔离的效果
• 高吞吐，支持百万级消息主题，日处理超过1000亿条消息
• 低延迟，非常低的消息发布延迟和端到端的延迟(< 5ms)
• 支持多种Topic订阅模式：独占（exclusive）、共享（shared）、灾备（failover）、Key_Shared
• 支持定时、延时消息，Pulsar可以给未被确认的消息设置存活时长（TTL）
• 支持消息重试与死信机制（防异常消息导致消费阻塞）
• 持久化机制，Plusar的持久化机制构建在Apache BookKeeper之上，提供了写与读之前的IO隔离，通过 Apache BookKeeper 提供的持久化消息存储机制保证消息传递
• 部署方式的多样化，既可以运行在裸机，也支持目前例如Docker、K8S的一些容器化方案以及不同的云厂商，同时在本地开发时也只需要一行命令即可启动整个环境

  Plusar将多地域/可用区的复制作为首要特性支持。用户只需配置好可用区，源区域中生产了消息，会自动在目标区域中创建相同的主题并复制消息。当某一个可用区挂掉或者发生网络阻塞，plusar会在之后不断的重试。

二、应用场景

• 实时数据分析
    数据快速流动产生价值，传统数据分析框架大部分基于批量计算模型，无法做到实时的数据分析。Pulsar 与流式计算相结合，Pulsar 作为实时分析的数据源和输出源，可支持模型训练、反作弊、广告计费等实时性要求较高的场景。

• 数据中转枢纽
    发布-订阅的模式可以根据不同的业务数据类型，将数据发布到不同的主题（topic）或同一份数据被多个业务订阅使用；构建应用系统和分析系统的数据中转枢纽，实现应用、分析解耦，增强系统可扩展性。

• 流式数据归档
    实时收集用户浏览页面、搜索及其他行为，利用 Pulsar 保证数据不重、数据可回放等特点，解决同步准确性问题，将数据流导入 AFS、HIVE 数据仓库系统，同步数据，快速支持业务监控、分析。

• 延迟消息传递
    用户能够在稍后的某个固定时间点或某一段固定时间后，消费对应的消息。典型应用场景：待支付订单的有效支付窗口、邮件的定时发送、会议的定时提醒。仅适用共享订阅的消费类型，独占订阅和故障转移订阅为确保FIFO特性，该功能不生效。

【功能的开启】
Broker配置中，delayedDeliveryEnabled开启
【实现原理】
1、共享模式订阅中会启动分发器中的消息延迟追踪器DelayedDeliveryTracker，检查每条消息的延迟消息参数
2、延迟消息被添加到延迟追踪器队列中
3、通过时间轮（Time Wheel approach）算法处理，到期后的消息重新被发送
4、最小触发时间deleyedDeliveryTickTimeMillis，默认为1s
5、独占订阅和故障转移订阅模式的消费者，正常消费延迟消息

三、架构设计

Pulsar系统整体架构图：

  一个pulsar实例可以由多个集群组成，集群间的消息数据可以进行复制。单个集群由以下部分组成：

• Producer：数据生成者，即发送消息的一方。生产者负责创建消息，将其投递到Pulsar中。
• Consumer：数据消费者，即接收消息的一方。消费者连接到 Pulsar 并接收消息，进行相应的业务处理。
• Broker：无状态的服务层，负责接收消息、传递消息、集群负载均衡等操作，Broker不会持久化保存元数据。
• BookKeeper：有状态的持久层，包含多个Bookie，负责持久化地存储消息。
• ZooKeeper：存储Pulsar、BookKeeper的元数据，集群配置等信息，负责集群间的协调(例如：Topic与Broker的关系)、服务发现等。

  从Pulsar的架构图上可以看出，Pulsar在架构设计上采用了计算与存储分离的模式，发布/订阅相关的计算逻辑在Broker上完成，而数据的持久化存储交由BookKeeper去实现。

Pulsar计算层与存储层交互示意图：

【Broker扩展】
  在Pulsar中Broker是无状态的，当需要支持更多的消费者或生产者时，可以简单地添加更多的Broker节点来满足业务需求。Pulsar支持自动的分区负载均衡，在Broker节点的资源使用率达到阈值时，会将负载迁移到负载较低的Broker节点，这个过程中分区也将在多个Broker节点中做平衡迁移，一些分区的所有权会转移到新的Broker节点。
【Bookie扩展】
  存储层的扩容，通过增加Bookie节点来实现。在BooKie扩容的阶段，由于分片机制，整个过程不会涉及到不必要的数据搬迁，即不需要将旧数据从现有存储节点重新复制到新存储节点。

四、Topic

  Topic即在生产者与消费者中传输消息的通道。消息可以以Topic为单位进行归类，生产者负责将消息发送到特定的Topic，而消费者指定特定的Topic进行消费。Topic中的每条消息，可以根据消费者的订阅需求，多次被使用，每个订阅对应一个消费者组（Consumer Group）,每个Topic可以有不同的消费组。

分区Topic（Topic-Partition）

  Pulsar的Topic可以分为非分区Topic和分区Topic。

  普通的Topic仅仅被保存在单个Broker中，这限制了Topic的最大吞吐量。分区Topic是一种特殊类型的主题，支持被多个Broker处理，从而实现更高的吞吐量。

  针对一个Topic，可以设置多个Topic分区来提高Topic的吞吐量。每个Topic Partition由Pulsar分配给某个Broker，该Broker称为该Topic Partition的所有者。生成者和消费者会与每个Topic分区的Broker创建链接，发送消息并消费消息。

  客户端的Producer和Consumer在初始化的时候，都会与每一个Topic-Partition创建链接，并且会监听是否有新的Partition，以创建新的链接。

  如下图所示，Topic1有Partition1、Partition2、Partition3、Partition4、Partition5五个分区，Partition1和Partition4由Broker1处理，Partition2和Partition5由Broker2处理，Partition3由Broker3处理。

非持久Topic

  默认情况下，Pulsar会保存所有没确认的消息到BookKeeper中。持久Topic的消息在Broker重启或者Consumer出现问题时保存下来。

  除了持久Topic，Pulsar也支持非持久Topic。这些Topic的消息只存在于内存中，不会存储到磁盘。

  因为Broker不会对消息进行持久化存储，当Producer将消息发送到Broker时，Broker可以立即将ack返回给Producer，所以非持久Topic的消息传递会比持久Topic的消息传递更快一些。相对的，当Broker因为一些原因宕机、重启后，非持久Topic的消息都会消失，订阅者将无法收到这些消息。

重试Topic

  由于业务逻辑处理出现异常，消息一般需要被重新消费。Pulsar支持生产者同时将消息发送到普通的Topic和重试Topic，并指定允许延时和最大重试次数。当配置了允许消费者自动重试时，如果消息没有被消费成功，会被保存到重试Topic中，并在指定延时时间后，重新被消费。

死信Topic

  当Consumer消费消息出错时，可以通过配置重试Topic对消息进行重试，但是，如果当消息超过了最大的重试次数仍处理失败时，该怎么办呢？Pulsar提供了死信Topic，通过配置deadLetterTopic，当消息达到最大重试次数的时候，Pulsar会将消息推送到死信Topic中进行保存。

五、通用的消费模型

消费模型一般包括以下 3 个方面：

• 消息消费：如何发送和消费消息
• 消息确认（ACK）：如何确认消息
• 消息保存（Retention）：消息保留时间，触发消息删除的原因以及怎样删除

  【Pulsar消息确认】当使用分布式消息系统时，可能会发生故障。比如在消费者从消息系统中的主题消费消息的过程中，消费者和Broker都可能发生错误。消息确认（ACK）的目的就是保证当发生这样的故障后，消费者能够从上一次停止的地方恢复消费，保证既不会丢失消息，也不会重复处理已经ACK的消息。
  【Pulsar消息保留】在消息被确认后，Pulsar 的 Broker 会更新对应的游标。当 Topic 里面中的一条消息，被所有的订阅都确认 ack 后，才能删除这条消息。Pulsar 还允许通过设置保留时间，将消息保留更长时间，即使所有订阅已经确认消费了它们。

队列模型

  队列模型主要是采用无序或者共享的方式来消费消息。通过队列模型，多个消费者可以从单个管道中接收消息；当一条消息从队列发送出来后，多个消费者中的只有一个（任何一个都有可能）接收和消费这条消息。消息系统的具体实现决定了最终哪个消费者实际接收到消息。

  队列模型通常与无状态应用程序一起结合使用。无状态应用程序不关心排序，但它们确实需要能够确认（ack）或删除单条消息，以及尽可能地扩展消费并行性的能力。典型的基于队列模型的消息系统包括 RabbitMQ 和 RocketMQ。

流模型

  流模型要求消费的消息严格排序或独占消息消费。对于一个管道，使用流式模型，始终只会有一个消费者使用和消费消息。消费者按照消息写入管道的确切顺序接收从管道发送的消息。

  流模型通常与有状态应用程序相关联。有状态的应用程序更加关注消息的顺序及其状态。消息的消费顺序决定了有状态应用程序的状态。消息的顺序将影响应用程序处理逻辑的正确性。

Pulsar既支持队列模型，也支持流模型

  Pulsar抽象出了统一的消费模型: producer-topic-subscription-consumer，通过这种模型，将队列模型和流模型结合在了一起，提供了统一的API接口。这种模型，既不会影响消息系统的性能，也不会带来额外的开销，同时还为用户提供了更多灵活性，方便用户根据自己的实际场景来使用消息系统。

六、订阅模式

  pulsar的主题订阅模式包括四种：独占（exclusive）、共享（shared）、灾备（failover）、key共享（key_shared）。

独占模式（流模型）

  Pulsar 默认的消息订阅模式，topic只能被一个消费者订阅，如果多于一个消费者以同样方式去订阅主题，消费者将会收到错误。

  如下图，只有Consumer A-0可以消费数据。

灾备模式（流模型）

  故障转移模式，多个消费者可以订阅同一个topic，消费者按消息者名称的字典序排列。第一个消费者被初始化为唯一接收消息的消费者。这个消费者被称为主消费者（master consumer）。当主消费者断开时，所有的消息（未被确认和后续进入的）将会被分发给下一个消费者。

分区Topic：Broker会按照消费者的优先级和消费名的顺序对消费者进行排序，将Topic均匀地分配给优先级最高的消费者。
非分区Topic：Broker会根据消费者订阅的非分区Topic的时间顺序选择消费者。

  如下图，Consumer-B-0是Master Consumer。当Consumer-B-0发生问题与Broker断开连接时，Consumer-B-1将成为下一个Master Consumer来消费数据。

共享模式（队列模型）

  多个消费者可以订阅同一个topic，无法保证消息的顺序。消息通过 round robin即轮询机制分发给不同的消费者，并且每个消息仅会被分发给一个消费者。当消费者断开连接，所有被发送给消费者但没有被确认的消息将被重新处理，分发给其它存活的消费者。

  多如下图, Consumer-C-1、Consumer-C-2、Consumer-C-3都可以订阅 Topic消费数据。

key共享模式（队列模型）

  Key_Shared 模式是共享模式的一种，不同的是它按 key 对消息做投递，具有相同 key 的消息会被投递到同一个Consumer。

七、Pulsar与其他消息队列对比

参考资料

颠覆Kafka的统治，新一代云原生消息系统Pulsar震撼来袭！