08 Confluent_Kafka权威指南 第八章：跨集群数据镜像

CHAPTER 8 Cross-Cluster Data Mirror 跨集群数据镜像

本书大部分内容都在讨论单个kafka集群的配置、维护和使用。但是，在一些场景中，可能需要多集群架构。 在某些情况下，集群是完全分离的，他们属于不同部门的不同实例，没有理由将数据从一个集群复制到另外一个集群。有时，不同的SLA或者工作负载使得单个集群提供多个用例服务的集群很难调优。在某些时候，还有不同的安全需求。这些场景非常容易管理多个不同的集群，就像多次允许单个集群一样。 在其他场景中，不同的集群是互相依赖的，管理有要不断地在集群之间复制数据。在大多数数据库中，在数据库服务之间持续复制数据称为复制。由于我们使用复制来描述属于同一集群的kafka节点之间的数据移动，因此我们将把kafak集群之间的数据复制称之为镜像。Apache kafka内置的跨集群 的复制器称为mirrormaker。 在本章中，我们将讨论所有或者部分数据的跨集群镜像。我们将首先讨论跨集群的镜像的一些常用用例。然后我们将展示一些用于实现这些用例的架构，并讨论每种架构的优缺点。然后我们将讨论MirrorMaker本书以及如何使用它。我们将分享一些操作技巧，包括部署的性能调优。最后我们将讨论mirrorMaker的一些替代方案。

Use Cases of Cross-Cluster Mirroring 跨集群镜像用例

下面是一个示例列表，说明何时应该使用跨集群镜像。

• Regional and central clusters（区域中心集群） 在某些情况下，公司在不同地理区域，城市或者大陆拥有一个或者多个数据中心。每个数据中心都有自己的kafka集群，有些应用程序只需要与本地集群通信就可以工作。但是有些应用程序需要来自多个数据中的数据，否则就不会考虑跨数据中心的复制解决方案。在很多情况下，这事一种要求，但是最经典的例子就是一家公司根据供求关系调整价格。该公司可能在其所在的每个城市都有一个数据中心，收集有关当地的供求信息，并相应的调整价格。所有这些信息将被镜像到一个中心集群中，业务分析师可以在此允许公司范围内的收入报告。
• Redundancy (DR) （冗余）应用程序只允许在一个kafka集群上，不需要来自其他位置的数据，但是你担心由于某种原因可能导致整个集群不可用。你希望有第二个kafka集群，其中包含存在于第一个集群中的所有数据，因此在此情况下，你可以将应用程序指向第二个集群，并像往常一样继续。
• Cloud migrations（迁移到云）如今，许多公司在内部数据中心和云提供商两方面都开展业务，为了冗余，应用程序通常允许在云提供商的多个区域上，有时会使用多个云提供商。在这种情况下，每个内部数据中心和每个云区域中通常至少有一个kafka集群。每个数据中心和区域的应用程序使用这些kafka集群来有效的在数据中心之间传输数据。丽日，如果一个应用程序部署在云上，但是需要一些数据更新的应用程序允许在本地数据中心并存储一个本地数据库，你可以使用kakfa的连接器捕获数据更改，然后将这些变化镜像到云服务上的kafka集群去。信的应用程序可以使用他们。这有助于控制跨数据中心的流量成本，并改善流程和治理的安全性。

Multicluster Architectures 多集群架构

现在我们已经看到了一些需要的多个kafka集群用例，让我们看看再实现这些用例时我们成功使用的一些常见的架构模式。在我们进入架构之前，我们将简要概述跨数据中心通信的实现。我们将讨论的解决方案可能看起来过于复杂，而没有理解他们代表了面对特定网络的权衡。

Some Realities of Cross-Datacenter Communication 一些跨数据中心的实现

下面列出了跨数据中心需要考虑的一些情况:

• High latencies 两个kafka集群中间的通信延迟随着两个集群之间的距离和网络中转次数的增加而增加。
• Limited bandwidth 带宽限制，广域网的可用带宽通常比你再单个数据中心看到的要低很多，而且可用带宽每分钟都在变化，此外更搞的延迟使得利用所有的可用带宽更加困难。
• Higher costs 更高的成本，无论你实在本地允许kafka还是在云中运行，集群之间通信的成本都比较高，这在一定程度上是因为带宽受限，增加带宽的成本太高，而且供应商在数据中心，区域和云之间传输数据的费用太高。 Apache kafka的broker和客户端都是在一个数据中心内设计、开发、测试和调优的，我们假设broker和客户机之间存在低延迟和高带宽。这在各种缓冲区的默认超时和大小调整中都很明显。处于这个原因，不建议（除非在特定情况下，我们将在后面讨论）在一个数据中心安装kafka集群的一部分broker，在另外一个数据中心安装集群的另外一部分broker。 在大多数情况下，最好避免向远程数据中心生成数据，在这样左的时候，你需要考虑更高的延迟和更多的网络错误出现的可能性。你可以通过增加生产者重试次数来处理错误，并通过增加保存记录的缓冲区的大小来处理较高的延迟。 如果我们需要在集群之间进行任何类型的复制，并且排除broker之间的通信和生产者到broker之间的通信，那么我们必须运行broker和消费者之间的通信。事实上，这事最安全的跨集群的通信方式，因为在网络可能影响消费者读数据的情况下，消息记录kafka上broker中是安全的，一直到通信恢复，消费者都可以读取他们。不存在由于网络而导致分区数据丢失的风险。但是，由于带宽受限，如果一个数据中心的多个应用程序需要从另外要给数据中心的kafka的broker读取数据，我们更常用的做法是在每个数据中心都安装一个kafka集群，并在他们之间进行镜像，拷贝必要的数据，而不是让多个应用跨wan使用相同的数据。 我们将更多的讨论kafka跨数据中心通信的调优，但是下面的指导原则则是我们接下来讨论大多数数据架构的基础：
• 每个数据中心不少于一个集群。
• 每个数据中心的消息实现精确一次的复制。
• 如果可能，从远程数据中心消费数据而不是将数据写入远程数据中心。

