Pulsar消息传递机制

Pulsar消息传递机制

官方文档：https://pulsar.apache.org/docs/en/concepts-messaging/

转载请链接到本文章

Pulsar 采用发布-订阅的设计模式(简称 pub-sub)， 该设计模式中，producer发布消息到 topic， Consumer订阅topic、处理发布的消息，并在处理完成后发送确认。

一旦创建订阅，即使 consumer 断开连接，Pulsar 仍然可以保存所有消息。 在 consumer 确认消息已处理成功后，才会删除消息。

1. 消息

Messages are the basic "unit" of Pulsar. The following table lists the components of messages.

ComponentDescriptionValue / data payload消息携带的数据。所有 Pulsar 消息都包含原始字节，尽管消息数据也可以符合数据模式。Key消息可以选择用键进行标记，这在topic压缩等 *** 作很有用。Producer name生成消息的生产者的名称。如果不指定生产者名称，则使用默认名称。Properties用户自定义属性的键值对（可选）。Sequence ID每个 Pulsar 消息都属于其主题的有序序列。消息的sequence ID 是它在该序列中的顺序。Publish time消息发布的时间戳。时间戳由生产者自动应用。Event time应用程序可以附加到消息的时间戳（可选）， 例如处理消息的时间。 如果没有明确设置，则消息的事件时间为 0。TypedMessageBuilder用于构造消息。 您可以使用 TypedMessageBuilder 设置消息的键值对属性。在设置 TypedMessageBuilder 时，最佳的选择是将 key 设置为字符串。 如果将 key 设置为其他类型（例如，AVRO 对象），则 key 会以字节形式发送，这时 consumer 就很难再拿到AVRO对象。

消息的默认大小为 5 MB。您可以使用以下配置来配置消息的最大大小。

在 broker.conf 文件中，maxMessageSize；在 bookkeeper.conf 配置文件中，nettyMaxframeSizeBytes

2. 生产者 2.1 发送模式

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

发送模式说明同步发送生产者在发送每条消息后等待代理的确认。如果没有收到确认，生产者将发送 *** 作视为失败。异步发送Producer 将把消息放于阻塞队列中，并立即返回 然后，客户端将在后台将消息发送给 broker。 如果队列已满（最大大小可配置），则调用 API 时，producer 可能会立即被阻止或失败，具体取决于传递给 producer 的参数。

2.2 访问方式 访问方式说明Shared多个生产者可以发布到同一个主题。这是默认设置Exclusive一个主题只有一个生产者能发布。 如果已经有生产者连接，其他生产者试图在这个主题上发布信息会立即出错。 如果“旧”生产者与broker发生网络分区，则“旧”生产者被驱逐，并选择“新”生产者作为下一个独占生产者。WaitForExclusive如果已经连接了生产者，那么生产者创建将挂起（而不是超时），直到即将创建的生产者获得Exclusive访问权。 成功成为唯一的生产者被视为leader。因此，如果你想为你的应用程序实现leader选举方案，您可以使用这种访问模式。

一旦应用程序成功创建了具有Exclusive或WaitForExclusive访问模式的生产者，该应用程序的实例就可以保证是该主题的唯一编写者。其他试图在此主题上生产的producers会立即出错，或者必须等到他们获得Exclusive访问权限。

2.3 其他

2.3.1 压缩

你可以在传输过程中压缩生产者生产的消息，Pulsar目前支持以下类型的压缩：

• LZ4

• ZLIB

• ZSTD

• SNAPPY

2.3.2 批量处理

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

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

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

为避免重发已经确认的消息给consumer，在Pulsar2.6.0引入批量索引确认。当批量索引确认启用时，consumer过滤掉已经被确认的batch index并发送批量索引确认请求给broker。Broker 维护批量索引的确认状态并跟踪每个批索引的确认状态，以避免向 consumer 发送已确认的消息。 当某一批消息的所有索引都被确认时，该批消息将被删除。

默认情况下，batch index确认是禁用的(acknowledgmentAtBatchIndexLevelEnabled=false)。您可以通过在broker端将acknowledgmentAtBatchIndexLevelEnabled参数设置为true来启用batch index确认。启用batch index确认将会导致更多内存开销。

