如何看懂Apache Pulsar？（究极缝合）

绝地灬酷狼 2022-09-05 01:46 308阅读 0赞

# Apache Pulsar #

官方文档直达，快速移步，本文基于`pulsar2.8`版本

[apache/pulsar: Apache Pulsar - distributed pub-sub messaging system (github.com)][apache_pulsar_ Apache Pulsar - distributed pub-sub messaging system _github.com]

[Set up a standalone Pulsar locally · Apache Pulsar][Set up a standalone Pulsar locally _ Apache Pulsar]

## 简介 ##

Pulsar 是一个用于服务器到服务器的消息系统，具有多租户、高性能等优势。 Pulsar 最初由 [Yahoo][] 开发，目前由 [Apache 软件基金会][Apache]管理。

Pulsar 的关键特性如下：

*  Pulsar 的单个实例原生支持多个集群，可[跨机房][Link 1]在集群间无缝地完成消息复制。
 *  极低的发布延迟和端到端延迟。
 *  可无缝扩展到超过一百万个 topic。
 *  简单的[客户端 API][API]，支持 [Java][]、[Go][]、[Python][] 和 [C++][C]。
 *  支持多种[ topic 订阅模式][topic]（[独占订阅][Link 2]、[共享订阅][Link 3]、[故障转移订阅][Link 4]）。
 *  通过 [Apache BookKeeper][] 提供的[持久化消息存储机制][Link 5]保证消息传递 。
 *   *  由轻量级的 serverless 计算框架 [Pulsar Functions][] 实现流原生的数据处理。
 *  基于 Pulsar Functions 的 serverless connector 框架 [Pulsar IO][] 使得数据更易移入、移出 Apache Pulsar。
 *  [分层式存储][Link 6]可在数据陈旧时，将数据从热存储卸载到冷/长期存储（如S3、GCS）中。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center]

## 集群架构 ##

单个 `Pulsar` 集群由以下三部分组成：

*  一个或者多个 `broker` 负责处理和负载均衡 `producer` 发出的消息，并将这些消息分派给 `consumer`；`Broker` 与 `Pulsar` 配置存储交互来处理相应的任务，并将消息存储在 `BookKeeper` 实例中（又称 `bookies`）；`Broker` 依赖 `ZooKeeper` 集群处理特定的任务，等等。
 *  包含一个或多个 `bookie`的 `BookKeeper` 集群负责消息的持久化存储。
 *  一个`Zookeeper`集群，用来处理多个`Pulsar`集群之间的协调任务。`ZooKeeper` 只用于为 `broker` 和 `bookie` 存储元数据。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 1]

### Brokers ###

`Pulsar`的`broker`是一个无状态组件, 主要负责运行另外的两个组件：

*  `http`服务器，可用于接收`Rest API`请求，并负责生产者连接生产消息，消费者连接消费消息。
 *  一个调度分发器，它是异步的`TCP`服务器，通过自定义二进制协议应用于所有相关的数据传输。

### Apache BookKeeper ###

`Pulsar`用`Apache BookKeeper`作为持久化存储。 `BookKeeper`是一个分布式的预写日志（`WAL`）系统，`BookKeeper`节点（`bookie`）存储消息与游标。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 2]

以`pulsar-manager`的某个`topic`为例：

![pulsar manager topic][]

### Ledgers ###

`Ledger`**是一个只追加的数据结构，并且只有一个写入器**，这个写入器负责多个`BookKeeper`存储节点（就是`Bookies`）的写入。 `Ledger`的条目会被复制到多个`bookies`。`Ledgers`本身有着非常简单的语义：

*  **Pulsar Broker可以创建ledeger，添加内容到ledger和关闭ledger**。
 *  当一个`ledger`被关闭后，除非明确的要写数据或者是因为写入器挂掉导致`ledger`关闭，这个`ledger`只会以只读模式打开。
 *  最后，当`ledger`中的条目不再有用的时候，整个`legder`可以被删除（`ledger`分布是跨`Bookies`的）。

#### Ledger读一致性 ####

`BookKeeper`的主要优势在于他能在有系统故障时保证读的一致性。 由于`Ledger`只能被一个进程写入（之前提的写入器进程），这样这个进程在写入时不会有冲突，从而写入会非常高效。 在一次故障之后，`ledger`会启动一个恢复进程来确定`ledger`的最终状态并确认最后提交到日志的是哪一个条目。 在这之后，能保证所有的`ledger`读进程读取到相同的内容。