Hub-and-Spokes Architecture 中心辐射架构

这种架构适用于多个本地的kafka集群和一个中央的kafka集群的情况。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Vra6Mzx2-1594726806051)(EDA429CA2E604C2CB50787628238C302)] 这个体系结构还有一个更简单的变体，它只有两个集群，leader和follower。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-pDGF9BcV-1594726806052)(5AD6774D81124D18B99AC40D7FA27619)] 当数据在多个数据中心生成，并且一些消费者需要访问整个数据集的时候，就会使用这种架构，这种架构还运行每个数据中心的应用程序只处理该数据中心本地的数据，但是它不允许访问每个数据中心的完整数据集。 这种体系架构的最主要的优点就是数据始终生成到本地数据中心，而来自每个数据中心的消息只会被镜像到数据中心一次。处理来自单个数据中心的数据的应用程序可以位于该数据中心，需要处理来自多个数据中心的数据的应用程序位于所有事件都被镜像的中心数据中心。由于复制总是单向的，而每个消费者总是从相同的集群读取数据，因此该体系结构易于部署，配置和监控。 这种体系结构的最主要的缺点是其优点和简单性的直接结果。一个区域数据中心的处理器不能访问另外要给区域数据中心的数据。为了更好地理解为什么这事要给限制，让我们看一下这种架构的示例。 假设我们是一家大型银行，在多个城市都有分行，假设我们决定在每个城市的kafka集群种存储用户配置文件和他们的账户历史，我们将把所有的这些信息复制到一个中央集群种，用于允许银行的业务分析。他们将被路由到本地集群发送事件，并从同一个本地集群种读取事件，但是，假设用户访问另外一个城市的分支。因为用于信息在其访问的城市中不存在，分支将被迫与远程集群交互，或者无法访问到用户信息。由于这个原因，此模式的使用通常仅限于数据中心可以在区域数据中心之间完全分离的部分。 在实现此体系结构时，对于每个区域数据中心，你需要在中心数据中心上至少有一个镜像过程。这个过程将使用来自每个远程区域集群的数据，并将其生成到数据中心集群。如果多个数据中心中存在相同的topic，你可以将此topic中的事件写入到中心集群中同名的topic中，或者每个数据中心中的事件写入一个单独的topic。

Active-Active Architecture 双活架构

当两个或者多个数据中心共享部分或者全部数据并且每个数据中心都能够生成和使用事件时，将使用这个体系架构： [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mqoJmE67-1594726806054)(79FFA3F8E2EE44A59C925F71069722E5)] 这种体系架构的主要好处是能够为附近的数据中心的用户提供服务，这通常会带来性能方面的好处，而不会因为数据可用性有限而牺牲功能，如我们在中心辐射架构中看到的那样。第二个好处就是冗余和弹性，由于每个苏剧中心都具有所有的功能，如果一个数据中心不可用，你可以将用户定向到其他数据中心。这种类型的故障转移只需要对用户的网络进行重定向，这通常是最简单和最透明的故障转移类型。 这种体系架构的最主要的缺点就是，在多个位置异步读取和更新数据时，避免冲突是很困难的。这包括镜像事件的技术挑战。丽日，我们如何确保相同的事件不会无休止的来回镜像》单更重要的是，维护两个数据中心之间的数据一致性是困难的，下面是一些你将会遇到的困难的例子：

• 如果用户向一个数据中心发送事件，并从另外要给数据中心读取事件，则有可能他们编写的事件尚未到达第二个数据中心，这对于用户来说，这看起来就像他刚刚在他的购物车中添加了一本书，点击了购物车之后，书并不存在。因此在使用这种体系结构时，开发人员通常会找到一种方法，将每个用户固定到特定的数据中心，并确保他们大多数时候使用相同的集群。（除非他们从远程位置连接或者数据中心不可用）。
• 来自一个数据中心的事件表示用户订购了图书A，而是来自第二个数据中心的事件或多火烧同时表示同一用户订购了图数A就像后，两个数据中心都有两个事件，因此我们可以说每个数据中心都有两个冲突事件。两耳数据中心上的应用程序都需要知道如何处理这种情况，我们是否选择一个正确的事件，如果这也，我们需要一致的规则来选择一个数据中心，这样两个数据中心上的应用程序会得出相同的结论。我们是否决定两者都为真，并简单地向用户发送两本书，并让其中一个部门处理退货？亚马逊曾用这种方式处理冲突，但是以处理股票交易的组织为例，这是行不通的。最小化冲突并在冲突发生时处理他们具体的方法是特定于每个用例的。需要特别注意的是，一旦你选择这种架构，你将会有冲突发生，需要妥善处理。 如果你找到了一些方法来处理从多个位置异步读写相同的数据集的挑战，那么强烈建议使用这种体系结构。这事我们所知道的最具有可扩展性、弹性、灵活性和成本效益的选择。因此，很有必要努力找出解决方案，以避免复制周期，将用户大部分保持在相同的数据中心以及在发生冲突时处理冲突。 双活架构其特别是对于超过两个的数据中心而言，最重要的挑战是在于，你需要将每个数据中心对每个数据传输的方向都需要实现一个镜像。对于5个书中心，你需要维护的镜像可能是20，也有可能是40，因为每个进程都需要冗余来实现高可用性。 此外，你将希望避免重复映射相同的事件的循环。为此，可以为每个数据中心的每个逻辑topic提供一个单独的topic，并确保避免复制源自远程数据中心的topic。例如，topic users中的SF.users在一个数据中心，NYC.users在另外一个数据中心。镜像进程将镜像SF.users从SF到NYC，将NYC.users从NYC到SF。因此，每个事件只会镜像一次，但是每个数据中心同时包含SF.users和NYC.users。这意味着每个数据中心将拥有所有的用户信息。消费者如果希望消费所有的用户事件，则订阅 .users事件即可。考虑到这种设置的另外一种方法是将其视为每个数据中心的单独命名空间，其中包含特定数据的中心和所有的topic，在我们的示例中，我们将拥有NYC和SF的名称空间。 在不久的将来，Apache kafka将添加记录头，这将允许用原始数据中心标记事件，并使用此头信息来避免无休止的镜像循环，还允许分别处理来自不同数据中心的事件消息。

