Pulsar初入门（一）

刺骨的言语ヽ痛彻心扉 2021-12-14 07:23 428阅读 0赞

**目录**

简介:

架构：

一、Messaging Concepts(消息概念)

Producer

模式：

压缩：

Batching

Consumer

模式：

client：

ack

死信主题：

topics：

namespace

订阅模式：

独占：

故障转移：

共享：

Key\_shared

多topic订阅

分区主题：

路由模式：

订购保证：

messages保留和到期：

messages去重：

生产者幂等性：

重复数据删除和有效一次的语义：

--------------------

参考--Apache-Pulsar官网---[http://pulsar.apache.org/][http_pulsar.apache.org]

\-选择pulsar而不是Kafka的7个原因---[https://kafkaesque.io/7-reasons-we-choose-apache-pulsar-over-apache-kafka/][https_kafkaesque.io_7-reasons-we-choose-apache-pulsar-over-apache-kafka]

\-选择pulsar而不是Kafka的7个原因--infoQ中文版--[https://baijiahao.baidu.com/s?id=1634132982881230076&wfr=spider&for=pc][https_baijiahao.baidu.com_s_id_1634132982881230076_wfr_spider_for_pc]

\-推荐阅读-----CSDN网友/Pulsar官网文档翻译计划参与者--[稀有气体][Link 1]\--[Kafka的时代已经过去了，未来是Pulsar的吗？][Kafka_Pulsar]等系列文章

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW94aW4xMDI0_size_16_color_FFFFFF_t_70][]

# 简介: #

Apache Pulsar是一个开源的分布式的pub-sub消息系统，最初是雅虎创建的，现在是Apache Software Foundation的一部分。

关于pulsar：

> 1.pulsar函数，使用开发人员友好的API部署轻量级计算逻辑，无需运行自己的流处理引擎
> 
> 2.低延迟，耐用-设计用于大规模的低延迟发布（<5ms），具有强大的耐用性保障
> 
> 3.持久存储，基于Apache Bookkeeper的持久消息存储。提供写和读操作之间的IO隔离。
> 
> 4.生产中证明，Pulsar在雅虎生产超过3年，每秒百万条消息设计百万个主题
> 
> 5.地域复制，跨地理位置，异地数据中心复制支持
> 
> 6.客户端库。灵活的消息传递模型，支持java，c++，py，Go
> 
> 7.水平扩展，水平无缝扩展到数百万个节点
> 
> 8多租户，原生支持多租户，支持隔离，验证，授权和配额
> 
> 9可操作性，REST Admin API ，用于配置管理，工具和监视。可以部署在本地和k8s上。
> 
> tips：
> 
> >多种订阅模式（独占，共享和failover）--默认独占
> 
> >broker无状态
> 
> >数据老化时，分层存储可以将数据从hot / warm存储卸载到cold / longterm存储（s3，GCS）

# 架构： #

![èå²æç»æå¾][pulsar-system-architecture.png]

## 一、Messaging Concepts(消息概念) ##

pulsar基于pub-sub模式，类似于Kafka(支持点对点和pub-sub)和其它的消息系统。

Messages是Pulsar的基本”单位“，它们是生产者向主题发布的内容以及消费者随后从主题中消费多的内容(并在消息处理时确认)。Messages类似于邮政系统中的信件。

<table>  Pulsar的消息构成  
 <tbody> 
  <tr> 
   <td>Component</td> 
   <td>Purpose</td> 
  </tr> 
  <tr> 
   <td>value&nbsp; /&nbsp; data payload</td> 
   <td>消息携带的数据。所有Pulsar消息都携带原始字节，尽管消息数据也可以符合数据的Schemas</td> 
  </tr> 
  <tr> 
   <td>key</td> 
   <td>可以选择使用key标记message，这对 主题压缩 等操作非常有用</td> 
  </tr> 
  <tr> 
   <td>properties</td> 
   <td>用户定义属性的可选 key/value映射</td> 
  </tr> 
  <tr> 
   <td>Producer Name</td> 
   <td>生产者名称(生产者自动给出默认名称，但是你可以给出自己的名称)</td> 
  </tr> 
  <tr> 
   <td>Sequence ID</td> 
   <td>每个Pulsar消息属于其主题上的有序队列。消息的序列是该序列中的排序</td> 
  </tr> 
  <tr> 
   <td>Publish time</td> 
   <td>messages的发布时间戳（由生产者自动应用）</td> 
  </tr> 
  <tr> 
   <td>event time</td> 
   <td>应用程序可以附加到表示发生事件的消息的可选时间戳，例如处理消息时。如果没有明确设置，则event time为0</td> 
  </tr> 
 </tbody> 
</table>

## Producer ##

### 模式： ###

生产者可以 同步 异步 向broker发送messages

> 同步：失败重试

> 异步：阻塞队列（可配置）已满，则生产者调用时被阻止或立即失败，具体取决与传递给生产者的参数。

### 压缩： ###

可以在messages运输过程中压缩，以节省带宽。目前支持：  
LZ4

ZLIB

ZSTD

SNAPPY

### Batching ###

如果启用了批处理，则producer在单个请求中累计并发送一批消息。批处理大小由最大消息数和发布延迟定义。

## Consumer ##

订阅后才能接收到消息

### 模式： ###

可以 同步 或 异步 从broker接收消息

> 同步：在消息可用之前，将阻止同步接收

> 异步：异步接收将立即返回一个未来值 - -例如CompleetableFuture，在java中，一旦新消息可用，他就会完成。

### client： ###

客户端为消费者提供了监听器实现，java client提供的是MessageListener接口，在该接口中，只要received收到新消息，就会调用该方法。

### ack ###

**1**.消费者成功接收到消息时，向broker发送确认请求，以便broker丢弃该消息，否则它将存储消息。

可以逐个累积地确认消息，通过逐个累加确认，消费者只需要确认收到的最后一条消息。直到（包括）提供的消息的流中的所有的消息都不会被重新传递该消费者。

> 累计确认不能与 --共享订阅模式-- 一起使用，因为共享订阅模式多个消费者可以访问相同的订阅。

在共享订阅模式下，可以单独确认消息。

**2.**消费者不成功消费时，向代理发送否定确认，broker将重发messages。

消息可以逐个或者累积的被确认否认，这取决有消费模式。

在独占和failover模式下，消费者只会对它们收到的最后一条消息进行否定确认。

在共享和key\_Shared订阅模式中，您可以单独地否定确认消息。

**3**.确认超时

如果您希望broker自动重发消息，可以采用未确认消息自动传递机制，客户端在acktime内跟踪未确认地消息，redeliver unacknowledged messages在指定确认超时时自动向broker发送请求。

> tips：  
> 在确认消息超时之前使用否定确认。否定确认控制跟精确地重新传递单个消息，并在消息处理时间超过确认超时时避免无效重传。

### 死信主题： ###

即消费者无法消费地某些消息，将存储多的地方，其它消息系统，如RabbitMQ支持--通过broker存入，而Kafka没有相关概念，可以自行封装以支持。

// java API Demo
    // Apache pulsar
    // http://pulsar.apache.org/docs/en/concepts-messaging/
    Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
                  .topic(topic)
                  .subscriptionName("my-subscription")
                  .subscriptionType(SubscriptionType.Shared)
                  .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(maxRedeliveryCount)
                        .build())
                  .subscribe();

死信主题取决于messagess是否重传。你需要确认消息重新传递方法：否定确认或确认超时。

再确认超时之前使用否定确认。

> 死信主题仅在共享订阅模式下支持

### topics： ###

{persistent|non-persistent}://tenant/namespace/topic

<table>  Pulsar topic构成  
 <tbody> 
  <tr> 
   <td>Topic name component</td> 
   <td>Description</td> 
  </tr> 
  <tr> 
   <td>persistent&nbsp; /&nbsp; non-persistent</td> 
   <td>标识了topic的类型。Pulsar支持两种主题：持久性和非持久性(默认持久性主题)，持久性主题存在磁盘，而非持久性topic则不会存储到磁盘</td> 
  </tr> 
  <tr> 
   <td>tenant</td> 
   <td>实例中的topic的租户。租户是pulsar对多租户支持的重要组成，可以分散在集群中</td> 
  </tr> 
  <tr> 
   <td>namespace</td> 
   <td>topic的管理单元，作为相关主题的分组机制。大多数主题配置都在命名空间级别执行，每个租户可以有多个namespace</td> 
  </tr> 
  <tr> 
   <td>topic</td> 
   <td>yopic的最后部分。topic的名字是自由格式的，在Pulsar实例中没有特殊含义</td> 
  </tr> 
 </tbody> 
</table>

> producer写入不存在的主题时 了会在提供的命名空间下自动创建该主题

### namespace ###

namespace是租户中多的逻辑命名法。租户可以通过管理API创建多个namespace，例如，具有不同应用程序的租户可以未每个应用程序创建单独的namespace。namespace允许应用程序创建和管理主题层次结构。主题my-tenant/app1是应用程序namespace app1的my-tenant。可以在命名空间下创建任意数量的主题。

## 订阅模式： ##

订阅模式是一种命名规则配置工具，用于确定如何将消息传递给使用者。Pulsar有三种可用的订阅模式：独占，共享和failover。这些模式如下图所示。

![è®¢éæ¨¡å¼][pulsar-subscription-modes.png]

Apache pulsar官网示例图片--[http://pulsar.apache.org/docs/en/concepts-messaging/][http_pulsar.apache.org_docs_en_concepts-messaging]

### 独占： ###

在独占模式下，只允许一个消费者附加到订阅。如果多个消费尝试使用相同的订阅订阅主题，则消费者会收到错误。

上面的示例图片中---只允许Consumer-A使用消息。

> 独占模式是默认订阅模式

### 故障转移：    ###

在failover模式下，多个使用者可以附加到同一个订阅。消费者按照消费者姓名进行词汇排序，第一个消费者最初将是唯一接收到消息的消费者。此消费者称为主消费者。

当主消费者断开连接时，所有（非确认的和后续的）消息将被传递给下一个消费者。

在上图中，Consumer-C-1是主消费者，而如果Consumer-C-1断开连接，Consumer-C-2将成为接收消息的下一个消费者。

![æéè½¬ç§»è®¢é][pulsar-failover-subscriptions.png]

### 共享：    ###

在共享或round robin模式下，多个消费者可以附加到同一订阅。消息以消费者的循环分发形式传递，并且任何给定的消息仅传递给一个消费者。当消费者断开连接时，发送给它且未被确认的所有消息将被重新安排以便发送给剩余的消费者。

> 局限性：  
> 不保证消息有序
> 
> 不能再此模式下使用累加确认

### Key\_shared ###

在Key\_shared模式下，多个消费者可以附加到同一个订阅。消息在消费者的分发中传递，并且具有相同key或者相同排序key的

消息仅被传递给一个消费者。无论重新传递多少次消息，当链接断开导致消费者更改某些消息的密钥，他都会传递给同一个消费者。

> 局限性：
> 
> 测试功能（2.4.0）--可在broker.config中禁用它
> 
> 需要为消息指定密钥或orderingKey
> 
> 不能再此模式下使用累计确认

![Key\_Sharedè®¢é][Key_Shared]

### 多topic订阅 ###

消费者订阅Pulsar主题时，默认情况下它订阅一个特定主题，从pulsar1.23.0开始，可以支持多个topic订阅。可以通过两种方式来定义主题列表。

> 1，正则 例如 persistent://public/default/finance-.\* ---主题必须位于同一个namespace下
> 
> 2，明确给出topic list

> 缺陷：  
> 没有订购保证 --如果生产条件要求严苛，则不要使用多topic订阅模式

// java api Demo
    // Apache Pulsar
    // http://pulsar.apache.org/docs/en/concepts-messaging/
    import java.util.regex.Pattern;
    
    import org.apache.pulsar.client.api.Consumer;
    import org.apache.pulsar.client.api.PulsarClient;
    
    PulsarClient pulsarClient = // Instantiate Pulsar client object
    
    // Subscribe to all topics in a namespace
    Pattern allTopicsInNamespace = Pattern.compile("persistent://public/default/.*");
    Consumer<byte[]> allTopicsConsumer = pulsarClient.newConsumer()
                    .topicsPattern(allTopicsInNamespace)
                    .subscriptionName("subscription-1")
                    .subscribe();
    
    // Subscribe to a subsets of topics in a namespace, based on regex
    Pattern someTopicsInNamespace = Pattern.compile("persistent://public/default/foo.*");
    Consumer<byte[]> someTopicsConsumer = pulsarClient.newConsumer()
                    .topicsPattern(someTopicsInNamespace)
                    .subscriptionName("subscription-1")
                    .subscribe();

### 分区主题： ###

普通主题只能由单个broker提供，限制了最大吞吐量。分区主题由多个broker处理的特殊主题模型，允许更高的吞吐量。

\--分区主题实际上实现为N个内部主题，其中N时分区数。将消息发布到分区主题时，每条消息会路由到多个代理，Pulsar自动会处理各个broker中的分区分配。

![partitioning.png][]

### 路由模式：    ###

发布到分布主题，必须指定路由模式。默认三个路由模式，默认轮询-和Kafka类似。

<table> 
 <tbody> 
  <tr> 
   <td>模式</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>RoundRobinPartition</td> 
   <td>如果未提供key，则以轮询方式往各分区上面发布消息，以实现最大吞吐量 --默认模式</td> 
  </tr> 
  <tr> 
   <td>SinglePartition</td> 
   <td>如果未提供key，则生产者随机选择一个分区并将所有消息发布到此分区。如果指定了key，将对key进行散列（默认javaStringHash=多客户端推荐Murmur3_32Hash）并将消息分配给特定的分区</td> 
  </tr> 
  <tr> 
   <td>CustomPartition</td> 
   <td>使用将调用的自定义消息路由器实现来特定消息的分区。用户在在java clent端实现MessageRouter接口来实现自定义路由模式。</td> 
  </tr> 
 </tbody> 
</table>

### 订购保证：    ###

消息的排序于消息的路由模式和消息的key有关。

如果有key，messages被路由到指定的分区，在使用SinglePartition或RoundRobinPartition模式。

<table> 
 <tbody> 
  <tr> 
   <td>订购保障</td> 
   <td>描述</td> 
   <td>路由模式和key</td> 
  </tr> 
  <tr> 
   <td> <p>Per-key-Partition</p> </td> 
   <td>具有相同message key的所有消息将按顺序排列并放置在同一个分区中。</td> 
   <td>使用SinglePartition或RoundRobinPartition模式，每个消息提供key</td> 
  </tr> 
  <tr> 
   <td>Pre-producer</td> 
   <td>来自同一Producer的所有消息都将按顺序排列</td> 
   <td>使用SinglePartition，并且不为每条消息提供key</td> 
  </tr> 
 </tbody> 
</table>

### messages保留和到期： ###

> 默认情况下，Pulsar消息代理：  
> 1.立即删除已经确认的消息
> 
> 2.持久性的将所有未确认的消息存储在消息积压中

但是pulsar，提供了其他的功能可以覆盖此默认行为。

> 1，消息保留，存储已经确认的消息
> 
> 2，消息到期，可以设置messages的TTL

### messages去重： ###

pulsar支持消息去重，可防止不必要的消息的重复。

![Pulsaræ¶æ¯éå¤æ°æ®å é¤][Pulsar]

该消息去重逻辑是由broker实现的。-见下面

### 生产者幂等性：    ###

消息重复数据的删除的另一种可用方法是确保每每条消息只生成一次。这种方法通常称为生产者幂等性。这种方法的缺点是它将消息多的重复数据删除的工作推到了应用程序。在Pulsar中，这是在代理级别处理的。

### 重复数据删除和有效一次的语义： ###

重复消息的删除使得Pulsar的成为理想的消息传递系统，可与流处理引擎(SPE)和其它系统结合使用，以寻求有效一次处理语义。不提供自动重复消息删除的消息系统需要借助SPE或其它系统来保证重复数据删除，这意味着严格的排序是以重复数据删除的应用程序的负担为代价的。使用pulsar，严格的订购保证不会产生应用级别的成本。

[http_pulsar.apache.org]: http://pulsar.apache.org/
[https_kafkaesque.io_7-reasons-we-choose-apache-pulsar-over-apache-kafka]: https://kafkaesque.io/7-reasons-we-choose-apache-pulsar-over-apache-kafka/
[https_baijiahao.baidu.com_s_id_1634132982881230076_wfr_spider_for_pc]: https://baijiahao.baidu.com/s?id=1634132982881230076&wfr=spider&for=pc
[Link 1]: https://blog.csdn.net/liyiming2017
[Kafka_Pulsar]: https://blog.csdn.net/liyiming2017/article/details/82875068
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW94aW4xMDI0_size_16_color_FFFFFF_t_70]: /images/20211214/6a1a0d8ee7284540bd9c69267117ed3e.png
[pulsar-system-architecture.png]: /images/20211214/938703d4c5b04c31b0a433a07760bce4.png
[pulsar-subscription-modes.png]: /images/20211214/f9ab643fe8e54f5c8ff31e37f6e4f82f.png
[http_pulsar.apache.org_docs_en_concepts-messaging]: http://pulsar.apache.org/docs/en/concepts-messaging/
[pulsar-failover-subscriptions.png]: /images/20211214/15ee0f5c5be8414eb13fa8cfd6195cbd.png
[Key_Shared]: /images/20211214/75c2f4531a034ef79f8c021626616338.png
[partitioning.png]: /images/20211214/533dc0f60e7549bfb465f62d4d4d794c.png
[Pulsar]: /images/20211214/c8e87d26a5e347f09f091484075bb19b.png