Spring Boot与运行在Kubernetes上的ksqlDB集成教程

绝地灬酷狼 2023-10-01 22:17 12阅读 0赞

在本文中，您将学习如何在 Kubernetes 上运行 ksqlDB 并将其与 Spring Boot 一起使用。您还将了解如何基于 Strimzi 运算符在 Kubernetes 上运行 Kafka。

为了将 Spring Boot 与 ksqlDB 服务器集成，我们将利用 ksqlDB 提供的轻量级 Java 客户端。此客户端支持拉取和推送查询。它还提供了用于插入行和创建表或流的 API。 您可以在此处 的 ksqlDB 文档中阅读有关它的更多信息。

![348ebdd1b105f98791084bfbf3f4d15d.png][]

我们的示例 Spring Boot 应用程序非常简单。我们将使用 Spring Cloud Stream Supplierbean 生成事件并将其发送到 Kafka 主题。有关使用 Spring Cloud Stream 的 Kafka 的更多信息，请参阅以下 文章 。

另一方面，我们的应用程序使用 kSQL 查询从 Kafka 主题获取数据。它还KTable在启动时创建。

**源代码**

如果您想自己尝试一下，可以随时查看我的源代码。为此，您需要克隆我的 GitHub 存储库 。然后进入transactions-service目录。之后，您应该按照我的指示进行操作。让我们开始。

**先决条件**

我们将使用几种工具。你需要有：

*  Kubernetes 集群——它可能是一个单节点的本地集群，例如 Minikube 或 Kind。就个人而言，我在 Docker 桌面上使用 Kubernetes
 *  kubectlCLI – 与集群交互
 *  Helm——我们将使用它在 Kubernetes 上安装 ksqlDB 服务器。如果您没有 Helm，则必须安装它

**使用 Strimzi 在 Kubernetes 上运行 Kafka**

当然，我们需要一个 Kafka 实例来执行我们的练习。有几种方法可以在 Kubernetes 上运行 Kafka。我将向您展示如何使用基于运算符的方法来实现。第一步，您需要在集群上安装 OLM（Operator Lifecycle Manager）。为此，您只需在 Kubernetes 上下文中执行以下命令：

$ curl -L https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.21.2/install.sh -o install.sh
 $ chmod +x install.sh
 $ ./install.sh v0.21.2

然后，您可以继续安装 Strimzi 操作员。这只是一个命令。

$ kubectl create -f https://operatorhub.io/install/stable/strimzi-kafka-operator.yaml

现在，我们可以在 Kubernetes 上创建一个 Kafka 集群。让我们从练习的专用命名空间开始：

$ kubectl create ns kafka

我假设你有一个单节点 Kubernetes 集群，所以我们还创建了一个单节点 Kafka。Kafka这是带有CRD的 YAML 清单。您可以在路径下的存储库中找到它k8s/cluster.yaml。

apiVersion: kafka.strimzi.io/v1beta2
 kind: Kafka
 metadata:
 name: my-cluster
 spec:
 entityOperator:
 topicOperator: {}
 userOperator: {}
 kafka:
 config:
 default.replication.factor: 1
 inter.broker.protocol.version: "3.2"
 min.insync.replicas: 1
 offsets.topic.replication.factor: 1
 transaction.state.log.min.isr: 1
 transaction.state.log.replication.factor: 1
 listeners:
 - name: plain
 port: 9092
 tls: false
 type: internal
 - name: tls
 port: 9093
 tls: true
 type: internal
 replicas: 1
 storage:
 type: jbod
 volumes:
 - id: 0
 type: persistent-claim
 size: 30Gi
 deleteClaim: true
 version: 3.2.0
 zookeeper:
 replicas: 1
 storage:
 type: persistent-claim
 size: 10Gi
 deleteClaim: true

让我们将它应用到命名空间中的 Kubernetes kafka：

$ kubectl apply -f k8s/cluster.yaml -n kafka