2.3.3 分块(chunking)

分块说明：
1. batching和chunking不能同时启用。如果想要启用分块(chunking) ，您必须提前禁用批量处理。
2. 仅对持久化的主题支持分块
3. 分块仅支持exclusive(独占)和failover(灾备)订阅模式

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

Consumer 会缓存收到的块状消息，直到收到消息的所有分块为止。 然后 consumer 将分块的消息拼接在一起，并将它们放入接收器队列(receiver-queue)中。 客户端从接收器队列(receiver-queue)中消费消息。 一旦 consumer 消费整个大消息并确认，consumer 就会在内部发送与该大消息关联的所有分块消息的确认。你能在consumer上设置 maxPendingChunkedMessage 参数。当达到阈值时，consumer 通过静默确认未分块的消息 或 通过将其标记为未确认，要求broker稍后重新发送这些消息。

broker不需要任何更改来支持非共享订阅的分块。broker仅使用chunkedMessageRate记录主题上的分块消息比率。

处理一个 producer 和一个安排好的 consumer 的分块消息

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

处理多个 producer 和一个安排好的 consumer 的分块消息

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

3. 消费者

消费者是通过订阅附加到主题然后接收消息的过程。

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

3.1 接收模式 接收模式说明同步接收同步模式，在收到消息之前都是被阻塞的。异步接收异步接收模式会立即返回一个 future 值（如 Java 中的 CompletableFuture），一旦收到新的消息就立刻完成。

3.2 其他

3.2.1 监听

客户端库为消费者提供监听器实现。例如， Java client 提供一个 MesssageListener 接口。 在这个接口中，一旦接受到新的消息，received方法将被调用。

3.2.2 确认

当消费者成功地消费了一条消息，这个消费者会发送一个确认信息给broker。 这个消息是永久保存的，只有在所有的订阅者都确认后才会被删除。 如果希望消息被 Consumer 确认后仍然保留下来，可配置消息保留策略实现。

对于批处理消息，如果启用了批索引确认，broker将维护批索引确认状态，并跟踪每个批索引的确认状态，以避免将已经确认的消息发送给consumer。当批处理消息的所有索引被确认后，该批处理消息将被删除。

消息有两种确认模式：单条确认或者累计确认。 累积确认时，consumer只需要确认最后一条它收到的消息。 所有之前（包含此条）的消息，都不会被再次重发给那个消费者。

消息有以下两种确认方式：

• 消息被单独确认。通过单独确认，消费者需要确认每条消息并向broker发送确认请求。

• 累积确认模式。累积确认时，消费者只需要确认最后一条他收到的消息。 所有之前（包含此条）的消息，都不会被再次发送给那个消费者。

累积确认不能在共享订阅模式中使用，因为共享订阅模式涉及多个可访问相同subscription的consumer。在共享订阅模式中，消息都是单条确认的。

3.2.3 否定确认(Negative acknowledgement)

当consumer在某个时间没有成功地消费某条消息，consumer想重新消费到这条消息，这个consumer可以发送一条negative acknowledgement消息到broker，broker会将这条消息重新发给consumer。

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

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

在共享(shared)和Key_Shared订阅模式下，您可以单条否定确认消息。

请注意，对有序订阅类型（例如 Exclusive、Failover 和 Key_Shared）的否定确认可能会导致失败的消息不按原始顺序到达消费者。

如果启用了批处理，则同一批处理中的其他消息和否定确认消息将被重新发送给consumer。

3.2.4 确认超时

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

如果启用批处理，则同一批次中的其他消息和未确认的消息将重新传递给consumer。

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

3.2.5 Dead letter topic

Dead letter topic使你可以在消费者无法成功消费某些消息时消费新消息。在这种机制中，消费失败的消息存储在一个单独的主题中，称为Dead letter topic。你可以决定如何处理Dead letter topic中的消息。

下面的示例展示了如何在java客户端中使用默认的dead letter topic去启用dead letter topic

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