Active-Standby Architecture 主从架构

再某些场景下，多集群的唯一需求是支持某种灾难场景。也许在同一个数据中心有两个集群，对于所有应用程序，你使用一个集群，你希望第二个集群包含原始集群中的所有消息，如果原始集群完全不可用，你可以使用这些消息。或者你需要跨地区做冗余，你的整个业务都是在加利福尼亚的数据中心进行的，但是你需要第二个苏剧中心应对突发情况如地震等。跨区域的数据中心可能处于一个非活动的状态，做为所有程序的副本，管理员在紧急情况下启用这些应用程序，这些应用程序将做为第二个集群。这通常是法律上的要求，而不是企业实际计划要做的事情，但是你仍然要做好准备。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0KbVL1Fg-1594726806055)(742C1D13C3A14C5BA502192884B815FC)] 这种设置的好处就是简单，可以在任何用例中使用，你只需要安装第二个集群并设置一个镜像进程，将所有的消息从一个集群流到另外一个集群，不需要担心数据访问，处理冲突和其他架构的复杂性。 但是缺点在于这样浪费了一个好的集群，而且kafka集群之间的故障转移实际上比看起来要困难得多。最重要的是，目前kafka中不可能在不丢数据或者不存在重复数据的情况下执行集群的故障转移，通常这两种情况下，你可以最小化他们，但是永远不能完全消除。 显然，除了等待灾难之外什么都不做的集群是一种资源浪费。由于灾难是罕见的，大多数时候我们看到的是一个什么也不做的集群。一些组织尝试使用比生产集群小得多的灾难恢复集群来解决这个问题。但是这是要给危险的决定，因为你不能确定这个最小规模的集群在紧急情况下是否能够运行。其他的组织希望通过将一些只读的工作负载转移到DR集群上运行，从而使集群在非灾难情况下能发挥作用。这意味着他们实际上运行的是带有单个辐射型架构的一个小版本。 更严重的问题是，如果在故障发生的时候实现故障转移到Apache Kafka 的DR集群？ 首先，无论你使用那种故障转移方法，你的SRE团队必须经常练习。今天还有效的计划可能在升级之后停止工作。胡总可能新的用例使现有的工具过时。 对于故障转移的实践来说，每季度进行一次通常是最少的，强大的SRE团队的训练频率要高得多。Netflix著名的Chaos Monkey ，导致灾难的通常是随机情况，每天都可能成为故障转移的练习日。 现在，让我们看看故障转移failover涉及哪些内容：

Data loss and inconsistencies in unplanned failover 计划外的failover导致数据丢失和不一致

因为kafka的各种镜像解决方案都是异步的，所以DR集群不会有主分片集群的最新消息。你应该始终对DR集群的落后程度进行监控，不要让它落后太多。但是在一个繁忙的系统种，你应该预期DR集群在主服务器后面有几百甚至几千条消息。如果你的kafka集群每秒处理100万条消息，并且主集群和DR集群之间存在5ms的延迟，那么在最好的情况下，DR集群将比主集群落后5000条消息。因此，为计划外的故障转移做好准备，以包括一些数据丢失，在计划的故障转移种，你可以停止主集群并等待镜像进程在将应用程序故障转移到DR集群之前的镜像剩余消息，从而避免这种情况导致的数据丢失。计划外的故障转移发生时，你失去了几千的消息，注意，kafka目前还没用事务的概念，这意味着，如果多个topic是有相互关联的一些消息，有可能只有一部分消息到达了DR节点。

Start offset for applications after failover 故障转移之后的起始offset

在出现故障转移到另外一个集群的时候，最具有挑战性的部分可能始在确保应用程序知道从何处来说使用数据，有几张常见的方法，有些很简单，但是可能导致额外的数据丢失或者重复的处理，其他的则更加复杂，但尽量减少额外的数据丢失和重复处理，然我们看看这些参数：

Auto offset reset 自动重置offset

Apache kafka的消费者有一个配置，用于他们无法知道预先提交的offset时候的行为：他们可以从分区的开始部分开始读取，也可以从分区的结束部分开始读取。如果你使用的是zookeeper提交offset的老的客户端，而你没用将这些offset做为DR的一部分的话，那么你需要从这些选项种做出选择。要么从分区的开头部分开始读取，并处理大量重复的数据，要么跳到末尾，忽略一些消息。如果你的应用程序能够处理重复的消息，或者丢失一些数据没有什么大问题，那么这个选项是比较容易的。选择最近的消息是最流行的故障转移办法。

Replicate offsets topic