#### Managed ledgers ####

managed ledger即消息流的抽象，有一个写入器进程不断在流结尾添加消息，并且有多个`cursors` 消费这个流，每个`cursor`有自己的消费位置。

![cursor][]

### 日志存储 ###

在`BookKeeper`中, `journal files`包含`BookKeeper transaction logs`. 在更新到`ledger`之前，`bookie`需要确保描述这个更新的事务被写到持久（非易失）存储上面。 在`bookie`启动和旧的日志文件大小达到上限（由`journalMaxSizeMB`参数配置）的时候，新的日志文件会被创建。

## Kafka对比 ##

### 组件 ###

`Kafka` 采用单片架构模型，将服务与存储相结合。而`Pulsar`则采用了**多层架构**，可以在单独的层内进行管理。`Pulsar`中的`broker`在一个层上进行计算，而 `bookie` 则在另一个层上管理有状态存储。

### 存储 ###

`Pulsar`的多层架构影响到了其存储数据的方式。`Pulsar`将`topic` 分区划分为分片，然后将这些分片存储在 `Apache BookKeeper`的存储节点上，以提高性能、可伸缩性和可用性。

`Pulsar`对日志进行分段，从而避免了拷贝大块的日志。通过`BookKeeper`， `Pulsar`将日志分段分散到多台不同的服务器上。也就是说，日志不会保存在单台服务器上，任何一台服务器都不会成为整个系统的瓶颈。这使故障处理和扩容更加简单，只需要加入新的服务器，而无需进行再均衡处理。

`Pulsar`消息只有被所有订阅消费后才会删除，不会丢失数据。`Kafka`根据设置的数据保留过期时间，过期后删除。同样，`Pulsar`也支持设置保留时间（`TTL`）。

### 无状态 ###

`Kafka`不是无状态的，每个 `broker` 都包含了分区的所有日志，如果一个 `broker`宕机，不是所有`broker`都可以接替它的工作。如果工作负载太高，也不能随意添加新的 `broker` 来分担，而是必须与持有其分区副本的 `broker` 进行状态同步。

在 `Pulsar` 架构中，`broker`是无状态的。但是完全无状态的系统无法持久化消息，所以`Pulsar` 不是依靠 `broker` 来实现消息持久化的。在 `Pulsar` 架构中，数据的分发和保存是相互独立的。`broker`从生产者接收数据，然后将数据发送给消费者，但数据保存在 `BookKeeper` 中。

`Pulsar` 的 `broker`是无状态的，所以如果工作负载很高，可以直接添加新的 `broker`，快速接管工作负载。

### Exactly-Once 处理 ###

目前，`Pulsar` 通过 [broker 端去重][broker]支持`exactly-once producer`。

## 生产者 ##

生产者连接到主题并生产消息到`pulsar`的`broker`上。

`pulsar`的`broker`处理这条消息。

### 发送模式 ###

`Producer`可以以同步(`sync`) 或 异步(`async`) 的方式发布消息到`broker`。

*  同步发送到`broker`，如果没`broker`没有返回`ack`，则该条消息视为发送失败。
 *  `Producer`将把消息放于阻塞队列中，并立即返回 然后，客户端将在后台将消息发送给 `broker`。 如果队列已满(最大大小可配置)，则调用 `API` 时，`producer`可能会立即被阻止或失败，具体取决于传递给`producer` 的参数。

### 访问模式 ###

<table> 
 <thead> 
  <tr> 
   <th>访问模式</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Shared</td> 
   <td>Multiple producers can publish on a topic.This is the <strong>default</strong> setting.</td> 
  </tr> 
  <tr> 
   <td>Exclusive</td> 
   <td>仅有一个生产者可以在主题上发布消息。如果已经有一个生产者连接了此主题，其他试图在此主题上发布消息的生产者会马上得到错误信息。如果旧的生产者发生了与broker的网络分区，旧的生产者被驱逐后接着会有一个新的生产者被选举出来成为下一个排他性生产者。</td> 
  </tr> 
  <tr> 
   <td>WaitForExclusive</td> 
   <td>如果已经有一个生产者连接了主题，生产者创建过程被挂起(而不是超时) 直到这个生产者获得了 <code>Exclusive</code> 访问权限。成功成为独家生产者的生产者被视为领导者。 因此，如果你打算为自己的应用实现领导者选举方案，你可以使用这种访问模式。</td> 
  </tr> 
 </tbody> 