默认的dead letter topic使用以下格式

--DLQ

如果你想指定dead letter topic的名字，使用以下java客户端示例：

Consumer consumer = pulsarClient.newConsumer(Schema.BYTES)
              .topic(topic)
              .subscriptionName("my-subscription")
              .subscriptionType(SubscriptionType.Shared)
              .deadLetterPolicy(DeadLetterPolicy.builder()
                    .maxRedeliverCount(maxRedeliveryCount)
                    .deadLetterTopic("your-topic-name")
                    .build())
              .subscribe();

Dead letter topic取决于消息重新传递。由于确认超时或否定确认，消息被重新传递。如果您打算对消息使用否定确认，请确保在确认超时之前对其进行否定确认。

目前，在 Shared 和 Key_Shared 订阅模式下启用了dead letter topic。

3.2.6 Retry letter topic

很多在线的业务系统，由于业务逻辑处理出现异常，消息一般需要被重新消费。 若要配置重新消费失败消息的延时，你可以配置producer同时发送消息到业务topic和retry letter topic，并允许consumer自动重试消费。 配置了允许consumer自动重试，如果消息没有被消费，它将被保存到retry letter topic当中，因此consumer在指定的延迟后自动重新消费retry letter topic里面的失败消息。

默认情况下，自动重试处于禁用状态。您可以设置enableRetry为true在consumer上启用自动重试。

如下例子展示如何从retry letter topic消费消息。

Consumer consumer = pulsarClient.newConsumer(Schema.BYTES)
                .topic(topic)
                .subscriptionName("my-subscription")
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true)
                .receiverQueueSize(100)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(maxRedeliveryCount)
                        .retryLetterTopic("persistent://my-property/my-ns/my-subscription-custom-Retry")
                        .build())
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();

4. 主题

和其他的发布-订阅系统一样，Pulsar的主题被命名为从生产者向消费者传递信息的通道。Topic的名称为符合良好结构的URL：

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

topic名称组成说明persistent / non-persistent用来标识 topic 的类型。 Pulsar 支持两种主题类型：持久化和非持久化。 主题默认是持久化类型，如果不指定主题类型，那主题就是持久化的。 对于持久化的主题，所有的消息都会被持久化的保存到磁盘当中(如果 broker不是单机模式，消息会被持久化到多块磁盘)，而非持久化的主题的数据不会被持久化保存到磁盘里面。tenant实例中的主题租户。租户对于Pulsar中的多租户至关重要，并且分布在集群中。namespace将相关联的topic作为一个组来管理，是管理Topic的基本单元。大多数topic的配置都是在namespace的级别执行。 每个租户里面可以有一个或者多个namespace。topic名字的最后一部分。主题名称在 Pulsar 实例中没有特殊意义。

No need to explicitly create new topics你不需要在Pulsar上明确地创建主题。如果客户端尝试从不存在的主题当中生产消息或消费消息，Pulsar 将会自动在该topicname的namespace下创建同名的主题。如果客户端创建主题时没有指定tenant和namespace，则该主题将在默认的租户和命名空间中创建。你也可以在指定的tenant和namespace中创建topic，如persistent://my-tenant/my-namespace/my-topic。persistent://my-tenant/my-namespace/my-topic意思是 在my-tenant tenant中的my-namespace namespace创建my-topic topic。

5. 命名空间

命名空间是租户内部逻辑上的命名术语。租户可以通过admin API创建多个命名空间。例如，包含多个应用程序的租户可以为每个应用程序创建单独的命名空间。namespace允许应用程序创建和管理主题的层次结构。Topic my-tenant/app1，它的namespace是app1这个应用，对应的租户是 my-tenant。 你可以在namespace下创建任意数量的topic。

6. 订阅

订阅是命名好的配置规则，指导消息如何投递给消费者。 Pulsar 中有四种订阅模式: 独占，共享，灾备和key共享。下图展示了这三种模式：

Pub-Sub or Queuing In Pulsar, 你可以灵活的选择不同的订阅模式。