如果你正在使用新的0.9.0版本以上的kafka消费者，消费者将把他们的offset承诺给一个特殊的topic，__consumer_offsets。对于你的DR集群，当消费者开始从DR集群消费时，他们将能够找打旧的offset,从他们之前暂停的地方开始己洗消费，这很简单，但是可能会触发一长串警告。 首先，不能保证topic主集群种的offset与辅助集群中的offset匹配。假设只在主集群中存储三天的数据，并且在创建topic一周后开始进行镜像操作。在这种情况下，主集群的第一个可用offset可能是57000000，较早的事件来自第四天，并且已经被删除。但是DR集群中的第一个offset将是0，因此，试图要读取57000003的消费在DR中将无法实现。 其次，即使你在topic的第一次创建topic时开始镜像，并且主题topic和DR的topic都以0开始，生产者重试也会导致offset偏离。简单地说，现有的kafka的镜像解决方案没有保留主集群和DR集群之间的offset。 第三，即使offset被完美的保存下来，由于主集群的DR集群之间的延迟以及kafka目前缺乏事务，kafka消费者提交的offset可能会提前或者晚于带有这个offset的记录，故障转移的消费者可能会发现这个已提交的offset没有匹配记录，或者它可能发现DR站点中最新提交的offset比主站点中最新提交的offset更早。参见下图： [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-V69LhSHO-1594726806056)(AD2F215D5996457C918C91048D986B25)] 在这些情况下，如果DR中最新提交的offset比主服务器上提交的更早，或者由于重试，则DR站点中记录中的offset高于主服务器，则需要接收一些重复。还需要弄清楚如何处理最近提交的offset。DR站点没有匹配的记录，你是从topic的开头开始处理，还是跳到结尾？ 可以看到，这种方法具有局限性。不过，与其他的方法相比，这个选项允许你将故障转移到另外一个DR，而重复或者丢失的事件的数量更少，同时易于实现。

Time-based failover 基于时间的故障转移

如果你使用的是0.10.0以上版本的kafka消费者，则每个消息都包含一个时间戳，表示消息被发送到kafka的时间，在真正新的kafka版本中（0.10.1.0以上），broker包括索引和用于按时间戳查找offset的API，因此，如果你将故障转移到DR集群，并且你知道问题开始于凌晨4点05分，你可以告诉消费者，从凌晨4点03分开始处理数据。 这两分钟会有一些重复，但是这可能比其他选择更好，而且我们这种行为更容易向公司的每个人解释：我们回到了4点03分故障之前。这听起来比我们没能回到最近承诺的commit offset上要好。所以这通常是一个很好的妥协。唯一的问题是，我们如何告诉消费者从凌晨4点03分开始处理数据？ 有一个用户可配置的选项来指定应用程序的开始时间。如果配置了这个选项。ing从程序可以使用新的api按时间获取offset。查找到哪个时间。并从正确的点开始使用，像往常一样提交offset。 如果你提交前以这种方式编写了所有的应用程序，那么这个选项非常棒。但是如果你没有呢？编写一个小的工具来获取时间戳和offset，然后做为特定的消费者组提交topic和分区列表的offset，这是相当简单的。我们希望在不久的将来把这个工具添加到kafka中，但是你也可以自己编写一个。消费者组应该在允许此类工具的时候停止，然再重新启动。 这个选项推荐给哪些使用kafka新版本的用户，他们希望在故障转移方面有一些确定性，并且愿意围绕这个过程编写一些自定义工具。

External offset mapping 外部offset映射

在讨论topic的镜像offset主题的时候，这种方法最大的挑战在于，主季芹和DR集群的offset可能会出现分歧。考虑到这一点，一些组织选择用外部数据存储如（Cassandra）来存储从一个集群到另外一个集群的offset映射。他们构建了kafka镜像工具，这样当一个消息产生到DR集群的时候，两个offset都被发送带外部数据存储中。或者当两个offset之间的差发生变化时，他们只存储两个offset。丽日如果主服务器上offset 495被映射到DR集群上的offset500，我们将在外部存储上记录(495,500)。如果稍后由于重复而导致差异发生变化，并且将offset 590映射位600，那么我们将记录新的映射(596,600)。 不需要存储495和596之间的所有offset 映射。我们假设差异是相同的，因此主集群的offset位550将映射到555.那么当发生故障转移的时候，他们将主集群中的offset映射到DR的pffset并使用这些offset。而不是将时间戳映射到offset（时间戳总是不准确）。 他们使用上面列出的两种技术之一来强制消费者开始使用映射中的新的offset，对于提前到达的记录本书的offset提交以及没有及时镜像到DR的offset提交。者仍然存在一个问题，就是他覆盖了一些数据。 这个解决方法非常复杂，在我看来，几乎是没有必要的。它可以追溯到时间所有存在之前可以用于故障转移。则夏天，我选择升级集群并使用基于时间的解决方案，而不是进行offset映射，因为它不能涵盖所有的故障转移的情况。

After the failover 故障转移之后

假设故障转移成功了，DR集群上的一切工作都很顺利。现在我们需要对主集群做一些事情，也许把它变成一个DR集群。简单的修改镜像过程来逆转他们的方向，交道的从新的主镜像到旧的主镜像是很诱人的，但是者导致了两个重要的问题。

• 我们怎么指定从哪里开始镜像？我们需要位镜像应用程序本书的所有消费者解决同样的问题。记住，我们所有的镜像解决方案都有导致重复或丢失数据的情况，通常是两种情况都有。
• 另外，由于我们上面讨论的原因，你最初的主服务器可能具有DR集群所没有的消息，如果你只是开始将新数据镜像回来，那么额外的历史将保留的话，这两个集群将不一致。

基于这些原因，最简单的解决方案是首先删除原始集群，删除所有的数据和已提交的offset，然后开始从新的主集群镜像到现在的新的DR集群，这将给你一个新的与初选完全相同的全新的开始。

A few words on cluster discovery 关于集群发现的一些说明

在规划备用集群时，需要考虑的重要的一点就是，在发生故障转移的时候，应用程序需要指定如何开始与故障转移集群通信。如果将主集群broker的主机名硬编码到生产者和消费者属性中，这是一个挑战。大多数组织为了保持简单，创建了一个DNS名称指向主集群。在紧急情况下，DNS名称可以指向备用集群，发现服务，DNS或者其他手段不需要包括所有的kafka broker，客户端只需要成功访问单个broker,就可以获得关于就请你的元数据并发现其他的broker。因此，只包括三个broker通常就可以。无论使用那种方法来实现服务发现，大多数故障转移的场景都要在故障转移之后切换到消费者的应用程序。使得他们能够找到需要开始使用的新的offset。

Stretch Clusters 集群扩展