您应该会看到一个 Kafka 实例和一个 Zookeeper 实例。如果 pod 正在运行，则意味着您在 Kubernetes 上安装了 Kafka。

$ kubectl get pod -n kafka
 NAME READY STATUS RESTARTS AGE
 my-cluster-entity-operator-68cc6bc4d9-qs88p 3/3 Running 0 46m
 my-cluster-kafka-0 1/1 Running 0 48m
 my-cluster-zookeeper-0 1/1 Running 0 48m

Kafka 在集群内以  
namemy-cluster-kafka-bootstrap和 port可用9092。

kubectl get svc -n kafka
 NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
 my-cluster-kafka-bootstrap ClusterIP 10.108.109.255 <none> 9091/TCP,9092/TCP,9093/TCP 47m
 my-cluster-kafka-brokers ClusterIP None <none> 9090/TCP,9091/TCP,9092/TCP,9093/TCP 47m
 my-cluster-zookeeper-client ClusterIP 10.102.10.251 <none> 2181/TCP 47m
 my-cluster-zookeeper-nodes ClusterIP None <none> 2181/TCP,2888/TCP,3888/TCP 47m

**在 Kubernetes 上运行 KsqlDB 服务器**

KsqlDB 服务器是 Confluent 平台的一部分。由于我们不是在 Kubernetes 上安装整个 Confluent Platform，而只是一个开源的 Kafka 集群，我们需要单独安装 KsqlDB Server。让我们用 Helm 来做。KSQL 服务器没有“官方”Helm 图表。因此，我们应该直接去 GitHub 上的 Confluent Helm 仓库：

$ git clone https://github.com/confluentinc/cp-helm-charts.git
 $ cd cp-helm-charts

在这个存储库中，您可以为每个单独的 Confluent 组件找到单独的 Helm 图表，包括控制中心或 KSQL Server。我们的图表在存储库中的位置是charts/cp-ksql-server. 我们需要在安装过程中覆盖一些默认设置。首先，我们必须禁用无头模式。在无头模式下，KSQL Server 不公开 HTTP 端点并从输入脚本加载查询。我们的 Spring Boot 应用程序将通过 HTTP 连接到服务器。在下一步中，我们应该覆盖 Kafka 集群的默认地址和仍然6.1.0存在的 KSQL Server 的默认版本。我们将使用最新版本7.1.1。这是helm您应该在 Kubernetes 集群上运行的命令：

$ helm install cp-ksql-server \
        --set ksql.headless=false \
        --set kafka.bootstrapServers=my-cluster-kafka-bootstrap:9092 \
        --set imageTag=7.1.1 \
      charts/cp-ksql-server -n kafka

让我们验证 KSQL 是否在集群上运行：

$ kubectl get pod -n kafka | grep ksql
    cp-ksql-server-679fc98889-hldfv               2/2     Running   0               2m11s

HTTP 端点可用于 namecp-ksql-server和 port下的其他应用程序8088：

$ kubectl get svc -n kafka | grep ksql
 cp-ksql-server ClusterIP 10.109.189.36 <none> 8088/TCP,5556/TCP 3m25s

现在，我们的 Kubernetes 集群上运行着所需的全部人员。因此，我们可以继续进行 Spring Boot 应用程序的实现。

**将 Spring Boot 与 ksqlDB 集成**

我没有发现 Spring Boot 和 ksqlDB 之间有任何开箱即用的集成。因此，我们将ksqldb-api-client直接使用。首先，我们需要包含 ksqlDB Maven 存储库和一些依赖项：

<dependencies>
 ...
 
 <dependency>
 <groupId>io.confluent.ksql</groupId>
 <artifactId>ksqldb-api-client</artifactId>
 <version>0.26.0</version>
 </dependency>
 <dependency>
 <groupId>io.confluent.ksql</groupId>
 <artifactId>ksqldb-udf</artifactId>
 <version>0.26.0</version>
 </dependency>
 <dependency>
 <groupId>io.confluent.ksql</groupId>
 <artifactId>ksqldb-common</artifactId>
 <version>0.26.0</version>
 </dependency>
 </dependencies>
 
 <repositories>
 <repository>
 <id>ksqlDB</id>
 <name>ksqlDB</name>
 <url>https://ksqldb-maven.s3.amazonaws.com/maven/</url>
 </repository>
 </repositories>

