

Kafka 消费积压影响写入？试试 Pulsar

360智汇云开发者

2025-11-13

导读：360智汇云是AI时代的企业智数云底座，了解更多产品，请搜索：360智汇云。

1. Pulsar 特性

Apache Pulsar 的设计背景，是雅虎为了替代团队内部多个业务线搭建小规模 Kafka 集群，带来的维护成本，于2012年开源。

因此 Pulsar 的多租户和集群容灾是所有开源MQ中最好的。

Pulsar 支持最全的消息队列 API，支持复杂业务场景。

1.1 极限性能

单分区 topic 写性能极限测试。

低延迟：单条同步发送，生产者发送一条消息收到回包，仅需0.3ms，服务端完成 2 副本持久化到磁盘。
高吞吐：赞批发送，单客户端单线程，可实现150万写QPS

线上 Pulsar 集群端到端读写延迟P99耗时监控如下，平均延迟再10ms以下，最高 P99 耗时再 20ms 以内：

1.2 长期存储

Pulsar 使用 SSD 磁盘存储热数据，冷数据存储在S3，Pulsar分层存储的实现原理是：

卸载阈值：Topic 的数据随着时间，生成很多 Segment，当 Segment 关闭时，触发上传 S3，已经上传完的Segment不删除，消费者可优先从磁盘读取数据进行消费，性能更好。
热数据删除阈值：到达热数据删除阈值后，本地磁盘上的 Segment 数据被删除，消费者只能从S3消费数据。此阈值支持按业务属性个性化配置。
数据过期阈值：即数据的保存周期，当 Segment 到达保存阈值后，删除S3上的Segment。

使用注意事项

冷数据消费性能上限：默认超过4小时的数据，需要从S3拉取，单分区读性能上限：60MB/s，可通过扩展分区数线性扩展冷数据消费能力。
支持个性化配置：分层存储的热数据阈值、数据过期阈值支持自定义配置，支持关闭分层存储

1.3 延迟队列

目前智汇云内部 Pulsar 版本对应 Apache Pulsar 4.0.x 版本，支持将延迟消息的索引持久化到磁盘，从而实现更大规模、更长延迟时间的延迟消息。

Pulsar 延迟消息的使用非常简单，普通类型 Topic 即可支持收发定时/延时消息，调用 SDK 的 API 即可发送定时/延时消息。

//定时消息producer.newMessage()    .value(value.getBytes())    .deliverAt(timeStamp)    .send();//延时消息producer.newMessage()    .value(value.getBytes())    .deliverAfter(delayTime, TimeUnit.SECONDS)    .send();

使用延迟消息时需要注意：

1) topic 的 TTL 自动确认时间需要比延时消息的时间更长，否则延迟消息会在TTL后自动确认，不投递给消费者。

2) 生产者不可以使用 batch 模式发送，在创建 producer 的时候把 enableBatch 参数设为 false。

3) 消费模式仅支持使用 Shared 模式进行消费，否则会失去定时效果（Key-shared 也不支持）。

// 构建消费者Consumer<byte[]> consumer = pulsarClient.newConsumer()        .topic("persistent://pulsar-xxx/sdk_java/topic1")        .subscriptionName("sub_topic1")        // 声明消费模式为Shared（共享）模式        .subscriptionType(SubscriptionType.Shared)    .subscribe()