主备体系架构用于保护业务免受故障影响。通过切换应用程序与集群中通信，以防止集群宕机带来的业务损失。扩展集群的目的是在数据中心发生故障的时候保护kafkajiqun不发生故障。通过跨多个数据中心安装一个kafka集群来实现这一点。 扩展集群与其他多数据中心场景有着根本上的不同，首先，他们不是多集群，它只是一个集群。因此我们不需要镜像程序来保持两个集群的同步。像往常一样，使用kafka的常规复制机制来使集群中的所有broker保持同步。此设置可以包括同步复制。 在扩展集群的情况下，我们可以配置一些东西，以便在消息成功写入两个数据中心的kafka broker之后发送确认。这涉及到使用机架定义来确保每个分区在多个数据中心都有副本，并且使用min.isr和acks=all来确保至少两个数据中心已经写入成功。 这种架构优势在于同步复制，有些类型的业务仅仅要求他们的DR站点始终与主集群100%同步。这通常是一项硬性要求。适用于整个公司的任何数据存储，包括kafka，另外一个优点是可以同时使用数据中心和集群中的所有broker,没有像主从架构那样浪费节点资源。 这种体系结构在它所包含的灾难类型上是有限的，它只保护数据中心故障，而不是任何类型的应用程序或者kafka故障。菜煮的复杂性也是有限的，这种体系结构需要物理基础设施的支持，比如专线等，这不是所有公司都能提供的。 如果你能在至少三个数据中心安装kafka和zookeeper集群，并且他们之间有高带宽和低延迟，那么这种架构就是可行的，如果你的公司在同一街道上有三栋大楼，或者更常见的做法是在云服务商的以给区域内使用三个可用区域就可以做到这一点。 三个数据中心之所以重要，是因为zookeeper在集群中需要的不均匀的节点数量，并且在大多数节点可用的情况下仍然可以使用，对于两个数据中心和不均匀的节点数，要给数据中心将始终包括大数据节点，这意味着如果该数据中心不可用，那么zookeeper和kafka都不可用。有了三个数据中心，你就可以轻松的分配节点，这样单个数据中心就不会占多数，因此如果一个数据中心不可用，那么其他两个数据中心的大多数节点都存在，zookeeper集群仍然可用，因此kafka集群也将如此。 可以用zookeeper组配置在两个数据中心运行zookeeper和kafka，运行在两个数据中心手动进行故障切换，但是这种设置通常不常见。

Apache Kafka’s MirrorMaker

Apache Kafka包含了一个简单的工具，用于在两个数据中心镜像数据，它被称为MirrorMaker,其核心是一个消费者的集合，称为streams in ,他们都属于同一个消费者组，并从选择复制的topic中读取数据。每一个MirrorMaker过程中只有一个生产者。工作流非常简单。MirrorMaker为每个消费者运行一个线程。每个消费者使用源集群上分配给他的topic和分区中的事件消息，并使用共享的生产者将这些消息发送到目标集群。每隔60秒消费者将告诉生产者它拥有的所有事件消息发送给kafka并等待kafka确认这些消息。然后消费者联系源kafka集群，提交这些事件的offset。这些保证没有了数据丢失，在向源提交offset之前，kafka会确认消息。而且如果mirrormaker进程崩溃，数据重复的时间不会超过60秒。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LTfGZiDK-1594726806058)(A51CB6CDCAC74C2E8C8B8C533CBEB533)]

More about MirrorMaker 关于MirrorMaker

MirrorMaker听起来很简单，但是因为我们试图提供Exactly-once的数据交付，这是一个非常困难的需求。在Apache Kafka 0.10.0.0中，MirrorMaker已经被重写了四次，将来还可能重写。这里描述的章节适用于0.9.0.0-0.10.2.0之间版本的mirrorMaker。

How to Configure 如何配置

MirrorMaker具备高可配置性，首先，它使用一个生产者和多个消费者，因此在配置MirrorMaker时可以使用生产者和消费者的每个配置属性。此外，MirrorMaker本身有一个相当大的配置参数列表。有时候他们之间有复杂的依赖关系。我们将在此展示一些重要的配置。但是关于Mirrormaker的详细配置不在此赘述。 让我们看看启动的例子：

bin/kafka-mirror-maker --consumer.config etc/kafka/consumer.properties --
producer.config etc/kafka/producer.properties --new.consumer --num.streams=2 --
whitelist ".*"

逐一对命令行参数进行解析：

consumer.config

这个配置指定了所有的消费者将从源集群中获取数据，他们都共享一个配置文件，这意味着你只能有一个源集群和group id 。所以所有的消费者都是同一个消费者组的一部分，这正是我们想要的。文件中的强制配置是bootstrap.servers和group.id。你可以为消费者使用任何其他的配置。但是auto.com mit.enable=false是你不想涉及的。MirrorMaker自动提交offset。更改此配置可能会导致数据丢失。auto.offset.reset这个配置的默认值是latest，这意味着MirrorMaker启动后的源消息才会被复制。如果你希望复制已存在的其他数据，那么你需要将这个参数更改为earliest。我们将在后面讨论关于MirrorMaker的调优。

producer.config

MirrorMaker写入目标集群的生产者配置。你唯一需要配置的是bootstrap.servers参数。我们将在后续的调优章节详细介绍。

new.consumer

MirrorMaker可以使用0.8或者0.9以后的消费者，我们推荐将此配置设置为new,用以配置新版本的客户端。

num.streams

如前面所述，每个流是另外从源集群读取的消费者，请记住，同一个MirrorMaker的所有消费者使用相同的生产者。如果你在此之后需要额外的吞吐量，那么你需要另外一个MirrorMaker进程。

whitelist 白名单

将要被MirrorMaker使用的消费者的topic名称的正则表达式。被这个正则表达式匹配的所有topic名称都将被镜像。在本例中，我们选择复制每个topic，但是使用类似prod.* 则可以复制所有prod的topic。在双活的架构中，从NYC数据中心复制到SF数据中心，那么MirrorMaker的配置 whitelist=“NYC.*” 从而避免复制SF的topic。

