在 OpenShift 中使用 AMQ Streams

流程

1. 在 OpenShift Web 控制台中进入到 Home > Projects 页面，再创建一个用于安装的项目（命名空间）。

   在这个示例中，我们使用名为 amq-streams-kafka 的项目。

2. 进入 Operators > OperatorHub 页面。

3. 在 Filter by keyword 框中输入关键字以查找 Red Hat Integration - AMQ Streams operator。

   operator 位于 Streaming 和 Messaging 目录中。

4. 点 Red Hat Integration - AMQ Streams 显示 Operator 信息。
5. 阅读有关 Operator 的信息，再点 Install。

6. 在 Install Operator 页面中，从以下安装和更新选项中选择：

   • 更新频道 ：选择 Operator 的更新频道。

     • （默认） stable 频道包含所有最新的更新和发行版本，包括主版本、次版本和微版本，这些版本被认为经过充分测试和稳定。
     • amq-streams-X.x 频道包含主发行版本的次要和微版本更新，其中 X 是主版本的版本号。
     • amq-streams-X.Y.x 频道包含次要发行本版本的微版本更新，其中 X 是主版本的版本号，Y 是次版本号。

   • Installation Mode ：选择您创建的项目，以便在特定命名空间中安装 Operator。

     您可以将 AMQ Streams Operator 安装到集群中的所有命名空间（默认选项）或特定命名空间。我们建议您将特定命名空间专用于 Kafka 集群和其他 AMQ Streams 组件。

   • 更新批准 ：默认情况下，AMQ Streams Operator 由 Operator Lifecycle Manager (OLM)自动升级到最新的 AMQ Streams 版本。另外，如果您希望手动批准将来的升级，请选择 Manual。如需更多信息，请参阅 OpenShift 文档中的 Operator 指南。

7. 点 Install 将 Operator 安装到所选命名空间中。

   AMQ Streams Operator 将 Cluster Operator、CRD 和基于角色的访问控制(RBAC)资源部署到所选命名空间中。

8. Operator 就绪可用后，进入 Operators > Installed Operators 来验证 Operator 是否已安装到所选命名空间中。

   状态将显示为 Succeeded。

   现在，您可以使用 AMQ Streams operator 部署 Kafka 组件，从 Kafka 集群开始。

流程

1. 在 Web 控制台中导航到 Operators > Installed Operators 页面，然后点 Red Hat Integration - AMQ Streams 来显示 Operator 详情。

   在 Provided APIs 中，您可以创建 Kafka 组件的实例。

2. 点 Kafka 下的 Create instance 创建 Kafka 实例。

   默认情况下，您将创建一个名为 my-cluster 的 Kafka 集群，它有三个 Kafka 代理节点和三个 ZooKeeper 节点。集群使用临时存储。

3. 点 Create 开始安装 Kafka。

   等待状态变为 Ready。

流程

1. 在 Web 控制台中进入到 Operators > Installed Operators 页面，然后点 Red Hat Integration - AMQ Streams 来显示 Operator 详情。
2. 选择 Kafka 页面来显示已安装的 Kafka 集群。

3. 点您要配置的 Kafka 集群的名称查看其详情。

   在这个示例中，我们使用名为 my-cluster 的 Kafka 集群。

4. 选择 Kafka 集群的 my-cluster 的 YAML 页面。

5. 添加路由侦听器配置，以创建名为 listener1 的 OpenShift 路由。

   侦听器配置必须设置为 route 类型。您可以在 Kafka 配置中的 listeners 中添加监听程序配置。

   外部路由监听程序配置

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     namespace: amq-streams-kafka
   spec:
     kafka:
       # ...
       listeners:
         # ...
         - name: listener1
           port: 9094
           type: route
           tls: true
   # ...

   客户端通过端口 443 连接默认的路由器端口，但流量会被路由到您配置的端口，本例中为 9094。

6. 保存更新的配置。

7. 选择 Kafka 集群 my-cluster 的 Resources 页面，以查找客户端需要的连接信息。

   在 Resources 页面中，您将查找路由监听程序和您连接到 Kafka 集群所需的公共集群证书的详情。

8. 点为 Kafka 集群创建的 my-cluster-kafka-listener1-bootstrap 路由名称来显示路由详情。

9. 记录主机名。

   主机名在 Kafka 客户端中使用端口 443 指定，作为连接到 Kafka 集群的 bootstrap 地址。

   您还可以通过进入到 Networking > Routes 并选择 amq-streams-kafka 项目来找到 bootstrap 地址。

   或者，您可以使用 oc 工具提取 bootstrap 详情。

   提取 bootstrap 信息

   oc get routes my-cluster-kafka-listener1-bootstrap -o=jsonpath='{.status.ingress[0].host}{"\n"}'