</table>

### 批量处理 ###

当批量处理启用时，`producer` 会在单个请求中积累并发送一批消息。 批量处理的量大小由最大消息数和最大发布延迟定义。 因此，积压数量是分批处理的总数，而不是信息总数。

在 `Pulsar` 中，批次被跟踪并存储为单个单元，而不是单个消息。 `Consumer` 将批量处理的消息拆分成单个消息。 但即使启用了批量处理，也始终将计划中的消息(通过 `deliverAt` 或者 `deliverAfter` 进行配置) 作为单个消息发送。

一般来说，当 `consumer` 确认了一个批的所有消息，该批才会被认定为确认。 这意味着当发生不可预料的失败、否定的确认（`negative acknowledgements`）或确认超时，都可能导致批中的所有消息都被重新发送，即使其中一些消息已经被确认了。

维护批量索引的确认状态并跟踪每批索引的确认状态，以避免向 `consumer`发送已确认的消息。 当某一批消息的所有索引都被确认时，该批消息将被删除。

默认情况下，批量索引是默认关闭的（`acknowledgmentAtBatchIndexLevelEnabled=false`），可以在`broker`配置中开启。启用批量索引确认将会导致更多内存开销

### 分块 ###

启用分块（`chunking`），需要注意下面说明：

*  批量处理不能和分块同时启用
 *  分块只支持`persisted topic`
 *  分块只支持`exclusive`和`failover`subscription modes.

当启用分块(`chunking`) 时(`chunkingEnabled=true`) ，如果消息大小大于允许的最大发布有效载荷大小，则 `producer` 将原始消息分割成分块的消息，并将它们与块状的元数据一起单独和按顺序发布到`broker`。 在 `broker` 中，分块的消息将和普通的消息以相同的方式存储在 `Managed Ledger` 上。 唯一的区别是，`consumer`需要缓冲分块消息，并在收集完所有分块消息后将其合并成真正的消息。 `Managed Ledger`上的分块消息可以和普通消息交织在一起。 如果 `producer` 未能发布消息的所有分块，则当 `consumer` 未能在过期时间(`expire time`) 内接收所有分块时，`consumer` 可以过期未完成的分块。 默认情况下，过期时间设置为1小时。

`Consumer`会缓存收到的块状消息，直到收到消息的所有分块为止。 然后 `consumer` 将分块的消息拼接在一起，并将它们放入接收器队列中。 客户端从接收器队列中消费消息。 一旦`consumer`使用整个大消息并确认，`consumer` 就会在内部发送与该大消息关联的所有分块消息的确认。 You can set the `maxPendingChunkedMessage` parameter on the consumer. 当达到阈值时，`consumer` 通过静默确认未分块的消息或通过将其标记为未确认，要求 `broker`稍后重新发送这些消息。

分块为例：

#### 处理一个 producer 和一个订阅 consumer 的分块消息 ####

如下图所示，当生产者向主题发送一批大的分块消息和普通的非分块消息时。 假设生产者发送的消息为 M1，M1 有三个分块 M1-C1，M1-C2 和 M1-C3。 这个`broker`在其管理的`ledger`里面保存所有的三个块消息，然后以相同的顺序分发给消费者（独占/灾备模式）。 消费者将在内存缓存所有的块消息，直到收到所有的消息块。将这些消息合并成为原始的消息M1，发送给处理进程。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 3]

#### 多个生产者和一个生产者处理块消息。 ####

当多个生产者发布块消息到单个主题，这个 `Broker`在同一个 `Ledger`里面保存来自不同生产者的所有块消息。 如下所示，生产者1发布的消息 M1，M1 由 M1-C1, M1-C2 和 M1-C3 三个块组成。 生产者2发布的消息 M2，M2 由 M2-C1, M2-C2 和 M2-C3 三个块组成。 这些特定消息的所有分块是顺序排列的，但是其在`ledger` 里面可能不是连续的。 这种方式会给消费者带来一定的内存负担。因为消费者会为每个大消息在内存开辟一块缓冲区，以便将所有的块消息合并为原始的大消息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 4]

## 消费者 ##

`Consumer` 向 `broker` 发送消息流获取申请（[flow permit request][]）以获取消息。 在`Consumer`端有一个队列，用于接收从 `broker`推送来的消息。 你能够通过[`receiverQueueSize`][receiverQueueSize]参数配置队列的长度 (队列的默认长度是`1000`) 每当 `consumer.receive()` 被调用一次，就从缓冲区（`buffer`）获取一条消息。

