Amazon 管理 Apache Kafka 建构库

AWS CDK v1在2023-06-01上达到了支持结束. 此软件包不再更新,用户应该迁移到 AWS CDK v2.

关于如何移徙的更多信息,请参见: 移动到 AWS CDK v2 指南。 。 。 .

亚马逊 MSK 是一种完全管理的服务,它使您可以轻松地构建和运行使用Apache Kafka处理流数据的程序.

以下例子创建了MSK Cluster.

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V2_8_1,
  vpc,
});

允许连接

为了控制谁可以访问集群,请使用 .connections 属性。 关于MSK使用的端口列表,请参见: MSK 文档。 。 。 .

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V2_8_1,
  vpc,
});

cluster.connections.allowFrom(
  ec2.Peer.ipv4('1.2.3.4/8'),
  ec2.Port.tcp(2181),
);
cluster.connections.allowFrom(
  ec2.Peer.ipv4('1.2.3.4/8'),
  ec2.Port.tcp(9094),
);

集群终点

您可以使用以下属性来获取 Kafka 经纪人或 Zookeeper 节点的列表

declare const cluster: msk.Cluster;
new CfnOutput(this, 'BootstrapBrokers', { value: cluster.bootstrapBrokers });
new CfnOutput(this, 'BootstrapBrokersTls', { value: cluster.bootstrapBrokersTls });
new CfnOutput(this, 'BootstrapBrokersSaslScram', { value: cluster.bootstrapBrokersSaslScram });
new CfnOutput(this, 'ZookeeperConnection', { value: cluster.zookeeperConnectionString });
new CfnOutput(this, 'ZookeeperConnectionTls', { value: cluster.zookeeperConnectionStringTls });

导入已存在的集群

要将已有的 MSK 集群导入到您的 CDK 应用程序中, 请使用 .fromClusterArn() 方法.

const cluster = msk.Cluster.fromClusterArn(this, 'Cluster', 
  'arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1',
);

客户端认证

MSK 支持 以下认证机制.

只能启用一种认证方法 .

电话

用 TLS 设置客户端认证 certificateAuthorityArns 以引用您的 ACM 私人 CA. 更多关于私人CA的信息.

import * as acmpca from '@aws-cdk/aws-acmpca';

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V2_8_1,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.tls({
    certificateAuthorities: [
      acmpca.CertificateAuthority.fromCertificateAuthorityArn(
        this,
        'CertificateAuthority',
        'arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111',
      ),
    ],
  }),
});

SASL/SRAM 软件

启用客户端认证 SASL/SRAM 软件数字 :

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V2_8_1,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.sasl({
    scram: true,
  }),
});

SASL/IAM 卫星

启用客户端认证 行政会议数字 :

declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'cluster', {
  clusterName: 'myCluster',
  kafkaVersion: msk.KafkaVersion.V2_8_1,
  vpc,
  encryptionInTransit: {
    clientBroker: msk.ClientBrokerEncryption.TLS,
  },
  clientAuthentication: msk.ClientAuthentication.sasl({
    iam: true,
  }),
});