Deploying MirrorMaker in Production 生产环境中部署

在前面给出的示例中，我们将MirrorMaker做为命令行程序运行。通常，在生产环境使用MirrorMaker的时候，你希望将MirrorMaker做为服务运行。使用nohup在后台运行并将其控制台输出重定向到日志文件中。从技术上来讲，使用-daemon可以完成上述所有工作。但是实际上，在最近的版本中，MirrorMaker并没有支持这个参数。 大多数使用MirrorMaker的公司都有自己的启动脚本，其中也不啊看他们使用的配置参数，生产部署系统如Ansible、puppet、chef和salt通常用于自动化部署和管理许多配置选项和文件。 一种更先进的配置选择是选项在docker镜像中，这是一种非常流行的方案。MirrorMaker是完全无状态的，不需要任何磁盘存储。所有的数据都存储在kafka本身。包装在Docker中的MirrorMaker也运行在一台机器上运行多个实例。由于单个MirrorMaker实例被限制为单个生产者的吞吐量。因此启动MirrorMaker的多个实例是非常重要的，而Docker使这一工程更加容易。 它还使得在高峰时期需要更多吞吐量的时候更容易扩展和在流量少的时候更容易收缩。如果你在云环境运行MirrorMaker。甚至可以根据吞吐量和需求启动运行容器的附加服务器。 如果可能的话，在目标数据中心运行MirrorMaker,因此，你送纽约发送数据到旧金山。MirrorMaker应该运行在旧金山，从纽约读取数据。这也做的原因是，长途网络的可靠性比数据中心的网看要差一些。如果存在网络分区，并且数据中心之间丢失了连接，那么让我飞连接到集群的消费者比让无法连接到集群的生产者更安全。如果消费者无法连接，它将无法读取消息，但是消息仍让存储在源的kafka中。可以在哪里停留一段时间。没有数据丢失的风险。另外一方面，如果数据已经被消费，而MirroMaker由于网络原因而无法写入，那么这些消息就有被MriiorMaker意外丢失的风险。所以远程消费比远程生产更安全。 什么时候你必须在本地消费和远程生产？答案是当数据在数据中心传输需要加密时，但是不需要加密数据中心内部的数据。当使用SSL加密连接到kafka时，消费者会比生产者遭受到更大的性能损失，而且这种性能损失也会影响到kafka broker的本身。如果你跨数据中心流量加密，你最好设置MirrorMaker让它在本地使用未加密数据，然后通过SSL加密连接的方式写入到远程的数据中心，通过这种方式，生产者使用的SSL连接到kafka而不是消费者，这对性能影响不大。如果你使用本地消费而远程生产，请确保将MrirrorMaker配置为永不丢失消息，acks=all。和足够的重试次数。此外，将MirrorMaker配置为在发送消息失败时退出通常比继续冒着数据丢失的风险发送数据更安全。 如果源集群和目标集群之间有很低的延迟，这是非常重要的。那么你可能希望在两个不同的服务器上运行至少两个实例。并谈他们使用同一个消费者组。如果某个服务器由于某种原因宕机，MirrorMaker实例可以继续镜像数据。 在生产环节中部署MirrorMaker时，一定要记住监视如下数据：

Lag monitoring

你肯定想指定目标集群是否落后于源集群，之后是指源kafka中的最新消息和目标中的最新消息之间offset的差异。 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gziBVSTL-1594726806059)(198A9BF04C7F4EB3AFA9EC5530498B3B)] 源集群中的最后一个offset是7，目的中的最后一个offset是5，这意味着有2条消息的延迟。 有如下两种方式可以追踪这些差异，但是都不完美：

• 检查MirrorMaker提交给源kafka集群的最新offset。可以使用kafka-consumer-groups工具检查每个分区。MirrorMaker正在读取分区中的最后一个事件的offset，MirrorMaker提交的最后一个offset以及它们之间的延迟，这个指标不是100%的准确，因为MirrorMaker并不是一直都有offset。默认情况下，它每分钟提交一次offset。因此你将在你的图表中看到延迟会增加1分钟然后突然下降。在图表中，真正的滞后是2.但是kafka消费者组工具报告的延迟是4。因为MirrorMaker还没有对最近的信息做出commit。LinkedIn的Burrow监视着同样的信息，但是有一个更复杂的方法来缺点延迟是否代表一个真正的问题，所以你不会收到错误的提醒。
• 检查MirroMaker读取最新offset，即使他没有提交，嵌入到MriiorMaker中的消费者在JMX中发布关键指标，其中之一就是消费者最大延迟（超过它正在使用的所有分区）。这个延迟也不是100%的准确，因为它是根据消费者读取的内容进行更新的，而没有考虑生产者师傅成功地将这些消息发送到目的kafka集群以及它们是否被成功的提交。在本例中，MirrorMaker消费者报告1条消息延迟，而不是2条，因为已经读取了消息6，即使消息还没有发送到目的地。

注意，如果MirrorMaker跳过或者删除消息，着两种方法都不会被检测到，因为它们只耿总最新的offset。confluent在控制中心监控消息的技术和校验和，并缩小此监控的差距。

Metrics monitoring 监控指标

MirrorMaker包含一个生产者和一个消费者，两者都有许多可用的度量指标，我们建议收集和跟踪它们，kafka的官方文档还列出了所有的可用的度量指标，下面是一些被真没在调优中有永的指标。 MirrorMaker性能：

• Consumer fetch-size-avg, fetch-size-max, fetch-rate, fetch-throttle-time-avg, and fetch-throttle-time-max
• Producer batch-size-avg, batch-size-max, requests-in-flight, and recordretry- rate
• Both io-ratio and io-wait-ratio

Canary

