Amazon Managed Streaming for Apache Kafka（Amazon MSK） 提供完全托管且高度可用的 Apache Kafka 服务，可简化您处理流数据的方式。使用 Apache Kafka 时，常见的架构模式是将数据从一个集群复制到另一个集群。

跨集群复制通常用于实施业务连续性和灾难恢复计划以及提高各个 AWS 区域 的应用程序弹性。在构建多区域应用程序时，另一个使用场景是将多个地区的流数据副本存储在离终端用户更近的地方，以降低访问延迟。您可能还需要将来自多个集群的数据聚合到一个集中式集群中进行分析。

为了满足这些需求，您必须编写自定义代码或 安装和管理诸如 MirrorMaker 2.0 之类的开源工具 ，这些工具从 2.4 版本开始作为 Apache Kafka 的一部分提供。但是，设置这些工具以实现可靠的复制可能既复杂又耗时，并且需要持续的监控和扩展。

今天，我们推出了 MSK Replicator ，这是 Amazon MSK 的一项新功能，使您可以更轻松地在 MSK 集群之间可靠地设置 跨区域 和 同区域 复制，并自动扩展以处理您的工作负载。您可以将 MSK Replicator 用于预配置和 无服务器 MSK 集群类型，包括使用 分层存储 的集群类型。

使用 MSK Replicator，您可以设置主动-被动和主动-主动集群拓扑，以提高 Kafka 应用程序跨区域的弹性：

在 主动-主动 设置中，两个 MSK 集群都在积极提供读取和写入服务。

在 主动-被动 设置中，一次只有一个 MSK 集群在主动提供流数据，另一个集群则处于待机状态。

接下来查看该服务的实际操作。

跨 AWS 区域创建 MSK Replicator
我在不同的区域部署了两个 MSK 集群。MSK Replicator 要求集群启用 IAM 身份验证。我可以继续为其他客户端使用其他身份验证方法，例如 mTLS 或 SASL。源集群还需要启用多 VPC 私有连接。

从网络的角度来看，集群的安全组允许在集群与 Replicator 使用的安全组之间传输流量。例如，我可以添加自引用的入站和出站规则，允许来自和流向同一个安全组的流量。为简单起见，我对两个集群都使用默认 VPC 及其 默认安全组 。

在创建复制器之前，我更新了源集群的 集群策略 ，使 MSK 服务（包括复制器）能够找到并访问集群。在 Amazon MSK 控制台 中，我选择源区域。我从导航窗格中选择 集群 ，然后选择源集群。首先，我在顶部复制源集群 ARN 。然后，在 属性 选项卡中，我选择 安全设置 中的 编辑集群策略 。在那里，我使用以下 JSON 策略（替换源集群 ARN）并保存更改：