之后，我们可以定义一个@Bean返回 ksqlDBClient实现的 Spring。由于我们将在与 KSQL Server 相同的命名空间中运行我们的应用程序，因此我们需要提供 Kubernetes 服务名称作为主机名。

@Configuration
 public class KSQLClientProducer {
 
 @Bean
 Client ksqlClient() {
 ClientOptions options = ClientOptions.create()
 .setHost("cp-ksql-server")
 .setPort(8088);
 return Client.create(options);
 }
 }

我们的应用程序通过 HTTP 端点与 KSQL Server 交互。KTable它在启动时创建一个单曲。为此，我们需要调用executeStatementKSQL Clientbean 实例上的方法。我们正在创建 SOURCE 表以启用对其运行 拉取查询 。该表从transactions主题中获取数据。它期望传入事件中的 JSON 格式。

public class KTableCreateListener implements ApplicationListener<ContextRefreshedEvent> {
 
 private static final Logger LOG = LoggerFactory.getLogger(KTableCreateListener.class);
 private Client ksqlClient;
 
 public KTableCreateListener(Client ksqlClient) {
 this.ksqlClient = ksqlClient;
 }
 
 @Override
 public void onApplicationEvent(ContextRefreshedEvent event) {
 try {
 String sql = """
 CREATE SOURCE TABLE IF NOT EXISTS transactions_view (
 id BIGINT PRIMARY KEY,
 sourceAccountId BIGINT,
 targetAccountId BIGINT,
 amount INT
 ) WITH (
 kafka_topic='transactions',
 value_format='JSON'
 );
 """;
 ExecuteStatementResult result = ksqlClient.executeStatement(sql).get();
 LOG.info("Result: {}", result.queryId().orElse(null));
 } catch (ExecutionException | InterruptedException e) {
 LOG.error("Error: ", e);
 }
 }
 }

创建表后，我们可以对其运行一些查询。有非常简单的查询。我们正在尝试查找与特定帐户相关的所有交易和所有交易。

@RestController
 @RequestMapping("/transactions")
 public class TransactionResource {
 
 private static final Logger LOG = LoggerFactory.getLogger(TransactionResource.class);
 Client ksqlClient;
 
 public TransactionResource(Client ksqlClient) {
 this.ksqlClient = ksqlClient;
 }
 
 @GetMapping
 public List<Transaction> getTransactions() throws ExecutionException, InterruptedException {
 StreamedQueryResult sqr = ksqlClient
 .streamQuery("SELECT * FROM transactions_view;")
 .get();
 Row row;
 List<Transaction> l = new ArrayList<>();
 while ((row = sqr.poll()) != null) {
 l.add(mapRowToTransaction(row));
 }
 return l;
 }
 
 @GetMapping("/target/{accountId}")
 public List<Transaction> getTransactionsByTargetAccountId(@PathVariable("accountId") Long accountId)
 throws ExecutionException, InterruptedException {
 StreamedQueryResult sqr = ksqlClient
 .streamQuery("SELECT * FROM transactions_view WHERE sourceAccountId=" + accountId + ";")
 .get();
 Row row;
 List<Transaction> l = new ArrayList<>();
 while ((row = sqr.poll()) != null) {
 l.add(mapRowToTransaction(row));
 }
 return l;
 }
 
 private Transaction mapRowToTransaction(Row row) {
 Transaction t = new Transaction();
 t.setId(row.getLong("ID"));
 t.setSourceAccountId(row.getLong("SOURCEACCOUNTID"));
 t.setTargetAccountId(row.getLong("TARGETACCOUNTID"));
 t.setAmount(row.getInteger("AMOUNT"));
 return t;
 }
 
 }