如果你监视其他所有内容，那么Canary不是绝对必要的，但是我们喜欢添加它来进行多层监视。它提供了一个进程，该进程每分钟向源集群中的特殊topic发送一个事件，并尝试从目标集群中读取事件。如果事件到达的事件可能超过可接收的时间，它还会发出警报。这意味着MirroMaker落后或者已经崩溃。

Tuning MirrorMaker MirrorMaker调优

MirrirMaker集群的大小取决于你需要的吞吐量和你能忍受的延迟。如果你不能容忍任何延迟，你必须配置足够多的MirrorMaker以让其能跟上最高的系统吞吐量。如果你能忍受一些延迟，你将MirrorMaker调到原有的75-80%，在95%-99%的时间内都能满足需求。当你处于峰值的吞吐量的时候，可能会出现一些延迟。由于MirrorMaker大部分时间都能进行复制，那么一旦高峰期结束，它将会追赶上来。 然而，你需要使用不同的数量的消费者线程，使用num.streams参数配置。来度量从Mirrormaker获得的吞吐量。我们可以给一些大概的建议数字。在linkedin,8个消费者线程可以达到6MB/s,12个消费者线程可以达到16MB/s。 但是由于这很大程度上取决于硬件，数据中心或者云供应商。因此你需要自行测试。kafka有一个kafkaperformance-producer工具。使用它在源集群上生成负载，然后连接MirrorMaker.就可以对MirrorMaker进行负载测试。使用1 2 4 8 16 24 和32个消费者线程。观察性能下降的位置，并将number.streams设置在这个点以下。如果使用或者生成压缩，推荐使用，因为带宽是跨数据中心镜像的主要瓶颈。MirroMaker将不得不解压和重新压缩消息。这会大量使用CPU，因此在增加线程数量时，请密切关注CPU的利用率。在此使用过程中，你将发现单个MirrorMaker实例可以获得最大的吞吐量，如果这还不过，你将希望使用其他的实例进行实验，然后使用其他的服务器。 此外你可能希望将敏感的topic（哪些绝对需要延迟以及镜像必须尽可能靠近源的topic）分离为单独的topic。MirrorMaker集群有自己的消费者组。这将防止topic膨胀或者生产者失控造成对topic的雍塞，让敏感的topic变慢。 这几乎是你可以对MirrorMaker机械能的所有调优，但是，你任然可以增加每个消费者线程和每个MirrorMaker实例的吞吐量。 如果你在跨数据中心运行MirrorMaker,你需要在Linux中优化网络配置，如下所示：

• 增加TCP缓冲区的大小(net.core.rmem_default, net.core.rmem_max, net.core.wmem_default, net.core.wmem_max, net.core.optmem_max)
• 启动TCP窗口缩放 sysctl –w net.ipv4.tcp_window_scaling=1 or add net.ipv4.tcp_window_scaling=1 to /etc/sysctl.conf
• 减少TCP慢启动时间 (set /proc/sys/net/ipv4/ tcp_slow_start_after_idle to 0) 注意，调优Linux网络是一个庞大而复杂的主题，要连接更多关于这些参数和其他参数的信息，我们建议阅读网络调优指南，如Sandra K. Johnson等人编写的《Linux服务器性能调优》。 此外，你可能希望调优正在运行的MriiorMaker的生产者和消费者。生产者是在等待消费者带来更多的数据，还是在等待消费者带来更多的数据？一种决定方法是查看你正在监控的生产者和消费者的指标，如果一个进程是空闲的，另外一个进程是充分利用的，那么你指定哪个进程需要调优，另外一种方法是执行几个线程转储如jstack。并查看MirrorMaker线程是否将大部分时间用于轮询或者发送，轮询花费的时间越多，通常意味着瓶颈是消费者。而将更多的时间用于向生产者发送转移数据。 如果你需要对生产者进行调优，那么如下几个参数可能有用：
• max.in.flight.requests.per.connection 默认情况下，MirrorMaker只允许一个飞行请求。这意味着，在发送下一条消息之前，生产者发送的每个请求都必须得到目标集群的确认。这可能会限制吞吐量。特别是在broker确认消息之前存在显著的延迟的时候。MirrorMaker限制飞行请求数量的原因是，这是在某些消费需要多次重试才能成功确认之前，确保kafka消息保持顺序性的唯一方法。如果消息顺序对你来说不是关键的，那么增加max.in.flight.requests.per.connection能够显著提升吞吐量。
• linger.ms and batch.size 如果你的监控中，生产者始终在发送一些不满的批次，即batch-size-avg 和 batch-size-max指标低于配置的batch.size.则可以通过引入latency.ms来增加吞吐量。生产者在发送之前等待几毫秒，等待批次消息填满。如果你正在发送完整批次，并且有空闲的内存，你可以增加batch.size或者发送较大的批次。 以下配置可以增加消费者的吞吐量： MirrorMaker中的分区分配策略，即确保哪个消费者被分配到哪个分区的算法，缺省值为range.这个策略有一些好处，这是为什么它是消费者的常规默认值的原因。但是它可能会导致分区分配不均匀。MirrorMaker通常最好将策略改为轮询调度。特别是在对镜像大量的topic和分区的时候，你可以通过添加partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor 到消费者的配置中。 -fetch.max.bytes 如果你收集的指标显示取值fetch-size-avg和fetch-size-max低于fetch.max.bytes。消费者正在从broker中读取尽可能多的数据，如果你有可用的内存，尝试增加fetch.max.bytes以允许消费者从每个请求中读取跟多的数据。
• fetch.min.bytes 和 fetch.max.wait 如果在消费者中的指标中看到者两个值很高，则说明消费者向broker发送了太多的请求，每个请求中没有收到足够多的数据，通过增加fetch.min.bytes和fetch.max.wait让消费者收到更多的数据。在broker中直到有足够的数据可用，然后才响应消费者的请求。

Other Cross-Cluster Mirroring Solutions 其他的镜像解决方案

