Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅中国的 Amazon Web Services 服务入门。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

教程：将自定义信任库与 Amazon MSK 结合使用

以下教程演示如何安全地连接（传输中的加密）到使用由自定义、私有甚至自托管的证书颁发机构 (CA) 颁发的服务器证书的 Kafka 集群。

为了通过 TLS 安全地将任何 Kafka 客户端连接到 Kafka 集群，Kafka 客户端（比如 Flink 应用程序示例）必须信任由 Kafka 集群的服务器证书（从颁发证书到根级 CA）提供的完整信任链。作为自定义信任库的示例，我们将使用启用了双向 TLS (MTLS) 身份验证的 Amazon MSK 集群。这意味着 MSK 群集节点使用由AmazonCertificate Manager 私有证书颁发机构 (ACM 私有 CA) 对您的账户和区域是私有的，因此不被执行 Flink 应用程序的 Java 虚拟机 (JVM) 的默认信任库信任。

您还可以使用本教程中的技术在 Kinesis Data Analytics 应用程序与其他 Apache Kafka 源之间进行交互，例如：

您的应用程序将使用自定义消费者 (CustomFlinkKafkaConsumer) 它会覆盖open方法来加载自定义信任库。这使得信任库在应用程序重新启动或替换线程后可供应用程序使用。

使用以下代码检索和存储自定义信任库，从CustomFlinkKafkaConsumer.javafile:

@Override
public void open(Configuration configuration) throws Exception {
    // write truststore to /tmp
    // NOTE: make sure that truststore is in JKS format for KDA/Flink. See README for details
    dropFile("/tmp");

    super.open(configuration);
}

private void dropFile(String destFolder) throws Exception
{
    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
    InputStream input = classLoader.getResourceAsStream("kafka.client.truststore.jks");
    byte[] buffer = new byte[input.available()];
    input.read(buffer);
        File destDir = new File(destFolder);
        File targetFile = new File(destDir, "kafka.client.truststore.jks");
        OutputStream outStream = new FileOutputStream(targetFile);
        outStream.write(buffer);
        outStream.flush();
}

使用亚马逊 MSK 集群创建 VPC

要创建示例 VPC 和 Amazon MSK 集群以从 Kinesis Data Analytics 应用程序中访问，请按照使用Amazon MSK 入门教程。

在完成本教程时，还执行以下操作：

创建自定义信任库并将其应用到您的集群

在本节中，您将创建自定义证书颁发机构 (CA)，使用它生成自定义信任库，然后将其应用于您的 MSK 集群。

要创建和应用您的自定义信任库，请按照客户端身份验证中的教程AmazManaged Streaming for Apache Kafka.

创建应用程序代码

在本节中，您下载并编译应用程序 JAR 文件。

该示例的 Java 应用程序代码可从 GitHub. 要下载应用程序代码，请执行以下操作：

上传 Apache Flink 流式处理 Java 代码

在本节中，您将应用程序代码上传到在一节中创建的 Amazon S3 存储桶入门 (DataStreamAPI)教程。

您的应用程序代码现在存储在存储Amazon S3 中，应用程序可以在其中访问代码。

创建 应用程序

配置应用程序

运行应用程序

通过运行应用程序，打开 Apache Flink 控制面板，然后选择所需的 Flink 作业，可以查看 Flink 作业图。

测试应用程序

在本节中，您将记录写入到源主题。应用程序从源主题中读取记录，并将其写入到目标主题中。您可以将记录写入到源主题以及从目标主题中读取记录，以验证应用程序是否正常工作。

要写入和读取主题中的记录，请按照中的步骤操作步骤 6：生成和使用数据中的使用Amazon MSK 入门教程。

要从目标主题中读取，请在到集群的第二个连接中使用目标主题名称，而不是源主题：

bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerString --consumer.config client.properties --topic AmazonKafkaTutorialTopicDestination --from-beginning

如果在目标主题中没有任何记录，请参阅故障排除主题中的无法访问 VPC 中的资源一节。