使用 Redpanda、Quarkus 和 Testcontainers 进行本地开发 _Redpanda

文章插图

在本文中，您将学习如何使用Redpanda和 Quarkus 加速本地开发。主要目标是展示您可以用 Redpanda 替换 Apache Kafka ?而无需对源代码进行任何更改。相反，您将获得一种在没有 Zookeeper 和 JVM 的情况下运行现有 Kafka 应用程序的快速方法。您还将看到 Quarkus 如何使用 Redpanda 作为本地实例进行开发。最后，我们将在 Testcontainers Cloud 中运行所有容器。
提醒您：我们正在构建一个简化版的股票市场平台。应用程序接收并处理传入的 stock-service 订单。有两种类型的订单：购买 ( BUY) 和销售 ( SELL) 。在stock-service消费 Kafka 流的同时，order-service生成事件并将其发送到orders.buy和orders.sell主题。这是我们的架构图。如您所见，stock-service它还使用 PostgreSQL 作为数据库。

文章插图

源代码如果您想自己尝试这个练习，您可以随时查看我的源代码。为此，您需要克隆我的 GitHub 存储库。然后切换到dev分支。之后，您应该按照我的指示进行操作。让我们开始。
安装小熊猫此步骤不是必需的。但是，值得安装 Redpanda，因为它提供了一个名为 Redpanda Keeper ( rpk) 的有用 CLI 来管理集群。要在 macOS 上安装 Redpanda，只需运行以下命令：
$ brew install redpanda-data/tap/redpanda现在，我们可以轻松地创建和运行一个新集群。出于开发目的，我们只需要一个单节点 Redpanda 集群。为了运行，您需要在笔记本电脑上安装 Docker 。
$ rpk container start在继续下一步之前，让我们删除当前集群。Quarkus 会自动为我们创造一切。
$ rpk container purgeQuarkus 与 Kafka 和 Postgres让我们从stock-service. 正如我之前提到的，它使用来自 Kafka 主题的流并连接到 PostgreSQL 数据库。因此，第一步是包含以下依赖项：

<dependency><groupId>io.quarkus</groupId><artifactId>quarkus-kafka-streams</artifactId></dependency><dependency><groupId>io.quarkus</groupId><artifactId>quarkus-hibernate-orm-panache</artifactId></dependency><dependency><groupId>io.quarkus</groupId><artifactId>quarkus-jdbc-postgresql</artifactId></dependency>

现在，我们可以继续执行。所有流的拓扑结构在以下方法中提供：
@Producespublic Topology buildTopology() {...}那里定义了一些不同的流。但是让我们看一下负责从传入订单创建交易的拓扑片段

final String ORDERS_BUY_TOPIC = "orders.buy";final String ORDERS_SELL_TOPIC = "orders.sell";final String TRANSACTIONS_TOPIC = "transactions";// ... other streamsKStream<Long, Order> orders = builder.stream(ORDERS_SELL_TOPIC,Consumed.with(Serdes.Long(), orderSerde));builder.stream(ORDERS_BUY_TOPIC, Consumed.with(Serdes.Long(), orderSerde)).merge(orders).peek((k, v) -> {log.infof("New: %s", v);logic.add(v);});builder.stream(ORDERS_BUY_TOPIC, Consumed.with(Serdes.Long(), orderSerde)).selectKey((k, v) -> v.getProductId()).join(orders.selectKey((k, v) -> v.getProductId()),this::execute,Joinwindows.of(Duration.ofSeconds(10)),StreamJoined.with(Serdes.Integer(), orderSerde, orderSerde)).filterNot((k, v) -> v == null).map((k, v) -> new KeyValue<>(v.getId(), v)).peek((k, v) -> log.infof("Done -> %s", v)).to(TRANSACTIONS_TOPIC, Produced.with(Serdes.Long(), transactionSerde));

整个实现更先进。具体可以参考我在介绍中提到的那篇文章。现在，让我们假设我们仍在开发我们的股票市场应用程序。首先，我们应该运行 PostgreSQL 和本地 Kafka 集群。我们使用 Redpanda，它很容易在本地运行。之后，我们通常会在Application.properties. 但是使用名为 Quarkus Dev Services 的功能，我们现在唯一需要配置的是用于消费 Kafka Streams 的主题名称和应用程序 ID 。这两个都是 Kafka Streams 所必需的。
现在，最重要的是：您只需要启动 Quarkus 应用程序。而已。不要自己运行任何外部工具，也不要在配置设置中为它们提供任何地址。只需添加您在下面看到的两行：
quarkus.kafka-streams.application-id = stockquarkus.kafka-streams.topics = orders.buy,orders.sell使用 Redpanda 在开发模式下运行 Quarkus在运行 Quarkus 应用程序之前，请确保您的笔记本电脑上运行了 Docker 。当您这样做时，您唯一需要做的就是启动两个测试应用程序。让我们从开始，stock-service因为它接收由order-service. 转到stock-service目录并运行以下命令：