Uber uReplicator

Uber大规模的运营了MirrorMaker,随着topic和分区数量的增加，以及集群的吞吐量的增加，Uber开始遇到了如下问题：

Rebalancing delays 重平衡延迟

mirrorMaker的消费者只是普通的消费者，添加MirrorMaker的线程，添加MriiorMaker的实例，关闭MirrorMaker的实例，甚至添加与白名单中使用的正则表达式匹配的新的topic，这都会导致用户的reblance。正如我们在第四章所看到的那样，reblance会停止所有的消费者，直到可以为每个消费者分配新的分区。对于大量的Topic和分区，这可能会花费很多时间。在Uber中尤其如此，在某些情况下，可能会导致5-10分钟的卡顿，这将导致镜像之后并累计大量需要镜像处理的事件，这可能需要很长的时间才能恢复，这会导致消费者从目标集群读取消息的延迟非常高。

Difficulty adding topics 添加topic困难

使用正则表达式做为消费者订阅topic的配置意味着每当有人向集群添加匹配topic的时候，MirrorMaker将重新进行reblance。我们已经看到reblance对Uber来说尤其痛苦。为了避免这种情况，他们决定简单的列出他们需要镜像的每个topic，以避免意外的reblance。这意味着他们需要手动添加他们想要在所有MirrorMaker实例上进行镜像的所有topic。一旦重启势必又会产生reblance。至少这些reblance发生在计划维护的时候，并不是每次有人添加一个topic它就会发生。这也意味着，如果维护没有正确完成，不同的实例有不同的topic列表，Mirrormaker将会无休止的执行reblance。因为用户将无法就他们订阅的topic达成一致。 考虑到这些问题，Uber决定编写他们自己的镜像软件uReplicator。他们决定使用Apache Helix做为中心连接，它将管理topic列表和分配给每个uReplicator实例的分区。管理员通过RestAPI向Helix中的列表添加新的topic，而uReplicator负责将分区分配给不同的消费者。为了实现这一点，Uber重写了kafka中的消费者取代了MirrorMaker中使用的kafka的消费者。他们被称为Helix consumer.这个消费者从ApacheHelix控制器中接收分区分配，而不是由消费者之间达成的协议。因此Helix consumer.可以避免reblance。而是监听从Helix 中分区的更改。 Uber写了一些博客文章，更详细的描述了该架构，并展示了他们所经历的改进，在撰写本文时，我不知道除Uber之外还有哪些公司使用了uReplicator。这可能是大多数公司都没有达到Uber的规模，也没有遇到相同的问题，或者因为对ApacheHelix的依赖导致引入了一个全新的组件来学习和管理，这增加了整个项目的复杂性。

Confluent’s Replicator

在Uber开放自己的uReplicator的同时，Confluent独立开发了Replicator。尽管名称相似，但是这些项目几乎没有任何共同点，他们是针对两组不同镜像的不同解决方案。Confluent和Replicator是为了解决企业客户在使用MirrorMaker管理多集群部署的时候遇到的问题而开发的。

Diverging cluster configuration

虽然MirrorMaker保持源和目标之间的数据同步，但是这是它唯一保持同步的东西。topic最终可能具有不同数量的分区，复制因子和topic的设置。在源集群上增加一到三州的topic保留率并忘记DR集群。当你的故障转移到第二个集群并发现几周数据丢失的时候，可能会导致相当令人讨厌的意外情况。试图手动保持所有这些设置同步很容易出错，并可能导致下游应用程序甚至复制本身失败。

Cluster management challenges

我们已经看到MirrorMaker部署为多个实例集群，这意味着需要另外一个集群来确定如何部署，监视和管理，由于有两个配置文件和大量的参数，MirrorMaker本身的配置管理可能是一个挑战。如果有两个以上的集群和单向复制，这种情况会增加，如果有三个双活的集群，则需要部署6个MirrorMaker集群监控和配置。每一个都可能至少有三个实例，5个双活集群，MirrorMaker的数量将增加到20个。 为了使繁忙的企业IT不满的管理开销最小化，Confluent决定实现Replicator做为源的连接器，kafka Connect框架它是从另外要给Kafka集群读取数据而不俗hi从数据库。如果你回忆起kafka connect架构，你记得每个连接器在可配置数量的任务之间公。在Replicator中，每个任务使一堆消费者和一堆生产者。Connect Framework根据需要将这些任务分配给不同的Connect工作节点。所以你可以在一台服务器上有多个任务或者把任务分配到多个服务器上。这取代了手工计算每个实例应该允许多少个MirrorMaker流以及没太及其应该运行多少个实例的工作。Connect还有一个Rest API来管理连接器和任务的配置。如果我们假设大多数kafka部署包括kafka Connect时时出于其他原因，（将数据库的更改实时发送到kafka集群），然后减少我们要管理的集群数量。另外一个显著的改进是，Replicator连接器除了从kafka的topic列表复制数据外，还从Zookeeper复制这些topic配置。

Summary 总结

在本章开始，我们描述了细羽管理多个kafka集群的原因，然后描述了几张常见的多集群架构，从简单到复杂。我们详细介绍了kafka实现故障转移的架构细节。并比较了当前可用的不同方案。然后我们继续讨论了可用的工具。从Apache Kafka的MirrorMaker，我们详细介绍了在生产环节中使用它的细节。最后我们回顾了两个备选方案，他们可以解决你可能遇到的一些问题。 无论你是最终使用那种架构和工具，请记住多集群配置和镜像管道都应该被监视和测试。就像你在生产环节中使用的其他所有东西一样。由于kafka中的多集群管理比关系数据库更容易，一些组织将其视为事后考虑。而忽视了适当的设计、计划】测试、部署自动化、监控和维护。 通过认真对多集群管理，最好将其设计的多应用程序和数据存储的整个组织的整体容灾和地理容灾计划的一部分。你将极大地增加和成功管理多个kafka集群的机会。