### 接收模式 ###

可以通过同步(`sync`) 或者异步(`async`)的方式从`brokers`接受消息。

<table> 
 <thead> 
  <tr> 
   <th align="left">发送模式</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">同步接收</td> 
   <td align="left">同步模式，在收到消息之前都是被阻塞的。</td> 
  </tr> 
  <tr> 
   <td align="left">异步接收</td> 
   <td align="left">异步接收模式会立即返回一个 future 值（如 Java 中的 <a href="http://www.baeldung.com/java-completablefuture" rel="nofollow"><code>CompletableFuture</code></a>），一旦收到新的消息就立刻完成。</td> 
  </tr> 
 </tbody> 
</table>

### 监听 ###

客户端提供`listener`的实现接口给消费者，例如，`Java Client`提供了`MessageListener`接口。  
在这个接口中，一旦接受到新的消息，`received`方法将被调用。

MessageListener myMessageListener = (consumer, msg) -> { 
      try { 
          System.out.println("Message received: " + new String(msg.getData()));
          consumer.acknowledge(msg);
      } catch (Exception e) { 
          consumer.negativeAcknowledge(msg);
      }
    }
    
    Consumer consumer = client.newConsumer()
         .topic("my-topic")
         .subscriptionName("my-subscription")
         .messageListener(myMessageListener)
         .subscribe();

### 确认 ###

当消费者成功的消费了一条消息，这个消费者会发送一个确认信息给`broker`。 这个消息时是永久保存的，只有在收到订阅者消费成功的消息确认后才会被删除。

如果希望消息被 `Consumer` 确认后仍然保留下来，可配置 [消息保留策略][Link 7]实现。

对于批量处理来说，当某一批消息的所有索引都被确认时，该批消息将被删除。

#### 消息保留策略 ####

`Pulsar`有两个特性，让你可以覆盖上面的默认行为。

*  Message **retention** enables you to store messages that have been acknowledged by a consumer（即保留）
 *  Message **expiry** enables you to set a time to live (TTL) for messages that have not yet been acknowledged（即过期）

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 5]

图中上面的是消息存留，存留规则会被用于某`namespace`下所有的`topic`，指明哪些消息会被持久存储，即使已经被确认过。 没有被留存规则覆盖的消息将会被删除。 Without a retention policy, *all* of the acknowledged messages would be deleted.

图中下面的是消息过期，有些消息即使还没有被确认，也被删除掉了。因为根据设置在`namespace`上的`TTL`，他们已经过期了。（例如，`TTL`为5分钟，过了十分钟消息还没被确认）

[Message retention and expiry · Apache Pulsar][Message retention and expiry _ Apache Pulsar]

在`Pulsar`中，你有两种方式在命名空间的级别去修改这种行为：

*  你可以通过设置[消息保留策略][Link 8]持久化存储不在 backlog 内的消息(因为他们已经在每个现有的订阅上被确认，或者并没有被订阅)。
 *  可以通过设置[ time to live][time to live](TTL)，设置消息在指定的时间内不被确认的话，自动确认。

#### broker参数设置 ####

可以通过以下两个参数来配置实例级别的消息保留策略:

# Default message retention time
    defaultRetentionTimeInMinutes=0
    
    # Default retention size
    defaultRetentionSizeInMB=0

默认情况下，这两个参数都设置为 `0`。

#### 设置保留策略 ####

有三种形式设置：

1.  使用[`set-retention`][set-retention]子命令并指定命名空间，使用`-s`/`--size`参数指定大小限制，使用`-t`/`--time`参数指定时间限制。
2.  `POST /admin/v2/namespaces/:tenant/:namespace/retention`
3.  Java程序中设置
    
        int retentionTime = 10; // 10 minutes
        int retentionSize = 500; // 500 megabytes
        RetentionPolicies policies = new RetentionPolicies(retentionTime, retentionSize);
        admin.namespaces().setRetention(namespace, policies);

### 取消确认 ###

当消费者在某个时间没有成功的消费某条消息，消费者想重新消费到这条消息，这个消费者可以发送一条取消确认消息到 broker，broker 会将这条消息重新发给消费者。

消息取消确认也有单条取消模式和累积取消模式 ，这依赖于消费者使用的订阅模式。