10. 返回到 Resources 页面，然后点 my-cluster-cluster-ca-cert 的名称以显示用于访问 Kafka 集群的 secret 详情。

    ca.crt 证书文件包含 Kafka 集群的公共证书。

    您需要证书才能访问 Kafka 代理。

11. 生成 ca.crt 公共证书文件的本地副本。

    您可以复制证书的详细信息，或使用 OpenShift oc 工具提取证书。

    提取公共证书

    oc extract secret/my-cluster-cluster-ca-cert --keys=ca.crt --to=- > ca.crt

12. 使用 keytool 为公共集群证书创建本地信任存储。

    创建本地信任存储

    keytool -keystore client.truststore.jks -alias CARoot -import -file ca.crt

    出现提示时，创建用于访问 truststore 的密码。

    truststore 在 Kafka 客户端中指定，用于身份验证对 Kafka 集群的访问。

    现在，您已准备好开始发送和接收信息。

1. 使用 oc 命令行界面部署 Kafka producer。

   本例部署一个 Kafka 生成者，它连接到 Kafka 集群 my-cluster

   名为 my-topic 的主题被创建。

   将 Kafka producer 部署到 OpenShift

   oc run kafka-producer -ti \
   --image=registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 \
   --rm=true \
   --restart=Never \
   -- bin/kafka-console-producer.sh
   --bootstrap-server my-cluster-kafka-bootstrap:9092 \
   --topic my-topic

2. 从命令提示符输入多个信息。
3. 在 OpenShift web 控制台中进入到 Home > Projects 页面，然后选择 amq-streams-kafka 项目。
4. 从 pod 列表中，点 kafka-producer 以查看 producer pod 详情。
5. 选择 Logs 页面，以检查您输入的消息是否存在。

6. 使用 oc 命令行界面部署 Kafka 用户。

   将 Kafka 使用者部署到 OpenShift

   oc run kafka-consumer -ti \
   --image=registry.redhat.io/amq7/amq-streams-kafka-31-rhel8:2.1.0 \
   --rm=true \
   --restart=Never \
   -- bin/kafka-console-consumer.sh \
   --bootstrap-server my-cluster-kafka-bootstrap:9092 \
   --topic my-topic \
   --from-beginning

   消费者使用生成给 my-topic 的消息。

7. 从命令提示符，确认您在使用者控制台中看到传入的信息。
8. 在 OpenShift web 控制台中进入到 Home > Projects 页面，然后选择 amq-streams-kafka 项目。
9. 从 pod 列表中，点 kafka-consumer 查看使用者 pod 详情。
10. 选择 Logs 页面，以检查您消耗的消息是否存在。

1. 从 AMQ Streams 软件下载页下载并提取 AMQ Streams < version > 安装和示例文件 存档。

   将文件解压缩到任何目的地。

2. 打开命令行界面，并使用主题 my-topic 和 TLS 的身份验证属性启动 Kafka 控制台制作者。

   添加使用 OpenShift 路由访问 Kafka 代理所需的属性。

   • 将主机名和端口 443 用于您正在使用的 OpenShift 路由。

   • 使用密码并引用您为代理证书创建的信任存储。

     启动本地 Kafka producer

     kafka-console-producer.sh \
     --bootstrap-server my-cluster-kafka-listener1-bootstrap-amq-streams-kafka.apps.ci-ln-50kcyvt-72292.origin-ci-int-gce.dev.rhcloud.com:443 \
     --producer-property security.protocol=SSL \
     --producer-property ssl.truststore.password=password \
     --producer-property ssl.truststore.location=client.truststore.jks \
     --topic my-topic

3. 在运行制作者的命令行界面中键入您的消息。
4. 按 enter 来发送邮件。

5. 打开新的命令行界面标签页或窗口，并启动 Kafka 控制台使用者来接收信息。

   使用与生成者相同的连接详情。

   启动本地 Kafka 消费者

   kafka-console-consumer.sh \
   --bootstrap-server my-cluster-kafka-listener1-bootstrap-amq-streams-kafka.apps.ci-ln-50kcyvt-72292.origin-ci-int-gce.dev.rhcloud.com:443 \
   --consumer-property security.protocol=SSL \
   --consumer-property ssl.truststore.password=password \
   --consumer-property ssl.truststore.location=client.truststore.jks \
   --topic my-topic --from-beginning

6. 确认您在使用者控制台中看到传入的信息。
7. 按 Crtl+C 退出 Kafka 控制台生成者和消费者。