**使用 Spring Cloud Stream 向主题发送事件**

最后，我们可以进行练习的最后一部分。我们需要生成测试数据并将其发送到 Kafkatransactions主题。实现它的最简单方法是使用 Spring Cloud Stream Kafka 模块。首先，让我们添加以下 Maven 依赖项：

<dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-starter-stream-kafka</artifactId>
 </dependency>

然后，我们可以创建一个基于 Spring Supplierbean 的生产者。Supplierbean 不断生成新事件并将其发送到目标通道。默认情况下，它每秒重复一次该操作。

@Configuration
 public class KafkaEventProducer {
 
 private static long transactionId = 0;
 private static final Random r = new Random();
 
 @Bean
 public Supplier<Message<Transaction>> transactionsSupplier() {
 return () -> {
 Transaction t = new Transaction();
 t.setId(++transactionId);
 t.setSourceAccountId(r.nextLong(1, 100));
 t.setTargetAccountId(r.nextLong(1, 100));
 t.setAmount(r.nextInt(1, 10000));
 Message<Transaction> o = MessageBuilder
 .withPayload(t)
 .setHeader(KafkaHeaders.MESSAGE_KEY, new TransactionKey(t.getId()))
 .build();
 return o;
 };
 }
 }

当然，我们还需要提供我们的 Kafka 集群的地址和频道的目标主题名称。Kafka的地址是在部署阶段注入的。

spring.kafka.bootstrap-servers = ${KAFKA_URL}
    spring.cloud.stream.bindings.transactionsSupplier-out-0.destination = transactions

最后，让我们在 Kubernetes 上部署我们的 Spring Boot。这是包含 KubernetesDeployment和Service定义的 YAML 清单：

apiVersion: apps/v1
    kind: Deployment
    metadata:
      name: transactions
    spec:
      selector:
        matchLabels:
          app: transactions
      template:
        metadata:
          labels:
            app: transactions
        spec:
          containers:
          - name: transactions
            image: piomin/transactions-service
            env:
              - name: KAFKA_URL
                value: my-cluster-kafka-bootstrap:9092
            ports:
            - containerPort: 8080
    ---
    apiVersion: v1
    kind: Service
    metadata:
      name: transactions
    spec:
      type: ClusterIP
      selector:
        app: transactions
      ports:
        - port: 8080

让我们在命名空间中部署应用程序kafka：

$ kubectl apply -f k8s/deployment.yaml -n kafka

**在 Kubernetes 上测试 ksqlDB**

在 Kubernetes 上部署应用程序后，让我们启用port-forward在本地端口上对其进行测试：

$ kubectl port-forward service/transactions 8080:8080

现在，我们可以测试我们的两个 HTTP 端点。让我们从搜索所有事务的端点开始：

$ curl http://localhost:8080/transactions

然后，您可以调用端点来搜索与 相关的所有事务targetAccountId，例如：

$ curl http://localhost:8080/transactions/target/10

**最后的想法**

在本文中，我想展示如何在 Kubernetes 上使用 ksqlDB。我们使用 Spring Boot 和 Spring Cloud Stream 等框架与 Kafka 和 ksqlDB 进行交互。您可以了解如何使用 Strimzi 运算符在 Kubernetes 上运行 Kafka 集群，或者如何直接从 Helm 存储库部署 KSQL Server。

![053709f50bc44e4c8e3a1b2e1fe039f6.jpeg][]

[348ebdd1b105f98791084bfbf3f4d15d.png]: https://img-blog.csdnimg.cn/img_convert/348ebdd1b105f98791084bfbf3f4d15d.png
[053709f50bc44e4c8e3a1b2e1fe039f6.jpeg]: https://img-blog.csdnimg.cn/053709f50bc44e4c8e3a1b2e1fe039f6.jpeg