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

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

Messages

消息是 Pulsar 的基础“单元”。 消息指 producer 发布到 topic的内容，也指 consumer 从 topic 中 consume 的内容（并在消息处理完成后发送确认）。 消息类似于邮政服务系统中的信件。

更多关于 Pulsar 消息的信息，参阅二进制协议。

Producers

生产者是连接 topic 的程序，它将消息发布到一个 Pulsar broker 上。

发送模式

Producer可以以同步（sync）或者异步（async）的方式发布消息到broker。

压缩

Messages published by producers can be compressed during transportation in order to save bandwidth. Pulsar currently supports the following types of compression:

• LZ4
• ZLIB
• ZSTD
• SNAPPY

批量处理

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

在 Pulsar 中，批次被跟踪并存储为单个单元，而不是单个消息。 Under the hood, the consumer unbundles a batch into individual messages. 但即使启用了批量处理，也始终将计划中的消息(通过 deliverAt 或者 deliverAfter 进行配置) 作为单个消息发送。

In general, a batch is acknowledged when all its messages are acknowledged by the consumer. This means unexpected failures, negative acknowledgements, or acknowledgement timeouts can result in redelivery of all messages in a batch, even if some of the messages have already been acknowledged.

To avoid redelivering acknowledged messages in a batch to the consumer, Pulsar introduces batch index acknowledgement since Pulsar 2.6.0. When batch index acknowledgement is enabled, the consumer filters out the batch index that has been acknowledged and sends the batch index acknowledgement request to the broker. Broker 维护批量索引的确认状态并跟踪每批索引的确认状态，以避免向 consumer 发送已确认的消息。 当某一批消息的所有索引都被确认时，该批消息将被删除。

默认情况下，批量索引确认是禁用的( batchIndexAcknowledgeEnable=false)。 你可以通过在 broker 中将 batchIndexAcknowledgeEnable 设置为 true 来启用批量索引确认。 Enabling batch index acknowledgement may bring more memory overheads. So, perform this operation with caution.

Consumers

A consumer is a process that attaches to a topic via a subscription and then receives messages.

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

接收模式

Messages can be received from brokers either synchronously (sync) or asynchronously (async).

监听

Client libraries provide listener implementation for consumers. For example, the Java client provides a MesssageListener interface. 在这个接口中，一旦接受到新的消息，received方法将被调用。

确认

当 consumer 成功消费一条消息后会向 broker 发送一个确认请求（acknowledgement request）。 仅当所有订阅都完成确认后，消息才会被删除，在这之前消息都是被永久保存 的。 如果希望消息被 Consumer 确认后仍然保留下来，可配置 消息保留策略实现。

For a batch message, if batch index acknowledgement is enabled, the broker maintains the batch index acknowledgement status and tracks the acknowledgement status of each batch index to avoid dispatching acknowledged messages to the consumer. 当某一批消息的所有索引都被确认时，该批消息将被删除。 For details about the batch index acknowledgement, see batching.

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

Messages can be acknowledged in the following two ways:

• Messages are acknowledged individually. With individual acknowledgement, the consumer needs to acknowledge each message and sends an acknowledgement request to the broker.
• Messages are acknowledged cumulatively. 累积确认时，消费者只需要确认最后一条他收到的消息。 All messages in the stream up to (and including) the provided message are not re-delivered to that consumer.

Note

Cumulative acknowledgement cannot be used in the shared subscription mode, because the shared subscription mode involves multiple consumers having access to the same subscription. In the shared subscription mode, messages can be acknowledged individually.

取消确认

When a consumer does not consume a message successfully at a time, and wants to consume the message again, the consumer can send a negative acknowledgement to the broker, and then the broker will redeliver the message.

Messages can be negatively acknowledged one by one or cumulatively, which depends on the consumption subscription mode.

In the exclusive and failover subscription modes, consumers only negatively acknowledge the last message they have received.

In the shared and Key_Shared subscription modes, you can negatively acknowledge messages individually.

