Message deduplication

When Message deduplication is enabled, it ensures that each message produced on Pulsar topics is persisted to disk only once, even if the message is produced more than once. 消息去重是在服务端自动执行的。

Pulsar中启用消息去重,你必须同时配置 Pulsar broker 和客户端。

运作模式

You can enable or disable message deduplication on a per-namespace basis. By default, it is disabled on all namespaces. You can enable it in the following ways:

  • Enable for all namespaces at the broker-level
  • Enable for specific namespaces with the pulsar-admin namespaces interface

配置消息去重

你可以通过broker.conf配置 Pulsar 启用消息去重。 下面是和去重有关的配置项。

参数名Description默认值
brokerDeduplicationEnabled设置 Pulsar broker 级别消息去重的默认值。 If it is set to true, message deduplication is enabled by default on all namespaces; if it is set to false, you have to enable or disable deduplication on a per-namespace basis.false
brokerDeduplicationMaxNumberOfProducers为了去重目的而存储信息的最大生产者数量。10000
brokerDeduplicationEntriesInterval去重快照生成后,允许存储的去重信息数量。 更大的时间间隔会减少快照的数量,尽管会延长主题的恢复时长(重播快照内容所需要的时间)。1000
brokerDeduplicationProducerInactivityTimeoutMinutes停止活动的时间(分钟级别),当生产者断开连接超过这个时间,就丢弃和该生产者有关的去重信息。360 (6小时)

设置 broker 级别的默认值

By default, message deduplication is disabled on all Pulsar namespaces. To enable it by default on all namespaces, set the brokerDeduplicationEnabled parameter to true and re-start the broker.

即使设置了brokerDeduplicationEnabled配置项,通过 Pulsar admin CLI 设置的值也会覆盖 broker 的配置。

启用消息去重

Though message deduplication is disabled by default at broker-level, you can enable message deduplication for specific namespaces using the pulsar-admin namespace set-deduplication command. You can use the --enable/-e flag and specify the namespace. The following is an example with <tenant>/<namespace>:

  1. $ bin/pulsar-admin namespaces set-deduplication \
  2. public/default \
  3. --enable # or just -e

禁用消息去重

Even if you enable message deduplication at broker-level, you can disable message deduplication for a specific namespace using the pulsar-admin namespace set-deduplication command. Use the --disable/-d flag and specify the namespace. The following is an example with <tenant>/<namespace>:

  1. $ bin/pulsar-admin namespaces set-deduplication \
  2. public/default \
  3. --disable # or just -d

Pulsar 客户端

If you enable message deduplication in Pulsar brokers, you need complete the following tasks for your client producers:

  1. 指定生产者的名称。
  2. 设置消息超时为0(即无超时)。

The instructions for Java, Python, and C++ clients are different.

Java clients

Python clients

C++ clients

To enable message deduplication on a Java producer, set the producer name using the producerName setter, and set the timeout to 0 using the sendTimeout setter.

  1. import org.apache.pulsar.client.api.Producer;import org.apache.pulsar.client.api.PulsarClient;import java.util.concurrent.TimeUnit;PulsarClient pulsarClient = PulsarClient.builder() .serviceUrl("pulsar://localhost:6650") .build();Producer producer = pulsarClient.newProducer() .producerName("producer-1") .topic("persistent://public/default/topic-1") .sendTimeout(0, TimeUnit.SECONDS) .create();

To enable message deduplication on a Python producer, set the producer name using producer_name, and set the timeout to 0 using send_timeout_millis.

  1. import pulsarclient = pulsar.Client("pulsar://localhost:6650")producer = client.create_producer( "persistent://public/default/topic-1", producer_name="producer-1", send_timeout_millis=0)

To enable message deduplication on a C++ producer, set the producer name using producer_name, and set the timeout to 0 using send_timeout_millis.

  1. #include <pulsar/Client.h>std::string serviceUrl = "pulsar://localhost:6650";std::string topic = "persistent://some-tenant/ns1/topic-1";std::string producerName = "producer-1";Client client(serviceUrl);ProducerConfiguration producerConfig;producerConfig.setSendTimeout(0);producerConfig.setProducerName(producerName);Producer producer;Result result = client.createProducer(topic, producerConfig, producer);