八张图带你全面了解Pulsar跨地域复制

本文转载自微信公众号《程序员jinjunzhu》，作者jinjunzhu。转载本文请联系程序员jinjunzhu公众号。跨区域复制是ApachePulsar企业级特性的重要组成部分。保证了系统的高可用性，并且在运维管理上非常方便。今天我就用5张图来带大家学习这个功能。1、多机房部署Pulsar多机房部署如下：在上面的Pulsar架构中，Pulsar的三个集群分别部署在北京、上海、贵阳三个机房。每个机房有一组集群，每个集群有一个Topic1。对应的订阅是Subscription1。但是3个集群之间没有数据同步。如果某个机房出现故障，则不会消耗该机房的库存信息。2.跨区域复制（GEO-Replication）Pulsar最初是在雅虎内部开发的。设计之初就加入了全球十多个雅虎机房跨区域复制的需求。在上面的例子中，如果三个机房可以互相同步数据，即使一个机房出现故障，这个机房的库存数据也可以被其他机房的消费者消费，因为已经同步到其他机房了。如下图所示：Pulsar的跨地域复制是如何实现的？2.1存储模型回顾首先，我们回顾一下Pulsar的存储模型。我们知道Pulsar的消息持久化使用的是存储系统BookKeeper，如下图所示：Producer生产消息后，会flush到底层的BookKeeper存储引擎进行持久化。当一个Consumer创建时，它需要订阅一个Topic，Pulsar会给它分配一个Subscription来进行绑定。如上图所示，Consumer绑定到Subscription2。Subscription会不断从Ledger中获取消息并推送给Consumer，当然前提是Consumer有消息缓存空间。Consumer消费消息后，向Subscription回复ACK，Subscription收到ACK后将游标往回推一位。这个cursor也保存在BookKeeper中，BookKeeper会为这个cursor打开一个专门的Ledger。2.2跨区域复制流程Pulsar的跨区域复制与上面的存储模型非常相似，在集群中多了一个Replicator。以上海机房到北京机房的复制为例，如下图所示：上海机房的Pulsar集群中有一个Replicator，这个Replicator中有一个Producer-R，它在北京机房绑定Topic1，向北京机房发送数据。上海机房集群产生的消息先持久化到本地集群，再异步转发到北京集群。上海机房Replicator中的Producer-R与集群中的Producer1没有任何关系，其配置的集群地址为北京机房的集群地址。整个复制过程如下：Producer1向上海机房的Topic1生产消息；上海机房将消息持久化到BookKeeper；BookKeeper返回成功后将消息推送到Replicator的Cursor；Replicator的Cursor通过Producer-R向北京机房Topic1发送消息；北京机房的Topic1写入BookKeeper成功后，向上海机房的Replicator的Cursor回复ACK。上海机房的Cursor收到ACK后，通过Producer-R推送下一条消息。2.3消息丢失和幂等性由于Replicator中维护了一个Cursor，如果一条消息没有收到北京机房的ACK，Replicator可以通过Producer-R再次向北京机房发送消息，可以防止消息丢失。如果因为网络问题，Producer-R向北京机房推送消息后，北京机房回复的ACK上海机房收不到，怎么办？Producer-R会再次向北京机房发送同样的消息。重复。为了解决消息幂等的问题，Pulsar提供了Producer的幂等配置。北京机房开启这个设置后，broker中会缓存一个内部的Cursor，用来保存最后收到的消息的MessageId。如果收到一条新消息，如果MessageId小于等于当前Cursor中缓存的MessageId，则这条消息将被丢弃。2.4消息序列上图中，上海机房的Producer-R和北京机房的Producer2都向北京机房的Topic1写消息。如何保证消息的顺序？因为跨机房复制是一个异步过程，Pulsar只能保证上海机房和北京机房写入的消息顺序。比如上海机房的Producer-R写了5条消息msg1~msg5，北京机房的Producer2写了5条消息msgA~msgE。最终的消息序列可能如下：2.5低延迟跨region从两个方面保证复制的低延迟：Replicator和broker在同一个进程中，减少了数据拷贝。跨地域复制采用异步方式。2.6ZooKeeper集群跨机房复制可以使用一个全局的ZooKeeper集群，将Pulsar集群信息注册到ZooKeeper集群。如下图所示：这样，每个集群都可以根据存储在ZooKeeper中的信息创建本地的Replicator。但是如果没有全局的ZooKeeper集群，因为保存的数据是轻量级的，使用本地的ZooKeeper集群也是可以的。如下图所示：这样各个机房的Pulsar集群从本地的ZooKeeper获取到需要复制的远程集群信息，然后就可以创建Replicator了。这种情况比较灵活。由于Pulsar集群采用如下方式，全局的ZooKeeper无法满足要求。比如现在西安机房有个Pulsar集群，不产生消息，只接受从北京、上海、贵阳三个机房复制过来的数据，如下图：3.复制原理Pulsar中topic的格式如下：persistent://tenant/namespace/topicTopic的父目录有namespace和tenant。要允许两个集群之间跨区域复制消息，首先要允许租户（tenant）有访问两个集群的权限。跨区域复制在命名空间级别进行管理。如果一个命名空间允许跨地域复制，则发布到该命名空间上任何主题的消息将被复制到指定集合中的所有集群。3.1租户授权要使用跨区域复制，首先要为租户设置访问权限。以下命令授予租户my-tenant访问pulsar-shanghai、pulsar-beijing和pulsar-guiyang的权限。bin/pulsar-admintenantscreatemy-tenant--admin-rolesmy-admin-role--allowed-clusterspulsar-shanghai,pulsar-beijing,pulsar-guiyang3.2命名空间级别的跨区域复制在命名空间级别管理，租户拥有权限后，为要复制的集群分配命名空间：bin/pulsar-adminnamespacesset-clustersmy-tenant/my-namespace--clusterspulsar-shanghai,pulsar-beijing,pulsar-guiyangnamespace-levelreplication可以随时更改,更改后立即生效。为命名空间配置跨地域复制后，默认情况下，该命名空间下创建的所有主题都会复制到列表中的其他集群。如果要选择固定的集群进行复制，可以使用PulsarClient指定。例如，JavaClient下面的代码只允许主题my-topic在pulsar-shanghai和pulsar-beijing两个集群之间进行复制。ListrestrictReplicationTo=Arrays.asList("pulsar-shanghai","pulsar-beijing");Producerproducer=client.newProducer().topic("my-topic").create();producer.newMessage().value("my-payload".getBytes()).setReplicationClusters(restrictReplicationTo).send();3.3Topic-levelstartup为了使一个topic能够跨区域复制，在topic级别启动它：bin/pulsar-admintopicsset-replication-clusters--clusterspulsar-shanghai,pulsar-beijing,pulsar-guiyangmy-tenant/my-namespace/Topic13.4防止循环复制如果配置了上海机房和北京机房的跨区域复制，复制自上海机房到达北京机房后，是否可以将消息从北京机房复制到上海机房？当然不是。上海机房向北京机房发送消息时，会在消息中添加一个Property，表示数据是哪个机房产生的。北京的数据中心收到这个数据后就会知道是从其他机房复制过来的，Replicator中的Cursor在订阅消息的时候会过滤掉这部分消息。总结一句话，Pulsar的跨地域复制其实就是在本地集群中创建一个Producer，使用远程集群作为Producer的发送地址，从本地集群发送消息，在本地维护一个Cusor来保证消息的可靠性和幂等性。