注意：如果开启批处理，在确认取消后，同一批中的其他消息都会被重新发送给 Consumer。

确认超时

When a message is not consumed successfully, and you want to trigger the broker to redeliver the message automatically, you can adopt the unacknowledged message automatic re-delivery mechanism. Client will track the unacknowledged messages within the entire acktimeout time range, and send a redeliver unacknowledged messages request to the broker automatically when the acknowledgement timeout is specified.

注意: 如果启用批处理，同一批次中的其他消息就跟未确认消息一样可能会重新发送给消费者。

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

死信主题

Dead letter topic enables you to consume new messages when some messages cannot be consumed successfully by a consumer. In this mechanism, messages that are failed to be consumed are stored in a separate topic, which is called dead letter topic. You can decide how to handle messages in the dead letter topic.

The following example shows how to enable dead letter topic in a Java client using the default dead letter topic:

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

The default dead letter topic uses this format:

<topicname>-<subscriptionname>-DLQ

If you want to specify the name of the dead letter topic, use this Java client example:

Consumer<byte[]> 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 depends on message re-delivery. Messages are redelivered either due to acknowledgement timeout or negative acknowledgement. If you are going to use negative acknowledgement on a message, make sure it is negatively acknowledged before the acknowledgement timeout.

Note
Currently, dead letter topic is enabled only in the shared subscription mode.

Retry letter topic

For many online business systems, a message needs to be re-consumed because any exception occurs in the business logic processing. Generally, users hope that they can flexibly configure the delay time for re-consuming the failed messages. In this case, you can configure the producer to send messages to both the business topic and the retry letter topic and you can enable automatic retry on the consumer. When automatic retry is enabled on the consumer, a message is stored in the retry letter topic if the messages fail to be consumed and therefore the consumer can automatically consume failed messages from the retry letter topic after a specified delay time.

By default, automatic retry is disabled. You can set enableRetry to true to enable automatic retry on the consumer.

This example shows how to consumer messages from a retry letter topic.

Consumer<byte[]> 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();

Topic

和其他的发布订阅系统一样，Pulsar 中的 topic 是被命名的通道，用做从producer到 consumer传输消息。 Topic的名称为符合良好结构的URL:

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

不需要显式的创建topic

你并不需要显式的创建topic。 如果客户端尝试从一个还不存在的topic写或者接受消息，pulsar将会按在topic名称提供的namespace下自动创建topic。 If no tenant or namespace is specified when a client creates a topic, the topic is created in the default tenant and namespace. You can also create a topic in a specified tenant and namespace, such as persistent://my-tenant/my-namespace/my-topic. persistent://my-tenant/my-namespace/my-topic means the my-topic topic is created in the my-namespace namespace of the my-tenant tenant.

命名空间

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

Subscriptions

订阅是命名好的配置规则，指导消息如何投递给消费者。 Pulsar 中有 4 种可用的订阅模式： 独占（exclusive）， 共享（shared），灾备（failover）和 键共享（key_shared）。 下图展示了这三种模式：

订阅-发布，队列，亦或两者都是

订阅可以灵活组合出很多可能性：
- 如果你想实现传统的 “发布-订阅消息”形式 ，你可以让每个消费者都有一个唯一的订阅名称（独占）
- 如果你想实现传统的“消息队列” 形式，可以使多个消费者使用同一个的订阅名称（共享、灾备、键共享）
- 如果你想同时做到这两点，你可以让一些消费者使用独占方式，剩余消费者使用其他方式。

Exclusive

In exclusive mode, only a single consumer is allowed to attach to the subscription. 如果多于一个消费者尝试以同样方式去订阅主题，消费者将会收到错误。

In the diagram below, only Consumer A-0 is allowed to consume messages.

Exclusive模式为默认订阅模式。

Failover（灾备）

In failover mode, multiple consumers can attach to the same subscription. A master consumer is picked for non-partitioned topic or each partition of partitioned topic and receives messages. When the master consumer disconnects, all (non-acknowledged and subsequent) messages are delivered to the next consumer in line.