如果你想在consumers当中使用传统的”发布-订阅消息“，你可以为每个consumer指定一个特定的订阅名称，这就是独占订阅模式。

如果你想在consumers当中实现”消息队列“的效果，则多个消费者会拥有相同的订阅名称(如共享模式，灾备模式，key共享模式)。

如果你想同时实现两种效果，则消费者可以将独占模式和其他的订阅模式结合起来使用。

6.1 无消费者订阅及其对应模式

当一个订阅没有消费者时，它的订阅模式是未定义的。订阅模式是在消费者连接到订阅时定义的，并且可以通过重新启动所有具有不同配置的消费者来更改模式。

6.2 Exclusive（独占）

在独占模式下，只允许一个消费者附加到订阅。如果多个消费者使用 同一个订阅 订阅一个主题，则会发生错误。在下图中，仅允许消费者 A-0消费消息。

独占模式是默认的消费模式。

6.3 Failover（灾备）

在failover模式中，多个消费者可以连接到同一个subscription。主消费者会消费非分区主题或者分区主题中的每个分区的消息。当主消费者断开连接时，所有(未确认的和后续的)消息都被传递给下一个消费者。

对于分区主题来说，Broker 将按照消费者的优先级和消费者名称的词汇表顺序对消费者进行排序。 然后试图将主题均匀的分配给优先级最高的消费者。

对于非分区主题来说，broker 会根据消费者订阅非分区主题的顺序选择消费者。

在下图中，消费者 B-0是主消费者，而如果消费者 B-0断开连接，消费者 B-1将是下一个接收消息的消费者。

6.4 Shared（共享）

在共享或轮询模式下，多个消费者可以附加到相同的订阅。消息通过round robin轮询机制分发给不同的消费者，并且每个消息仅会被分发给一个消费者。 当消费者断开连接，所有被发送给他，但没有被确认的消息将被重新安排，分发给其它存活的消费者。

在下图中，消费者 C-1和消费者 C-2能够订阅该主题，但消费者 C-3和其他人也可以。

共享模式的限制

使用共享模式时，请注意：①不保证消息排序。②您不能在共享模式下使用累积确认。

6.5 Key_Shared

在Key_Shared模式下，多个消费者可以附加到同一个订阅。消息在跨消费者的分布中传递，具有相同键或相同排序键的消息仅传递给一个消费者。无论消息被重新传递多少次，它都会传递给同一个消费者。当消费者连接或断开连接时，将导致服务的消费者更改某些消息键。

Key_Shared 模式的限制

当您使用 Key_Shared 模式时，请注意：①您需要为消息指定一个 key 或 orderingKey。②您不能在 Key_Shared 模式下使用累积确认。③您的producer应禁用批处理或使用基于密钥的批处理构建器。

可以在broker.config文件中禁用Key_Shared订阅模式。

7. 多主题订阅

当consumer订阅pulsar的主题时，它默认指定订阅了一个主题，例如：persistent://public/default/my-topic。 从Pulsar的1.23.0-incubating的版本开始，Pulsar消费者可以同时订阅多个topic。 你可以用以下两种方式定义topic的列表：

• 基于正则表达式（regex），例如persistent://public/default/finance-.*

• 通过明确指定的topic列表

当使用正则匹配订阅多个主题的时候，所有的主题必须是在同一个namespace里面的。

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

没有跨多个主题的排序保证

当生产者向单个主题发送消息时，保证所有消息都以相同的顺序从该主题读取。但是，这些保证不适用于多个主题。因此，当生产者向多个主题发送消息时，不能保证从这些主题读取消息的顺序是相同的。

如下是 Java 订阅多个主题的代码示例：

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 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 someTopicsConsumer = pulsarClient.newConsumer()
                .topicsPattern(someTopicsInNamespace)
                .subscriptionName("subscription-1")
                .subscribe();

关于代码示例，请参阅 Java。

8. 分区topic

普通的主题仅仅被保存在单个 broker中，这限制了主题的最大吞吐量。分区主题(Partitioned topics) 是由多个broker处理的特殊类型的主题，因此允许更高的吞吐量。

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

下图对此做了阐明：

