Apache Pulsar 消息存储模型：理解分布式消息平台背后的机制

2023-09-12 17:37:58

Apache Pulsar：深入剖析其卓越存储模型

揭开消息持久化的秘密

在当今数据驱动的世界中，分布式消息平台已成为管理海量数据流的基石。Apache Pulsar 以其卓越的性能和可扩展性而闻名，而其存储模型正是其秘密所在。本文将为您深入剖析 Pulsar 的存储机制，揭示其令人印象深刻的持久性和可扩展性的奥秘。

分层存储架构

Pulsar 采用了分层的存储架构，巧妙地组织消息以实现高效访问。最底层是 Apache BookKeeper Ledger，一种分布式日志服务，负责将消息持久化到磁盘。每个 Ledger 由多个 Entry 组成，每个 Entry 存储一个写入操作。最后，Pulsar 消息作为 Entry 中的可修改对象出现，允许更新和删除操作。

严格的消息排序

Pulsar 对消息排序一丝不苟，确保严格的顺序交付。消息在 Entry 中按顺序写入，并根据消息 ID 分配，该 ID 由主题、分区和序列号组成。这种设计消除了消息乱序的可能性，使消费者能够可靠地处理数据。

多版本支持

Pulsar 突破了传统消息平台的界限，提供了多版本支持。消息的旧版本在 Entry 中保留一定时间，使消费者能够访问历史数据。这种特性对于回放、审计和数据恢复至关重要。

数据存储优化

Pulsar 采用了多项优化技术来提高存储效率。首先，页缓存将常用数据缓存在内存中，加快访问速度。其次，块存储将数据存储在固定大小的块中，优化了磁盘 I/O 操作。最后，数据压缩使用 snappy 或 zstd 等算法压缩数据，从而节省存储空间。

无缝扩展和高可用性

Pulsar 的存储模型为无缝扩展和高可用性奠定了基础。Ledger 可以水平扩展，通过添加更多 BookKeeper 服务器来增加存储容量。此外，BookKeeper 的复制机制确保了数据冗余，即使发生故障也能保证消息的持久性。

代码示例：创建 Ledger

import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.zookeeper.ZooKeeper;

public class CreateLedgerExample {

    public static void main(String[] args) throws Exception {
        // 配置 BookKeeper 客户端
        ClientConfiguration conf = new ClientConfiguration();
        conf.setMetadataServiceUri("zookeeper://localhost:2181");

        // 创建 BookKeeper 客户端
        BookKeeper bk = new BookKeeper(conf);

        // 配置 BookKeeper 服务器
        ServerConfiguration serverConf = new ServerConfiguration();
        serverConf.setZkServers("localhost:2181");

        // 创建 ZooKeeper 连接
        ZooKeeper zk = ZkUtils.createZooKeeperConnection(serverConf.getZkServers(), 10000);

        // 创建 Ledger
        bk.asyncCreateLedger(3, 2, BookKeeper.DigestType.MAC, new GenericCallback<LedgerMetadata>() {
            @Override
            public void operationComplete(int rc, LedgerMetadata ledgerMetadata) {
                if (rc == 0) {
                    System.out.println("Ledger created successfully: " + ledgerMetadata.getName());
                } else {
                    System.err.println("Failed to create ledger: " + ZkUtils.getZKException(rc));
                }
            }
        });

        // 等待创建 Ledger 完成
        Thread.sleep(5000);

        // 关闭 BookKeeper 客户端
        bk.close();

        // 关闭 ZooKeeper 连接
        zk.close();
    }
}