在本文中，您将学习如何使用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 purge
Quarkus 与 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>
现在，我们可以继续执行。所有流的拓扑结构在以下方法中提供：
@Produces
public Topology buildTopology() {
...
}
那里定义了一些不同的流。但是让我们看一下负责从传入订单创建交易的拓扑片段
final String ORDERS_BUY_TOPIC = "orders.buy";
final String ORDERS_SELL_TOPIC = "orders.sell";
final String TRANSACTIONS_TOPIC = "transactions";
// ... other streams
KStream<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 = stock
quarkus.kafka-streams.topics = orders.buy,orders.sell
使用 Redpanda 在开发模式下运行 Quarkus
在运行 Quarkus 应用程序之前，请确保您的笔记本电脑上运行了 Docker。当您这样做时，您唯一需要做的就是启动两个测试应用程序。让我们从 开始，stock-service因为它接收由order-service. 转到stock-service目录并运行以下命令：
$ cd stock-service
$ mvn quarkus:dev
如果您看到以下日志，则表示一切顺利。我们的应用程序已在 13 秒内启动。在此期间，Quarkus 还在 Docker 上启动了 Kafka、PostgreSQL，并构建了 Kafka Streams。只需一个命令，无需任何额外配置，一切都在 13 秒内完成。不错，对吧？让我们看看后台发生了什么：
首先，让我们找到下面一行以“Dev Services for Kafka started”开头的日志。它完美地描述了 Quarkus 称为Dev Services的特性。我们的 Kafka 实例已作为 Docker 容器启动，可在动态生成的端口下使用。应用程序连接到它。您现在将运行的所有其他 Quarkus 应用程序将共享同一代理实例。quarkus.kafka.devservices.shared您可以通过将属性设置为来禁用该功能false。
这可能有点令人惊讶，但 Quarkus Dev Services for Kafka 使用 Redpanda 来运行代理。当然，Redpanda 是一个兼容 Kafka 的解决方案。由于它在一秒钟内启动并且不需要 Zookeeper，因此它是本地开发的绝佳选择。
为了在 Docker 上运行代理或数据库等工具，Quarkus 使用了测试容器。如果您对有关 Quarkus Dev Services for Kafka 的更多详细信息感兴趣，请阅读以下文档。docker ps现在，让我们使用该命令显示正在运行的容器列表。有一个包含 Redpanda、PostgreSQL 和 Testcontainers 的容器。
使用 Redpanda 和 Quarkus 管理 Kafka 流
让我们验证应用程序端的一切工作方式。运行应用程序后，我们可以利用另一个有用的 Quarkus 功能，称为 Dev UI。我们的 UI 控制台在地址下可用http://localhost:8080/q/dev/。访问它后，您可以通过单击Apache Kafka Streams 磁贴内的按钮来显示 Kafka Streams 的拓扑。
在这里，您将看到可用流的摘要。对我来说，它是 12 个主题和 15 个国家商店。您还可以看到 Kafka Streams 拓扑的可视化。下图显示了拓扑的片段。您可以通过单击屏幕右侧可见的绿色下载按钮来下载完整图像。
现在，让我们使用 Redpanda CLI 显示已创建主题的列表。就我而言，Redpanda 在55001本地端口下可用。所有主题都是 Quarkus 在应用程序启动期间自动创建的。我们需要定义用于两个测试应用程序之间通信的主题名称。这些主题是orders.buy：orders.sell和transactions。它们由order-service. 正在创建下面可见的stock-service所有其他主题，这些主题负责处理流。
$ rpk topic list --brokers localhost:55001
NAME PARTITIONS REPLICAS
orders.buy 1 1
orders.sell 1 1
stock-KSTREAM-JOINOTHER-0000000016-store-changelog 1 1
stock-KSTREAM-JOINOTHER-0000000043-store-changelog 1 1
stock-KSTREAM-JOINOTHER-0000000065-store-changelog 1 1
stock-KSTREAM-JOINTHIS-0000000015-store-changelog 1 1
stock-KSTREAM-JOINTHIS-0000000042-store-changelog 1 1
stock-KSTREAM-JOINTHIS-0000000064-store-changelog 1 1
stock-KSTREAM-KEY-SELECT-0000000005-repartition 1 1
stock-KSTREAM-KEY-SELECT-0000000006-repartition 1 1
stock-KSTREAM-KEY-SELECT-0000000032-repartition 1 1
stock-KSTREAM-KEY-SELECT-0000000033-repartition 1 1
stock-KSTREAM-KEY-SELECT-0000000054-repartition 1 1
stock-KSTREAM-KEY-SELECT-0000000055-repartition 1 1
stock-transactions-all-summary-changelog 1 1
stock-transactions-all-summary-repartition 1 1
stock-transactions-per-product-summary-30s-changelog 1 1
stock-transactions-per-product-summary-30s-repartition 1 1
stock-transactions-per-product-summary-changelog 1 1
stock-transactions-per-product-summary-repartition 1 1
transactions 1 1
为了进行完整的测试，我们还需要运行order-service. 它不断生成订单并将它们发送到orders.buy或orders.sell主题。让我们这样做。
使用 Quarkus 向 Redpanda 发送消息
在我们运行之前order-service，让我们看看一些实现细节。在生产者方面，我们需要包含一个负责与 Kafka 代理集成的依赖项：
<dependency>
<groupId>io.quarkus</groupId>
<artifactId>quarkus-smallrye-reactive-messaging-kafka</artifactId>
</dependency>
我们的应用程序生成随机订单并将其发送到orders.buy或orders.sell主题。有两种方法，每种方法都专用于一个主题。让我们看一个生成BUY订单的方法。我们需要对其进行注释@Outgoing并设置频道名称 ( orders-buy)。我们的方法每 500 毫秒生成一个订单。
@Outgoing("orders-buy")
public Multi<Record<Long, Order>> buyOrdersGenerator() {
return Multi.createFrom().ticks().every(Duration.ofMillis(500))
.map(order -> {
Integer productId = random.nextInt(10) + 1;
int price = prices.get(productId) + random.nextInt(200);
Order o = new Order(
incrementOrderId(),
random.nextInt(1000) + 1,
productId,
100 * (random.nextInt(5) + 1),
LocalDateTime.now(),
OrderType.BUY,
price);
log.infof("Sent: %s", o);
return Record.of(o.getId(), o);
});
}
之后，我们需要将频道名称映射为目标主题名称。另一个必需的操作是为消息键和值设置序列化程序。
mp.messaging.outgoing.orders-buy.connector = smallrye-kafka
mp.messaging.outgoing.orders-buy.topic = orders.buy
mp.messaging.outgoing.orders-buy.key.serializer = org.apache.kafka.common.serialization.LongSerializer
mp.messaging.outgoing.orders-buy.value.serializer = io.quarkus.kafka.client.serialization.ObjectMapperSerializer
最后，转到order-service目录并运行应用程序。
$ cd order-service
$ mvn quarkus:dev
一旦您开始order-service，它将创建主题并开始发送订单。它使用与 Redpanda 相同的实例stock-service。您可以docker ps再次运行该命令来验证它。
现在，只需进行简单的更改stock-service即可重新加载应用程序。它还将重新加载 Kafka Streams 拓扑。之后，它开始接收来自order-service. 最后，它将根据传入的订单创建交易并将它们存储在transactions主题中。
使用测试容器云
在我们的开发过程中，我们需要有一个本地安装的 Docker 生态系统。但是，如果我们没有它怎么办？这就是 Testcontainers Cloud 的用武之地。Testcontainers Cloud 是开发人员优先的 SaaS 平台，用于与真实数据库、消息代理、云服务或应用程序基础架构的任何其他组件进行现代集成测试。为简化起见，我们将做与以前相同的事情，但我们的 Redpanda 和 PostgreSQL 实例不会在本地 Docker 上运行，而是在远程 Testcointainers 平台上运行。
您需要做什么才能启用 Testcontainers Cloud？首先，从以下站点下载代理。您还需要成为 beta 测试人员才能获得授权令牌。最后，只需运行代理并终止您的本地 Docker 守护程序。您应该会在正在运行的应用程序中看到 Testcontainers 图标，其中包含有关与云的连接的信息。
Docker 不应该在本地运行。
和以前一样，只需使用quarkus:dev命令运行两个应用程序。您的 Redpanda 代理在 Testcontainers Cloud 上运行，但是，由于代理，您可以通过 localhost 访问它。
您可以再次使用以下命令为新代理验证主题列表：
$ rpk topic list --brokers localhost:59779
最后的想法
在本文中，我重点向您展示了 Quarkus、Redpanda 和 Testcontainers 等令人兴奋的新技术如何协同工作。本地开发是用例之一，但您也可以使用它们来编写集成测试。
京公网安备 11010802035086号