Topic1主题有五个分区(P0到P4)，它们分别分布在三个broker上。因为分区多于broker数量，其中有两个broker要处理两个分区。第三个broker则只处理一个。（再次强调，分区的分布是Pulsar自动处理的）。

这个topic的消息被广播给两个consumer。路由模式确定每条消息该发往哪个分区，而订阅模式确定消息传递给哪个消费者。

大多数境况下，路由和订阅模式可以分开制定。通常来讲，吞吐能力的要求，决定了分区/路由的方式。订阅模式则应该由应用的语义来做决定。

分区topic和普通topic，对于订阅模式如何工作，没有任何不同。分区只是决定了从生产者生产消息到消费者处理及确认消息过程中发生的事情。

需要通过admin API显式创建分区主题。创建主题时可以指定分区数。

8.1 路由模式

发布到分区主题时，必须指定路由模式。路由模式决定了每条消息应该发布到哪个分区——即哪个内部主题。

有三种 MessageRoutingMode 可用:

发送模式说明RoundRobinPartition如果消息没有指定 key，为了达到最大吞吐量，生产者会以 round-robin 方式将消息发布到所有分区。 请注意round-robin并不是作用于每条单独的消息，而是作用于延迟处理的批次边界，以确保批处理有效。 如果消息指定了key，分区生产者会根据key的hash值将该消息分配到对应的分区。 这是默认的模式。SinglePartition如果消息没有指定 key，生产者将会随机选择一个分区，并发布所有消息到这个分区。 如果消息指定了key，分区生产者会根据key的hash值将该消息分配到对应的分区。CustomPartition使用自定义消息路由器实现来决定特定消息的分区。 用户可以创建自定义路由模式：使用 Java client 并实现MessageRouter 接口。

8.2 顺序保证

消息的排序与 MessageRoutingMode 和 Message Key 相关。通常，用户需要按key分区保证的顺序。

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

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

8.3 散列scheme

HashingScheme 是代表一组标准散列函数的枚举。为一个指定消息选择分区时使用。

有两种可用的标准散列函数： JavaStringHash 和Murmur3_32Hash. 生产者默认的散列函数是 JavaStringHash. 请注意，当producer可能来自于不同语言客户端时，JavaStringHash是不起作用的。这种情况下，建议使用Murmur3_32Hash。

9. 非持久topic

默认情况下，Pulsar将所有未确认的消息持久化存储在多个BookKeeper bookies(存储节点)上。因此，持久性主题上的消息数据可以在 broker重启和subscriber故障转移之后继续存在。

不过，Pulsar也支持非持久主题，即消息永远不会持久化到磁盘上，而只存在于内存中。当使用非持久topic分发时，杀掉Pulsar的broker或者断开主题的订阅者，此topic(非持久化)上所有的瞬时(in-transit)消息都会丢失，意味着客户端可能会遇到消息丢失。

非持久性主题具有这种形式的名称(注意名称中的 non-persistent)：

non-persistent://tenant/namespace/topic

如何使用非持久topic的更多信息，请参考 Non-persistent messaging cookbook

在非持久主题中，brokers立即将消息传递给所有连接的订阅者，而无需将消息持久化到BookKeeper中。如果有一个订阅者断开连接，broker将无法重发这些瞬时(in-transit)消息，订阅者将永远也不能收到这些消息了。 去掉持久化存储的步骤，在某些情况下，使得非持久topic的消息比持久topic稍微变快。但是同时，Pulsar的一些核心优势也丧失掉了。

非持久topic，消息数据仅存活在内存。 如果一个消息broker挂掉或者因其他情况不能从内存重新取到消息数据，你的消息数据就可能丢失。只有当你确定你的用例需要并且能够维持它时，才使用非持久性主题。

默认非持久topic在broker上是启用的。 你可以通过broker的配置禁用。你可以使用pulsar-admin topics命令管理非持久主题。更多消息，请参考 pulsar-admin。

9.1 性能

非持久性消息传递通常比持久性消息传递更快，因为brokers不会持久化消息，并且一旦该消息传递给连接的brokers，就会立即将acks发送回生产者。非持久topic让producer有更低的发布延迟。