"Version": "2012-10-17", "Statement": [ "Effect": "Allow", "Principal": { "Service": "kafka.amazonaws.com" "Action": [ "kafka:CreateVpcConnection", "kafka:GetBootstrapBrokers", "kafka:DescribeClusterV2" "Resource": "<SOURCE_CLUSTER_ARN>"

我在控制台中选择目标区域。我从导航窗格中选择 复制器 ，然后选择 创建复制器 。我输入复制器的名称和描述。

在 源集群 部分，我选择源 MSK 集群所在的区域。然后，我选择 浏览 ，从列表中选择源 MSK 集群 。请注意，只能为设置了集群策略的集群创建复制器。

我将 子网 和 安全组 保留为默认值，以便使用我的默认 VPC 及其默认安全组。此网络配置可用于放置弹性网络接口（EIN），以促进与集群的通信。

源集群的 访问控制方法 设置为 基于 IAM 角色的身份验证 。或者，我可以同时开启多种身份验证方法，以便在复制器使用 IAM 时继续使用需要其他身份验证方法（例如 mTLS 或 SASL）的客户端。对于跨区域复制，源集群无法启用未经身份验证的访问，因为我们使用多 VPC 来访问其源集群。

在 目标集群 部分， 集群区域 设置为我使用控制台的区域。我选择 浏览 ，从列表中选择目标 MSK 集群 。

与我对源集群执行的操作类似，我将 子网组 和 安全组 保留为默认值。此网络配置用于放置与目标集群通信所需的 ENI。目标集群的 访问控制方法 也设置为 基于 IAM 角色的身份验证 。

在 复制器设置 部分，我使用默认的 主题复制 配置，这样就可以复制所有主题。或者，我可以指定一个以逗号分隔的正则表达式列表，这些正则表达式表示要复制或要从复制中排除的主题的名称。在 其他设置 中，我可以选择复制主题配置、 访问控制列表（ACL） 以及检测和复制新主题。

通过 使用者组复制 ，我可以指定是否应复制使用者组偏移，以便在切换后，使用应用程序可以在上次在主集群中停下来的位置附近继续处理。我可以指定一个以逗号分隔的正则表达式列表，这些正则表达式表示要复制或从复制中排除的使用者组的名称。我还可以选择检测和复制新的使用者组。我使用复制所有使用者组的默认设置。

在 压缩 中，我从正在复制的数据的可用压缩类型列表中选择 无 。

Amazon MSK 控制台可以自动创建一个服务执行角色，该角色具有复制器运行所需的必要权限。MSK 服务使用该角色连接到源集群和目标集群、从源集群读取数据以及向目标集群写入数据。但是，我也可以选择创建和提供自己的角色。在 访问权限 中，我选择 创建或更新 IAM 角色 。

最后，我要为复制器添加标签。我可以使用标签来搜索和筛选我的资源或跟踪我的费用。在 复制器标签 部分，我输入 Environment 作为密钥，输入 AWS News Blog 作为值。然后，我选择 创建 。

几分钟后，复制器运行。现在就来使用吧！

跨 AWS 区域测试 MSK 复制器
为了连接到源集群和目标集群，我已经在两个区域设置了两个 Amazon Elastic Compute Cloud（Amazon EC2） 实例。我按照 MSK 文档中的说明 安装了 Apache Kafka 客户端工具。由于我使用的是 IAM 身份验证，所以这两个实例附加了一个 IAM 角色，使它们能够连接、发送和接收来自集群的数据。为了简化联网，我对 EC2 实例和 MSK 集群使用了 默认安全组 。

首先，我在源集群中创建了一个新主题并发送了几条消息。我使用 Amazon EC2 实例连接 登录源区域的 EC2 实例。我将目录更改为安装 Kafka 客户端可执行文件的路径（路径取决于您使用的版本）：

cd /home/ec2-user/kafka_2.12-2.8.1/bin

要连接到源集群，我需要知道它的引导服务器。我使用源区域的 MSK 控制台，从导航页面中选择 集群 ，然后从列表中选择源集群。在 集群摘要 部分，我选择 查看客户端信息 。在那里，我复制了 引导服务器 的列表。由于 EC2 实例与集群位于同一 VPC 中，因此我复制了 私有端点（单 VPC） 列中的列表。

回到 EC2 实例，我将引导服务器列表放在 SOURCE_BOOTSTRAP_SERVERS 环境变量中。

export SOURCE_BOOTSTRAP_SERVERS=b-2.uscluster.esijym.c9.kafka.us-east-1.amazonaws.com:9098,b-3.uscluster.esijym.c9.kafka.us-east-1.amazonaws.com:9098,b-1.uscluster.esijym.c9.kafka.us-east-1.amazonaws.com:9098

现在，我在源集群上创建一个主题。

./kafka-topics.sh --bootstrap-server $SOURCE_BOOTSTRAP_SERVERS --command-config client.properties --create --topic my-topic --partitions 6

我使用新主题向源集群发送了几条消息。

./kafka-console-producer.sh --broker-list $SOURCE_BOOTSTRAP_SERVERS --producer.config client.properties --topic my-topic
>来自美国的问候
>这是我的消息

让我们看看目标集群会发生什么情况。我连接到目标区域的 EC2 实例。与我对另一个实例执行的操作类似，我获得了目标集群的引导服务器列表并将其放置在 TARGET_BOOTSTRAP_SERVERS 环境变量中。

在目标集群上，源集群别名将作为前缀添加到复制的主题名称中。为了查找源集群别名，我在 MSK 控制台导航窗格中选择 复制器 。在那里，我选择了我刚刚创建的复制器。在 属性 选项卡中，我在 源集群 部分查找 集群别名 。

我通过查看目标集群中的主题列表（它是输出列表中的最后一项）来确认复制的主题的名称：

./kafka-topics.sh --list --bootstrap-server $TARGET_BOOTSTRAP_SERVERS --command-config client.properties
. . .
us-cluster-c78ec6d63588.my-topic

现在，我已经知道了复制的主题在目标集群上的名称，我启动一个使用者来接收最初发送到源集群的消息：

./kafka-console-consumer.sh --bootstrap-server $TARGET_BOOTSTRAP_SERVERS --consumer.config client.properties --topic us-cluster-c78ec6d63588.my-topic --from-beginning
来自美国的问候
这是我的消息

请注意，我可以在主题订阅中使用通配符（例如， .*my-topic ）来自动处理前缀，并在源集群和目标集群中使用相同的配置。

正如预期的那样，我发送到源集群的所有消息都被复制并被连接到目标集群的使用者接收。

我可以使用 监控 选项卡监控 MSK Replicator 延迟、吞吐量、错误和延迟指标。因为这是通过 Amazon CloudWatch 实现的，因此我可以轻松创建自己的 警报 ，并将这些指标包含在我的 控制面板 中。

为了将配置更新为主动-主动设置，我按照类似的步骤在另一个区域创建了一个复制器，并以另一个方向在集群之间复制流数据。有关如何管理失效转移和失效自动恢复的详细信息，请参阅 MSK Replicator 文档 。

可用性和定价
MSK Replicator 现已在美国东部（俄亥俄州）、美国东部（弗吉尼亚州北部）、美国西部（俄勒冈州）、亚太地区（新加坡）、亚太地区（悉尼）、欧洲地区（法兰克福）和欧洲地区（爱尔兰）推出。

使用 MSK Replicator 时，您需要为复制的每 GB 数据付费，并为每个 Replicator 按小时付费。您还需要为源和目标 MSK 集群支付 Amazon MSK 的常规费用，以及跨区域数据传输的标准 AWS 费用。有关更多信息，请参阅 MSK 定价 。

通过使用 MSK Replicator，您可以快速实施跨区域和同区域复制，从而提高架构的弹性，并将数据存储在靠近合作伙伴和最终用户的位置。您还可以使用这项新功能将流数据复制到单个集中式集群，以便更轻松地进行分析，从而获得更深入的见解。

使用 Amazon MSK Replicator 简化您的数据流架构。

— Danilo