For partitioned topics, broker will sort consumers by priority level and lexicographical order of consumer name. Then broker will try to evenly assigns topics to consumers with the highest priority level.

For non-partitioned topic, broker will pick consumer in the order they subscribe to the non partitioned topic.

In the diagram below, Consumer-B-0 is the master consumer while Consumer-B-1 would be the next consumer in line to receive messages if Consumer-B-0 is disconnected.

Shared（共享）

In shared or round robin mode, multiple consumers can attach to the same subscription. 消息通过round robin轮询机制分发给不同的消费者，并且每个消息仅会被分发给一个消费者。 当消费者断开连接，所有被发送给他，但没有被确认的消息将被重新安排，分发给其它存活的消费者。

In the diagram below, Consumer-C-1 and Consumer-C-2 are able to subscribe to the topic, but Consumer-C-3 and others could as well.

Shared模式的限制

When using shared mode, be aware that: * Message ordering is not guaranteed. * You cannot use cumulative acknowledgment with shared mode.

Key_Shared

In Key_Shared mode, multiple consumers can attach to the same subscription. Messages are delivered in a distribution across consumers and message with same key or same ordering key are delivered to only one consumer. No matter how many times the message is re-delivered, it is delivered to the same consumer. When a consumer connected or disconnected will cause served consumer change for some key of message.

Limitations of Key_Shared mode

使用 Key_Shared 模式时需要注意两点：
* 需要为消息指定一个 key 或 orderingKey
* Key_Shared 模式不能使用累积确认

可以在 broker.config 中禁用 Key_Shared 模式。

多主题订阅

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

• On the basis of a regular expression (regex), for example persistent://public/default/finance-.*
• 通过明确指定的topic列表

通过正则订阅多主题时，所有的主题必须在同一个namespace。

当订阅多主题时，Pulsar客户端会自动调用Pulsar的API来发现匹配表达式或者列表的所有topic，然后全部订阅。 如果此时有暂不存在的topic，那么一旦这些topic被创建，conusmer会自动订阅。

No ordering guarantees across multiple topics

When a producer sends messages to a single topic, all messages are guaranteed to be read from that topic in the same order. However, these guarantees do not hold across multiple topics. So when a producer sends message to multiple topics, the order in which messages are read from those topics is not guaranteed to be the same.

下面是多主题订阅在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<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();

代码例子，请见：

• Java

分区 topic

通常一个topic仅被一个broker服务，这限制了topic的最大吞吐量。 Partitioned topics are a special type of topic that can be handled by multiple brokers, which allows for much higher throughput.

其实在背后，分区的topic通过N个内部topic实现，N是分区的数量。 当向分区的topic发送消息，每条消息被路由到其中一个broker。 Pulsar自动处理跨broker的分区分布。

下图对此做了阐明：

Here, the topic Topic1 has five partitions (P0 through P4) split across three brokers. 因为分区多于broker数量，其中有两个broker要处理两个分区。第三个broker则只处理一个。（再次强调，分区的分布是Pulsar自动处理的）。

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

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

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

Partitioned topics need to be explicitly created via the admin API. The number of partitions can be specified when creating the topic.

路由模式

When publishing to partitioned topics, you must specify a routing mode. The routing mode determines which partition—-that is, which internal topic—-each message should be published to.

有三种 MessageRoutingMode 可用:

顺序保证

The ordering of messages is related to MessageRoutingMode and Message Key. Usually, user would want an ordering of Per-key-partition guarantee.

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

散列scheme

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

有两种可用的散列函数： JavaStringHash 和Murmur3_32Hash. The default hashing function for producer is JavaStringHash. 请注意，当producer可能来自于不同语言客户端时，JavaStringHash是不起作用的。建议使用Murmur3_32Hash。

非持久topic

By default, Pulsar persistently stores all unacknowledged messages on multiple BookKeeper bookies (storage nodes). 因此，持久性主题上的消息数据可以在 broker 重启和订阅者故障转移之后继续存在。