在独占消费模式和灾备订阅模式中，消费者仅仅只能对收到的最后一条消息进行取消确认。

### 确认超时 ###

如果消息没有被成功消费，你想去让 `broker`自动重新交付这个消息， 你可以采用未确认消息自动重新交付机制。 客户端会跟踪 `超时` 时间范围内所有未确认的消息。 并且在指定超时时间后会发送一个 `重发未确认的消息` 请求到 `broker`。

> **Note** If batching is enabled, other messages and the unacknowledged messages in the same batch are redelivered to the consumer.

> **Note**  
> Prefer negative acknowledgements over acknowledgement timeout. 确认取消是以更高的精度在控制单条消息的重新传递。当消息处理时间超过确认超时时间时，要避免无效的消息重传。

### 死信主题 ###

死信主题使您能够在使用者无法成功地使用某些消息时使用新消息。在此机制中，无法使用的消息存储在单独的主题中，称为死信主题。您可以决定如何处理死信主题中的消息。

在`java client`例子中

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
                  .topic(topic)
                  .subscriptionName("my-subscription")
                  .subscriptionType(SubscriptionType.Shared)
                  .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(maxRedeliveryCount)
                        .build())
                  .subscribe();

默认的`topic`名称为：

<topicname>-<subscriptionname>-DLQ

也可以自己指定：

## Topic ##

`pulsar`的`topic`命名是符合良好结构的`URL`：

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

<table> 
 <thead> 
  <tr> 
   <th align="left">Topic名称组成</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><code>持久化</code> / <code>非持久化</code></td> 
   <td align="left">用来标识 topic 的类型。 Pulsar 支持两种主题类型：<a href="https://pulsar.apache.org/docs/zh-CN/concepts-architecture-overview#persistent-storage" rel="nofollow">持久化</a>和<a href="https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#non-persistent-topics" rel="nofollow">非持久化</a>。 主题默认是持久化类型，如果不特殊指定主题类型，那主题就是持久化的。 对于持久化的主题，所有的消息都会被持久化的保存到磁盘当中(如果 broker 不是单机模式，消息会被持久化到多块磁盘)，而非持久化的主题的数据不会被保存到磁盘里面。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>租户</code></td> 
   <td align="left">The topic tenant within the instance. Tenants are essential to multi-tenancy in Pulsar, and spread across clusters.</td> 
  </tr> 
  <tr> 
   <td align="left"><code>命名空间</code></td> 
   <td align="left">将相关联的 topic 作为一个组来管理，是管理 Topic 的基本单元。 大多数对 topic 的管理都是对<a href="https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#namespaces" rel="nofollow">命名空间</a>的一项配置。 每个租户里面可以有一个或者多个命名空间。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>topic</code></td> 
   <td align="left">主题名称</td> 
  </tr> 
 </tbody> 
</table>

## 命名空间（namespace） ##

命名空间是租户内部逻辑上的命名术语。 可以通过[admin API][]在租户下创建多个命名空间。 例如，包含多个应用程序的租户可以为每个应用程序创建单独的命名空间。 `Namespace`使得程序可以以层级的方式创建和管理topic  
例如：`Topic my-tenant/app1` ，它的`namespace`是`app1`这个应用，对应的租户是 `my-tenant`。 你可以在`namespace`下创建任意数量的`topic`。

## 订阅 ##

Pulsar 中有四种订阅模式: [独占][Link 9]，[共享][Link 10]，[灾备][Link 11]和[key共享][key] 下图展示了这三种模式：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 6]

Pulsar提供了灵活的消息模型，支持以下订阅类型：

<table> 
 <thead> 
  <tr> 
   <th>订阅类型</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Exclusive subscription</td> 
   <td>排他的，只能有一个Consumer，接收一个Topic所有的消息</td> 
  </tr> 
  <tr> 
   <td>Shared subscription</td> 
   <td>共享的，可以同时存在多个Consumer，每个Consumer处理Topic中一部消息（Shared模型是不保证消息顺序的，Consumer数量可以超过分区的数量）</td> 
  </tr> 
  <tr> 
   <td>Failover subscription</td> 
   <td>Failover模式，同一时刻只有一个有效的Consumer，其余的Consumer作为备用节点，在Master Consumer不可用后进行替代（看起来适用于数据量小，且解决单点故障的场景）</td> 
  </tr> 
  <tr> 
   <td>Key_Shared Subscription</td> 
   <td>可以存在多个Consumer，消息以分布的形式跨消费者交付，具有相同键或相同顺序键的消息只交付给一个消费者。无论消息被重新传递多少次，它都被传递给相同的消费者。当使用者连接或断开连接时，将导致所服务的使用者更改消息的某些键。</td> 
  </tr> 
 </tbody> 