9.2 客户端API

Producer和consumer以连接持久topic同样的方式连接到非持久topic。重要的区别是，topic的名称必须以non-persistent开头。三种订阅模式---exclusive，shared，failover对于非持久topic都是支持的。

下面是一个非持久topic的 java consumer 例子：

PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650")
        .build();
String npTopic = "non-persistent://public/default/my-topic";
String subscriptionName = "my-subscription-name";

Consumer consumer = client.newConsumer()
        .topic(npTopic)
        .subscriptionName(subscriptionName)
        .subscribe();

下面是一个相同非持久topic的 Java producer 示例：

Producer producer = client.newProducer()
                .topic(npTopic)
                .create();

10. 消息处理 10.1 消息保留和过期

Pulsar broker默认如下：

• 立即删除所有已被消费者确认的消息，以及

• 以消息backlog的形式，持久保存所有的未被确认消息

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

• 消息保留使您能够存储已被消费者确认的消息

• 消息到期使您可以为尚未确认的消息设置生存时间 (TTL)(Time To Live)

所有消息保留和到期都在 namespace 级别进行管理。有关 *** 作方法，请参阅 消息保留和到期 说明书。

下图说明了这两种概念：

图中上面的是消息保留，保留规则会被用于某namespace下所有的topic，指明哪些消息会被持久存储，即使已经被确认过。 没有被留存规则覆盖的已经确认的消息将会被删除。如果没有保留策略，所有已确认的消息将被删除。

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

10.2 消息去重

消息去重保证了一条消息只能在 Pulsar 服务端被持久化一次。 消息去重是一个 Pulsar 可选的特性，它通过 即使接收到某消息的次数不止一次，每条消息也只处理一次，能够阻止不必要的消息重复。

下图展示了开启和关闭消息去重的场景：

最上面的场景中，消息去重被关闭。 Producer发布消息1到一个topic，消息到达broker后，被 持久化 到BookKeeper。然后producer又发送了消息1（可能因为某些重试逻辑），然后消息被接收后又持久化在BookKeeper，这意味着消息重复发生了。

在第二个场景中，producer发布了消息1，消息被broker接收然后持久化，和第一个场景是一样的。 当producer尝试再次发送该消息时，broker知道已经收到了消息1，因此不会再持久化消息1。

消息去重在namespace级别或topic级别进行处理。有关更多说明，请参阅 消息去重手册 。

10.2.1 生产者幂等

消息去重的另一种可用方法是确保每个消息只生成一次。这种方法通常被称为生产者幂等。这种方式的缺点是，把消息去重的工作推给了应用去做。在 Pulsar 中，这是在 broker 上处理的，用户不需要去修改客户端的代码。 相反，你只需要通过修改配置就可以实现。 可以通过查看 消息去重指南 去了解更多详细信息。

10.2.2 去重和一次有效语义

消息去重，使 Pulsar 成为了结合了 流处理引擎（SPE） 和 其他寻求"一次有效"处理语义的系统 的理想消息系统。 如果消息系统没有提供自动去重能力，那么 SPE (流处理引擎) 或者其他连接系统就必须自己实现去重语义，这意味着严格的消息排序是以消息去重加重应用程序负担为代价的。使用Pulsar，严格的顺序保证则不会带来任何应用层面的代价。

你能够在 这篇博客 上获得更详细的信息。

10.3 消息延迟传递

延时消息功能允许你能够过一段时间才能消费到这条消息，而不是消息发布后，就马上可以消费到。在这种机制中，消息被存储在BookKeeper中，DelayedDeliveryTracker在发布到broker之后维护内存中的时间索引(time -> messageId)，一旦指定的延迟时间一过，它就被传递给消费者。

延迟消息传递仅适用于共享订阅模式。在 Exclusive（独占） 和 Failover（灾备） 订阅模式下，延迟消息会立即发送。

如下图所示，说明了延时消息的实现机制：

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

Broker

默认情况下启用延迟消息传递。您可以在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();