Pulsar also, however, supports non-persistent topics, which are topics on which messages are never persisted to disk and live only in memory. Pulsar也提供了非持久topic。非持久topic的消息不会被保存在硬盘上，只存活于内存中。当使用非持久topic分发时，杀掉Pulsar的broker或者关闭订阅者，此topic（ non-persistent)）上所有的瞬时消息都会丢失，意味着客户端可能会遇到消息缺失。

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

non-persistent://tenant/namespace/topic

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

In non-persistent topics, brokers immediately deliver messages to all connected subscribers without persisting them in BookKeeper. 如果有一个订阅者断开连接，broker将无法重发这些瞬时消息，订阅者将永远也不能收到这些消息了。 去掉持久化存储的步骤，在某些情况下，使得非持久topic的消息比持久topic稍微变快。但是同时，Pulsar的一些核心优势也丧失掉了。

非持久topic，消息数据仅存活在内存。 如果broker挂掉或者因其他情况不能从内存取到，你的消息数据就可能丢失。 Use non-persistent topics only if you’re certain that your use case requires it and can sustain it.

默认非持久topic在broker上是开启的。 你可以通过broker的配置关闭。 你可以通过使用pulsar-admin-topics接口管理非持久topic。

性能

Non-persistent messaging is usually faster than persistent messaging because brokers don’t persist messages and immediately send acks back to the producer as soon as that message is delivered to connected brokers. 非持久topic让producer有更低的发布延迟。

客户端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<byte[]> consumer = client.newConsumer()
        .topic(npTopic)
        .subscriptionName(subscriptionName)
        .subscribe();

这里还有一个非持久topic的java producer例子：

Producer<byte[]> producer = client.newProducer()
                .topic(npTopic)
                .create();

消息保留和到期

Pulsar broker默认如下：

• immediately delete all messages that have been acknowledged by a consumer, and
• 以消息backlog的形式，持久保存所有的未被确认消息

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

All message retention and expiry is managed at the namespace level. For a how-to, see the Message retention and expiry cookbook.

下图说明了这两种概念：

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

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

Message deduplication

Message duplication occurs when a message is persisted by Pulsar more than once. Message deduplication is an optional Pulsar feature that prevents unnecessary message duplication by processing each message only once, even if the message is received more than once.

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

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

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

Message deduplication is handled at the namespace level. For more instructions, see the message deduplication cookbook.

生产者幂等

The other available approach to message deduplication is to ensure that each message is only produced once. This approach is typically called producer idempotency. 这种方式的缺点是，把消息去重的工作推给了应用去做。 在Pulsar中，去重被broker处理的，这意味着你不需要修改你的客户端代码。 你只需要做一些管理上的变化（参考Managing message deduplication ）

去重和实际一次语义

消息去重，使Pulsar成为与流处理引擎（SPE）或者其他寻求实际一次处理语义的系统连接的完美消息系统。 消息系统若不提供自动消息去重，则需要SPE或者其他系统保证去重。这意味着严格的消息顺序来自于让程序承担额外的去重工作。 使用Pulsar，严格的顺序保证不会带来任何应用层面的代价。

更深入的信息可以参考 Streamlio blog上的此篇博文

消息延迟传递

Delayed message delivery enables you to consume a message later rather than immediately. In this mechanism, a message is stored in BookKeeper, DelayedDeliveryTracker maintains the time index(time -> messageId) in memory after published to a broker, and it is delivered to a consumer once the specific delayed time is passed.

Delayed message delivery only works well in Shared subscription mode. In Exclusive and Failover subscription mode, the delayed message is dispatched immediately.

The diagram below illustrates the concept of delayed message delivery:

A broker saves a message without any check. When a consumer consumes a message, if the message is set to delay, then the message is added to DelayedDeliveryTracker. A subscription checks and gets timeout messages from DelayedDeliveryTracker.

Broker

Delayed message delivery is enabled by default. You can change it in the broker configuration file as below:

# 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

The following is an example of delayed message delivery for a producer in Java:

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