</table>

## 多主题订阅 ##

当`consumer`订阅`pulsar`的主题时，它默认指定订阅了一个主题，例如：

// 通过明确指定的topic列表
    persistent://public/default/my-topic
    // 当使用正则匹配订阅多个主题的时候，所有的主题必须是在同一个命名空间里面的。
    persistent://public/default/finance-.*

当订阅多个主题的时候，`Pulsar` 客户端将自动调用 `Pulsar API`找到符合匹配规则的主题列表，然后订阅这些主题。 如果此时有暂不存在的主题，那么一旦这些主题被创建，消费者会自动订阅这些主题。

java client example.
    
    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();

## 分区 topic ##

普通的主题仅仅被保存在单个 `broker`中，这限制了主题的最大吞吐量，分区主题实际是通过在底层拥有 N 个内部主题来实现的，这个 N 的数量就是等于分区的数量。 当向分区的`topic`发送消息，每条消息被路由到其中一个`broker`。 `Pulsar`自动处理跨`broker`的分区分布。

![partitions][]

### 路由模式 ###

有三种 `MessageRoutingMode` 可用：

<table> 
 <thead> 
  <tr> 
   <th align="left">发送模式</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><code>RoundRobinPartition</code></td> 
   <td align="left">如果消息没有指定 key，为了达到最大吞吐量，生产者会以 round-robin 方式将消息发布到所有分区。 请注意round-robin并不是作用于每条单独的消息，而是作用于延迟处理的批次边界，以确保批处理有效。 如果消息指定了key，分区生产者会根据key的hash值将该消息分配到对应的分区。 这是默认的模式。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>SinglePartition</code></td> 
   <td align="left">如果消息没有指定 key，生产者将会随机选择一个分区，并发布所有消息到这个分区。 如果消息指定了key，分区生产者会根据key的hash值将该消息分配到对应的分区。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>CustomPartition</code></td> 
   <td align="left">使用自定义消息路由器实现来决定特定消息的分区。 用户可以创建自定义路由模式：使用 <a href="https://pulsar.apache.org/docs/zh-CN/client-libraries-java" rel="nofollow">Java client</a> 并实现<a href="https://pulsar.incubator.apache.org/api/client/2.8.0-SNAPSHOT/org/apache/pulsar/client/api/MessageRouter" rel="nofollow">MessageRouter</a> 接口。</td> 
  </tr> 
 </tbody> 
</table>

### 顺序保证 ###

当使用 `SinglePartition`或者`RoundRobinPartition`模式时，如果消息有`key`，消息将会被路由到匹配的分区，这是基于[ProducerBuilder][] 中[HashingScheme][] 指定的散列`shema`。

<table> 
 <thead> 
  <tr> 
   <th align="left">顺序保证</th> 
   <th align="left">说明</th> 
   <th align="left">路由策略与消息Key</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">按键分区</td> 
   <td align="left">所有具有相同 key 的消息将按顺序排列并放置在相同的分区（Partition）中。</td> 
   <td align="left">使用 <code>SinglePartition</code> 或 <code>RoundRobinPartition</code> 模式，每条消息都需要有key。</td> 
  </tr> 
  <tr> 
   <td align="left">生产者排序</td> 
   <td align="left">来自同一生产者的所有消息都是有序的</td> 
   <td align="left">路由策略为<code>SinglePartition</code>, 且每条消息都没有key。</td> 
  </tr> 
 </tbody> 
</table>

## 非持久topic ##

一般，`pulsar`会持久化所有未被消费的消息数据到`bookkeep bookies`中，以保证持久性主题上的消息数据可以在 `broker` 重启和订阅者故障转移之后继续存在。

`Pulsar`也提供了非持久`topic`。非持久`topic`的消息不会被保存在硬盘上，只存活于内存中。当使用非持久`topic`分发时，杀掉`Pulsar`的`broker`或者关闭订阅者，此`topic（ non-persistent)`上所有的瞬时消息都会丢失，意味着客户端可能会遇到消息缺失。

非持久性主题具有这种形式的名称：

non-persistent://tenant/namespace/topic

所以，非持久`topic`，消息数据仅存活在内存。 如果`broker`挂掉或者因其他情况不能从内存取到，你的消息数据就可能丢失。

默认非持久`topic`在`broker`上是开启的。 你可以通过broker的[配置][Link 12]关闭。

## 消息去重 ##

消息去重保证了一条消息只能在 `Pulsar`服务端被持久化一次。 消息去重是一个`Pulsar`可选的特性，它能够阻止不必要的消息重复，它保证了即使消息被消费了多次，也只会被保存一次。

下面为官网配置链接：

[Message deduplication · Apache Pulsar][Message deduplication _ Apache Pulsar]

## 消息延迟传递 ##

延时消息功能允许你能够过一段时间才能消费到这条消息，而不是消息发布后，就马上可以消费到

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 7]

Broker 保存消息是不经过任何检查的。 当消费者消费一条消息时，如果这条消息是延时消息，那么这条消息会被加入到`DelayedDeliveryTracker`当中。 订阅检查机制会从`DelayedDeliveryTracker`获取到超时的消息，并交付给消费者。

### Broker ###

默认情况下启用延迟消息传递。您可以在代理配置文件中更改它：

# Whether to enable the delayed delivery for messages.
    # If disabled, messages are immediately delivered and there is no tracking overhead.
    delayedDeliveryEnabled=true
    
    # Control the ticking time for the retry of delayed message delivery,
    # affecting the accuracy of the delivery time compared to the scheduled time.
    # Default is 1 second.
    delayedDeliveryTickTimeMillis=1000

### 生产者（Producer） ###

下面是 Java 当中生产延时消息一个例子：

// message to be delivered at the configured delay interval
    producer.newMessage().deliverAfter(3L, TimeUnit.Minute).value("Hello Pulsar!").send();

## 租户 ##

成为一个多租户系统是 `Pulsar` 最初设计理念的一部分。 并且，`Pulsar` 提出了租户的概念。 租户可以跨集群分布，每个租户都可以有单独的认证和授权机制。 租户也是存储配额、消息`TTL`和隔离策略的管理单元。

`Pulsar` 的多租户性质主要体现在 `topic` 的 `URL` 中：

persistent://tenant/namespace/topic

`Pulsar`通过租户和命名空间这两个关键概念支持多租户。

\-`Pulsar`为指定的多个租户配置了合适的容量。

*  命名空间是一个术语，指租户的管理单元。 命名空间上设置的配置策略适用于在该命名空间中创建的所有 `topic`。 租户可以使用`REST API`和 \[`pulsar-admin CLI` 工具来创建多个命名空间。 例如，包含多个应用程序的租户可以为每个应用程序创建单独的命名空间。

### 命名空间更改事件和主题级策略 ###

`Pulsar`是一个多租户的事件流处理系统。 管理员可以通过设置不同层次的策略来管理租户和命名空间。 然而，有些策略，例如数据保留策略和数据存储配额策略，仅仅只能在命名空间级别设置。 在许多使用场景中，用户需要对主题设置对应的策略。 命名空间更改事件提供了一个简单有效的方式去修改主题级别的策略。 在这种方法中，Pulsar 使用事件日志去保存命名空间的事件改变记录(比如主题策略改动)。

这种方式有以下好处：

*  避免过多使用`Zookeeper`, 给 `Zookeeper` 带来更高的负载。
 *  使用 `Pulsar` 作为传播策略缓存的事件日志。可以有效地扩展。
 *  可以使用`Pulsar SQL` 可以查询命名空间的改变日志，并对系统进行审计。

每个命名空间有一个叫做`__change_events`的系统主题。 这个系统主题用来保存这个命名空间的事件改变信息。

你可以使用 [`pulsar-admin`][pulsar-admin] 工具来管理租户。

### 创建新租户 ###

以下是租户创建命令的示范：

$ bin/pulsar-admin tenants create my-tenant \
      --admin-roles my-admin-role \
      --allowed-clusters us-west,us-east

此命令会创建一个新租户 `my-tenant`，并允许它使用 `us-west` 和 `us-east` 集群。

成功自识别为拥有 `my-admin-role` 角色的客户端可以在这个租户上执行所有的管理型任务。

persistent://tenant/namespace/topic

## 总结 ##

`Apache pulsar`作为架构合理的消息系统，将存储（`bookkeeper`）和计算（`broker`）分离，其集群伸缩性更好，并且由于broker为无状态的，其计算扩容也更为方便。总的来说，`Apache pulsar`值得一试。

[apache_pulsar_ Apache Pulsar - distributed pub-sub messaging system _github.com]: https://github.com/apache/pulsar
[Set up a standalone Pulsar locally _ Apache Pulsar]: https://pulsar.apache.org/docs/zh-CN/standalone/
[Yahoo]: http://yahoo.github.io/
[Apache]: https://www.apache.org/
[Link 1]: https://pulsar.apache.org/docs/zh-CN/administration-geo
[API]: https://pulsar.apache.org/docs/zh-CN/concepts-clients
[Java]: https://pulsar.apache.org/docs/zh-CN/client-libraries-java
[Go]: https://pulsar.apache.org/docs/zh-CN/client-libraries-go
[Python]: https://pulsar.apache.org/docs/zh-CN/client-libraries-python
[C]: https://pulsar.apache.org/docs/zh-CN/client-libraries-cpp
[topic]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging#subscription-modes
[Link 2]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging#exclusive
[Link 3]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging#shared
[Link 4]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging#failover
[Apache BookKeeper]: http://bookkeeper.apache.org/
[Link 5]: https://pulsar.apache.org/docs/zh-CN/concepts-architecture-overview#persistent-storage
[Pulsar Functions]: https://pulsar.apache.org/docs/zh-CN/functions-overview
[Pulsar IO]: https://pulsar.apache.org/docs/zh-CN/io-overview
[Link 6]: https://pulsar.apache.org/docs/zh-CN/concepts-tiered-storage
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center]: /images/20220829/79d91307b1184e64853602a4e4029b6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20220829/5d310123935f436b80cfcf6b16c5ba53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20220829/b2866daccd624f83aaa2a0b8928ae04c.png
[pulsar manager topic]: /images/20220829/e815facb42a3456183aed727aa4a6b37.png
[cursor]: /images/20220829/31f3c1090c694f1287ca447533d384e4.png
[broker]: https://link.zhihu.com/?target=https%3A//github.com/apache/pulsar/wiki/PIP-6%3A-Guaranteed-Message-Deduplication
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20220829/93ddfc8b6fb64b108a457bd5df27f1b1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20220829/d51eb54cd0c74f3f974e09fec5539b9c.png
[flow permit request]: https://pulsar.apache.org/docs/zh-CN/develop-binary-protocol#flow-control
[receiverQueueSize]: https://pulsar.apache.org/docs/zh-CN/client-libraries-java#configure-consumer
[Link 7]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging#message-retention-and-expiry
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20220829/f9523e48cd014a969631a6f948348bc3.png
[Message retention and expiry _ Apache Pulsar]: https://pulsar.apache.org/docs/zh-CN/cookbooks-retention-expiry/
[Link 8]: https://pulsar.apache.org/docs/zh-CN/cookbooks-retention-expiry/#retention-policies
[time to live]: https://pulsar.apache.org/docs/zh-CN/cookbooks-retention-expiry/#time-to-live-ttl
[set-retention]: https://pulsar.apache.org/docs/zh-CN/pulsar-admin#namespaces-set-retention
[admin API]: https://pulsar.apache.org/docs/zh-CN/admin-api-namespaces#create
[Link 9]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#exclusive
[Link 10]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#shared
[Link 11]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#failover
[key]: https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#key_shared
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20220829/0a55428173c744bca9a26add4f6926a1.png
[partitions]: /images/20220829/3bc0e8b520cd4248a727e31aedd474eb.png
[ProducerBuilder]: https://pulsar.incubator.apache.org/api/client/2.8.0-SNAPSHOT/org/apache/pulsar/client/api/ProducerBuilder
[HashingScheme]: https://pulsar.incubator.apache.org/api/client/2.8.0-SNAPSHOT/org/apache/pulsar/client/api/HashingScheme
[Link 12]: https://pulsar.apache.org/docs/zh-CN/reference-configuration#broker-enableNonPersistentTopics
[Message deduplication _ Apache Pulsar]: https://pulsar.apache.org/docs/zh-CN/cookbooks-deduplication/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20220829/764b9158f7d4458fba3d0d7c2405dd8f.png
[pulsar-admin]: https://pulsar.apache.org/docs/zh-CN/pulsar-admin