[关闭]
@levinzhang 2022-08-20T01:36:09.000000Z 字数 14965 阅读 260

Kafka Streams与Quarkus：实时处理事件

by

摘要：

消费Kafka消息是很容易的，只要有消息生成，我们就可以得到它们，但是除此之外，我们也做不了其他的事情了。如果我们需要实时处理数据（过滤、连接或操作事件）的话，仅仅使用Kafka的消费API可能就不是最好的方式了，因为这将导致代码变得非常复杂。在实时处理Kafka事件方面，Kafka Streams和Quarkus是绝配。

核心要点

在本系列的第一部分中，我们学习了Apache Kafka和Quarkus的集成，并开发了一个简单的应用，从两个Kafka主题生产和消费事件。

在那个样例中，我们模拟了一个影视流公司，在一个Kafka主题中存储电影信息，在另外一个主题中存储了用户停止观看电影时所发生的每个事件，并捕获了电影已播放的时间。

下图展示了该应用的架构：

我们可以看到，消费消息很简单，只要有消息生成，我们就可以得到它们，但是除此之外，我们也做不了其他的事情了。如果我们需要实时处理数据（比如过滤或操作事件）或者我们需要在事件之间做一些关联，单纯使用Kafka的消费API可能就不是最佳的方式了，因为这会导致代码非常复杂。

Kafka Streams

Kafka Streams项目能够帮助我们在事件产生时实时消费它们，应用各种转换，执行流连接等，并且可以选择性地将新的数据表述写回主题中。

对于有状态和无状态的流应用来说，Kafka Streams都是理想方案，它能够实现基于时间的操作（例如，围绕给定的时间段对事件进行分组），并且考虑到了Kafka生态系统中普遍存在的可扩展性、可靠性和可维护性。

Kafka Stream由三个元素组成，即输入（源处理器）、输出（sink处理器）和处理器（流处理器）。

源处理器（Source processor）：源处理器代表一个Kafka主题。源处理器会发送事件到一个或多个流处理器中。

流处理器（Stream processor）：流处理器会将转换/逻辑应用于输入流中，比如连接、分组、计数、映射等。流处理器可以连接至另一个流处理器和/或sink处理器。

Sink处理器（Sink processor）：Sink处理器代表了输出的数据，它会连接至一个Kafka主题。

拓扑结构（topology）是由源、处理器和sink组成的无循环图，事件会传入到一个Kafka Streams中，该实例将开始拓扑结构的执行。

Kafka Streams和Quarkus

Quarkus使用Quarkus KStreams扩展实现与Kafka Streams集成。

Quarkus起步

使用Quarkus最快捷的方式是通过初始化页面添加所需的依赖。每个服务可能需要不同的依赖，你可以选择Java 11或Java 17。为了实现Quarkus与Kafka Streams的集成，我们至少需要添加Kafka Streams扩展。

要开发的应用

正如在本文开始时所提到的，在本系列的第一部分中，我们开发了一个影视流公司，它有两个Kafka主题，其中一个用来存储电影的列表，另外一个主题会在用户停止播放电影时存储用户所在的区域（事件的键），并且会以电影id和播放时间作为事件的值。

所有的这些逻辑都是在名为Movie Plays Producer的生成者服务中创建的，该服务是使用Quarkus开发的。

除此之外，我还使用Quakus开发了一个Movie Plays Consumer服务，它会消费这两个主题的事件并且会在控制台上展示它们（并实现了HTTP服务器端事件）。

但是，这里没有对数据进行任何处理，它只是按照原样进行了接收。如果我们想要在movies和playtimemovies主题之间进行一下连接，在获取电影播放时长的时候得到电影的详细信息而不是id的话，那又该怎么办呢？

如果仅仅使用Kafka消息来实现这样的逻辑会变成一项很复杂的任务，因为我们需要在一个Map中存储Movie信息，并且在每个playedmovie事件发生时，进行匹配处理。

Movie Plays KStream

与其为每个用例手工编写代码，不如看一下如何使用Kafka Streams，以及它是如何与Quarkus集成来解决这个问题的。

创建项目

导航至Quarkus的初始化页面，并选择Apache Kafka Streams扩展来实现与Kafka Streams的集成。然后，选择RestEasy和RestEasy Jackson扩展实现事件从Java对象和JSON之间的编排/解排。同时，取消选中Started Code生成选项。

请参照下面的截图：

你也可以跳过这个手动的步骤并导航至Kafka Stream Quarkus Generator链接，在这里，所有的依赖都已经选择好了。然后，点击Generate your application按钮，以下载应用骨架的压缩文件。

解压文件，并在你最喜欢的IDE中打开项目。

开发

当开发Kafka Stream应用时，我们需要做的第一件事就是创建Topology实例，并定义源、处理器和sink。

在Quarkus中，我们只需要创建一个CDI类，这个类需要包含一个返回Topology实例的方法。

创建名为TopologyProducer的类，它将会实现从这两个主题消费事件并连接它们的逻辑。最后，生成的结果将会发送至一个sink处理器，该处理器以控制台输出的形式展示结果。

还有一个元素我们没有提到，在这些场景中它非常有用，那就是Kafka Tables。

一个主题可以包含具有相同键的多个事件。例如，我们可以使用某个键插入一个电影，然后我们可以使用相同的键创建一个新的事件来对电影进行更新：

但是，如果我们想要让movies主题与playtimemovies主题进行连接的话，那我们该选择使用哪个值为1的事件呢？第一个还是第二个？在这个具体的情况中，应该选择最新的那一个，因为它包含了电影的最新版本。为了获取每个事件的最新版本，Kafka Streams有一个表的概念（KTable/GlobalKTable）。

Kafka Streams会浏览指定的主题，获取每个事件的最新版本，并将其放到一个表实例中。

KafkaStream扩展并不会像Kafka Messaging集成那样自动注册SerDes，所以我们需要在拓扑中手动注册它们。

  1. package org.acme;
  3. import javax.enterprise.context.ApplicationScoped;
  4. import javax.enterprise.inject.Produces;
  6. import org.apache.kafka.common.serialization.Serdes;
  7. import org.apache.kafka.streams.KeyValue;
  8. import org.apache.kafka.streams.StreamsBuilder;
  9. import org.apache.kafka.streams.Topology;
  10. import org.apache.kafka.streams.kstream.Consumed;
  11. import org.apache.kafka.streams.kstream.GlobalKTable;
  12. import org.apache.kafka.streams.kstream.KStream;
  13. import org.apache.kafka.streams.kstream.Printed;
  15. import io.quarkus.kafka.client.serialization.ObjectMapperSerde;
  17. @ApplicationScoped
  18. public class TopologyProducer {
  20.    private static final String MOVIES_TOPIC = "movies";
  21.    private static final String PLAY_MOVIES_TOPIC = "playtimemovies";
  23.    @Produces
  24.    public Topology getTopCharts() {
  26.        final StreamsBuilder builder = new StreamsBuilder();
  28. // 用于Movie和PlayedMovie的SerDes
  30.        final ObjectMapperSerde<Movie> movieSerder = new ObjectMapperSerde<>(Movie.class);
  31.        final ObjectMapperSerde<MoviePlayed> moviePlayedSerder = new ObjectMapperSerde<>(MoviePlayed.class);
  33.     // 为Movies主题创建一个Global Kafka Table
  35.        final GlobalKTable<Integer, Movie> moviesTable = builder.globalTable(
  36.                MOVIES_TOPIC,
  37.                Consumed.with(Serdes.Integer(), movieSerder));
  39.     // 连接至playtimemovies主题的流，每当该主题有事件生成都会被该流所消费
  41.        final KStream<String, MoviePlayed> playEvents = builder.stream(
  42.                PLAY_MOVIES_TOPIC, Consumed.with(Serdes.String(), moviePlayedSerder));
  44. // PlayedMovies使用区域作为键，对象作为值。我们对内容进行map操作，让电影的id作为key（以便于进行连接）并让对象继续作为值
  45. // 另外，我们使用movies表的键（movieId）以及流的键（在前面的map方法中，我们也将其变成了movieId）进行连接
  47. // 最后，结果会流向控制台
  49.     playEvents
  50.            .map((key, value) -> KeyValue.pair(value.id, value)) // Now key is the id field
  51.            .join(moviesTable, (movieId, moviePlayedId) -> movieId, (moviePlayed, movie) -> movie)
  52.            .print(Printed.toSysOut());
  53.        return builder.build();
  55.    }
  56. }

Movie和MoviePlayed POJO包含了实现逻辑所需的属性：

Movie对象如下所示：

  1. package org.acme;
  3. public class Movie {
  5.    public int id;
  6.    public String name;
  7.    public String director;
  8.    public String genre;
  10.    public Movie(int id, String name, String director, String genre) {
  11.        this.id = id;
  12.        this.name = name;
  13.        this.director = director;
  14.        this.genre = genre;
  15.    }
  16. }

MoviePlayed对象如下所示：

  1. package org.acme;
  3. public class MoviePlayed {
  5.    public int id;
  6.    public long duration;
  8.    public MoviePlayed(int id, long duration) {
  9.        this.id = id;
  10.        this.duration = duration;
  11.    }
  13. }

运行Kafka Stream应用之前的最后一步是配置参数，其中最重要的是quarkus.kafka-streams.topics。它是一个主题列表，在拓扑结构开始处理数据之前，它们就要存在于Kafka集群中，这是一个前提条件。

打开src/main/resources/application.properties文件并添加如下的代码行：

  1. kafka-streams.cache.max.bytes.buffering=10240
  2. kafka-streams.commit.interval.ms=1000
  3. kafka-streams.metadata.max.age.ms=500
  4. kafka-streams.auto.offset.reset=earliest
  5. kafka-streams.metrics.recording.level=DEBUG
  7. quarkus.kafka-streams.topics=playtimemovies,movies

现在，我们可以测试一下流了。我们启动在上一篇文章中开发的生产者。生产者的源码可以在这里找到。

Quarkus KStreams集成了Quarkus DevServices。所以，我们不需要启动Kafka集群，也不需要配置它的位置，因为Quarkus Dev模式会处理好所有的事情。我们只需要记住在自己的计算机上要有一个运行中的容器环境即可，比如Podman或其他兼容OCI的工具。

在终端窗口中启动生产者服务：

  1. cd movie-plays-producer
  2. ./mvnw compile quarkus:dev
  4. 2022-04-11 07:49:31,900 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Cruella played for 287 minutes
  5. 2022-04-11 07:49:31,941 INFO  [io.quarkus] (Quarkus Main Thread) movie-plays-producer 1.0.0-SNAPSHOT on JVM (powered by Quarkus 2.7.3.Final) started in 4.256s.
  6. 2022-04-11 07:49:31,942 INFO  [io.quarkus] (Quarkus Main Thread) Profile dev activated. Live Coding activated.
  7. 2022-04-11 07:49:31,943 INFO  [io.quarkus] (Quarkus Main Thread) Installed features: [cdi, kafka-client, smallrye-context-propagation, smallrye-reactive-messaging, smallrye-reactive-messaging-kafka, vertx]
  8. 2022-04-11 07:49:32,399 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Encanto played for 162 minutes
  9. 2022-04-11 07:49:32,899 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie The Hobbit played for 255 minutes
  10. 2022-04-11 07:49:33,404 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Sing 2 played for 264 minutes
  11. 2022-04-11 07:49:33,902 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Encanto played for 28 minutes
  12. 2022-04-11 07:49:34,402 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Star Trek: First Contact played for 137 minutes
  13. 2022-04-11 07:49:34,903 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Star Trek: First Contact played for 277 minutes
  14. 2022-04-11 07:49:35,402 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie The Hobbit played for 141 minutes

在另一个终端窗口中，启动我们刚刚开发的Kafka Stream代码：

  1. ./mvnw compile quarkus:dev
  3. 2022-04-11 07:54:59,321 INFO  [org.apa.kaf.str.pro.int.StreamTask] (movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1) stream-thread [movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1] task [1_0] Restored and ready to run
  4. 2022-04-11 07:54:59,322 INFO  [org.apa.kaf.str.pro.int.StreamThread] (movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1) stream-thread [movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1] Restoration took 74 ms for all tasks [1_0]
  5. 2022-04-11 07:54:59,322 INFO  [org.apa.kaf.str.pro.int.StreamThread] (movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1) stream-thread [movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1] State transition from PARTITIONS_ASSIGNED to RUNNING
  6. 2022-04-11 07:54:59,324 INFO  [org.apa.kaf.str.KafkaStreams] (movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea-StreamThread-1) stream-client [movie-plays-kstreams-22c86daa-cd28-4956-9d0d-57b6b282a2ea] State transition from REBALANCING to RUNNING
  7. [KSTREAM-LEFTJOIN-0000000005]: 4, Movie [director=Craig Gillespie, genre=Crime Comedy, id=4, name=Cruella]
  8. [KSTREAM-LEFTJOIN-0000000005]: 2, Movie [director=Jonathan Frakes, genre=Space, id=2, name=Star Trek: First Contact]
  9. [KSTREAM-LEFTJOIN-0000000005]: 4, Movie [director=Craig Gillespie, genre=Crime Comedy, id=4, name=Cruella]
  10. [KSTREAM-LEFTJOIN-0000000005]: 1, Movie [director=Peter Jackson, genre=Fantasy, id=1, name=The Hobbit]
  11. [KSTREAM-LEFTJOIN-0000000005]: 4, Movie [director=Craig Gillespie, genre=Crime Comedy, id=4, name=Cruella]
  12. [KSTREAM-LEFTJOIN-0000000005]: 4, Movie [director=Craig Gillespie, genre=Crime Comedy, id=4, name=Cruella]
  13. [KSTREAM-LEFTJOIN-0000000005]: 3, Movie [director=Jared Bush, genre=Animation, id=3, name=Encanto]
  14. [KSTREAM-LEFTJOIN-0000000005]: 5, Movie [director=Garth Jennings, genre=Jukebox Musical Comedy, id=5, name=Sing 2]

在输出中打印的是事件（连接所生成的结果），其中键是movieId，值是movie本身。我们现在所具有的功能是，每当一个电影停止播放时，Kafka Stream会对其进行处理并以Movie全量信息的形式对其进行展现。

到目前为止，还不算复杂，对于这样的场景，你可能会想我们根本没有必要使用Kafka Streams。但是，我们再加一些需求，这样你就能看到它的强大之处了。

现在，我们不是在用户每次停掉电影的时候都生成事件，而是只对用户观看时间超过10分钟的电影发送事件。

我们可以使用filter方法按照持续时长进行过滤。

  1. playEvents
  2.        .filter((region, event) -> event.duration >= 10) // filters by duration
  3.        .map((key, value) -> KeyValue.pair(value.id, value))
  4.        .join(moviesTable, (movieId, moviePlayedId) -> movieId, (moviePlayed, movie) -> movie)
  5.         .print(Printed.toSysOut());

重启应用，我们可以发现观看时长小于10分钟的电影将不会被处理。

我们发现，Kafka Streams有助于代码的整洁性，接下来，我们添加最后的需求。现在，我们对每部电影的播放时长并不感兴趣，而是对每部电影有多少次超过10分钟的播放感兴趣。

到目前为止，对事件的处理都是无状态的，因为事件都是遵循这样的步骤，即接收、处理并发送至sink处理器（也就是发送至一个主题或控制台输出），但是，为了统计某部电影播放的次数，我们需要在内存记住电影被播放了多少次，并且当任意用户再次观看超过10分钟的时候，要对这个统计数字递增一次。此时，事件的处理就要以有状态的方式进行了。

我们需要做的第一件事就是创建一个Java类，以存储电影的名称及其播放的次数。

  1. public class MoviePlayCount {
  2.    public String name;
  3.    public int count;
  5.    public MoviePlayCount aggregate(String name) {
  6.        this.name = name;
  7.        this.count++;
  9.        return this;
  10.    }
  12.    @Override
  13.    public String toString() {
  14.        return "MoviePlayCount [count=" + count + ", name=" + name + "]";
  15.    }
  17. }

这是一个计数器类，它依然需要两样东西：

关于第一个问题，我们需要使用KeyValueBytesStoreSupplier接口。

  1. public static final String COUNT_MOVIE_STORE = "countMovieStore";
  3. KeyValueBytesStoreSupplier storeSupplier = Stores.persistentKeyValueStore(COUNT_MOVIE_STORE);

对于第二个问题，Kafka Streams有一个用来聚合结果的aggregate方法。

在我们的用例中，也就是每部电影播放时长超过10分钟的次数。

  1. // MoviePlayCount可以被序列化和反序列化
  2. ObjectMapperSerde<MoviePlayCount> moviePlayCountSerder = new ObjectMapperSerde<>(MoviePlayCount.class);
  4. // 这是之前的连接操作，其中键是电影id，值是电影
  5. .join(moviesTable, (movieId, moviePlayedId) -> movieId, (moviePlayed, movie) -> movie)
  6. // 根据键对事件进行分组，在本例中，也就是电影的id
  7. .groupByKey(Grouped.with(Serdes.Integer(), movieSerder))
  8.  // 聚合方法，如果MoviePlayCount已经创建的话，获取该对象（如果尚未创建的话，会创建该实例）并调用其aggregate方法，以对观看次数进行递增
  9.  .aggregate(MoviePlayCount::new,
  10.              (movieId, movie, moviePlayCounter) -> moviePlayCounter.aggregate(movie.name),
  11.               Materialized.<Integer, MoviePlayCount> as(storeSupplier)
  12.                   .withKeySerde(Serdes.Integer())
  13.                   .withValueSerde(moviePlayCountSerder)
  14.              )

重启应用，在控制台上将会展示电影播放的次数。

提示：要重启应用，只需在终端输入“s”，应用将会自动重启。

应用重启之后，控制台将会展示每部电影的状态。

  1. [KTABLE-TOSTREAM-0000000011]: 4, MoviePlayCount [count=13, name=Cruella]
  2. [KTABLE-TOSTREAM-0000000011]: 3, MoviePlayCount [count=11, name=Encanto]
  3. [KTABLE-TOSTREAM-0000000011]: 5, MoviePlayCount [count=14, name=Sing 2]
  4. [KTABLE-TOSTREAM-0000000011]: 2, MoviePlayCount [count=15, name=Star Trek: First Contact]
  5. [KTABLE-TOSTREAM-0000000011]: 1, MoviePlayCount [count=16, name=The Hobbit]
  6. [KTABLE-TOSTREAM-0000000011]: 2, MoviePlayCount [count=16, name=Star Trek: First Contact]
  7. [KTABLE-TOSTREAM-0000000011]: 3, MoviePlayCount [count=12, name=Encanto]
  8. [KTABLE-TOSTREAM-0000000011]: 2, MoviePlayCount [count=17, name=Star Trek: First Contact]
  9. [KTABLE-TOSTREAM-0000000011]: 5, MoviePlayCount [count=15, name=Sing 2]
  10. [KTABLE-TOSTREAM-0000000011]: 4, MoviePlayCount [count=14, name=Cruella]
  11. [KTABLE-TOSTREAM-0000000011]: 1, MoviePlayCount [count=17, name=The Hobbit]
  12. [KTABLE-TOSTREAM-0000000011]: 4, MoviePlayCount [count=15, name=Cruella]
  13. [KTABLE-TOSTREAM-0000000011]: 4, MoviePlayCount [count=16, name=Cruella]

交互式查询

因为我们将sink处理器设置成了System.out流，所以聚合结果会流向控制台。

  1. .toStream()
  2. .print(Printed.toSysOut());

但是，我们也可以将结果流发往一个Kafka主题：

  1. .to("counter_movies",                      Produced.with(Serdes.Integer(), moviePlayCountSerder)
  2. );

但是，如果我们感兴趣的不是每次对新事件的反应，而只是想查询特定的电影此时播放的次数，那又该怎么办呢？

Kafka Streams的交互式查询(interactive query)允许我们直接查询底层存储，以获取给定键相关的值。

首先，我们创建一个名为MoviePlayCountData的类来存储查询结果。按照这种方式，我们可以解耦Kafka Streams使用的类与应用中其他部分所使用的类。

  1. public class MoviePlayCountData {
  3.    private String name;
  4.    private int count;
  6.    public MoviePlayCountData(String name, int count) {
  7.        this.name = name;
  8.        this.count = count;
  9.    }
  11.    public int getCount() {
  12.        return count;
  13.    }
  15.    public String getName() {
  16.        return name;
  17.    }
  19. }

现在，创建名为InteractiveQueries的类来实现对状态存储（KeyValueBytesStoreSupplier）的访问并根据电影的id查询它被播放的次数。

  1. import javax.enterprise.context.ApplicationScoped;
  2. import javax.inject.Inject;
  4. import org.apache.kafka.streams.KafkaStreams;
  5. import org.apache.kafka.streams.errors.InvalidStateStoreException;
  6. import org.apache.kafka.streams.state.QueryableStoreTypes;
  7. import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  9. import static org.apache.kafka.streams.StoreQueryParameters.fromNameAndType;
  11. import java.util.Optional;
  13. @ApplicationScoped
  14. public class InteractiveQueries {
  16.    @Inject
  17.    KafkaStreams streams;
  19.    public <MoviePlayCountData> getMoviePlayCountData(int id) {
  20.        // 获取状态存储并根据电影id获取播放次数
  21.        MoviePlayCount moviePlayCount = getMoviesPlayCount().get(id);
  22.        // 如果有结果的话
  23.        if (moviePlayCount != null) {
  24.            // 将结果包装到MoviePlayCountData中
  25.            return Optional.of(new MoviePlayCountData(moviePlayCount.name, moviePlayCount.count));
  26.        } else {
  27.            return Optional.empty();
  28.        }
  29.    }
  31.    // 获取状态存储
  32.    private ReadOnlyKeyValueStore<Integer, MoviePlayCount> getMoviesPlayCount() {
  33.        while (true) {
  34.            try {
  35.                return streams.store(fromNameAndType(TopologyProducer.COUNT_MOVIE_STORE, QueryableStoreTypes.keyValueStore()));
  36.            } catch (InvalidStateStoreException e) {
  37.                // 忽略之，此时存储尚未就绪
  38.            }
  39.        }
  40.    }
  42. }

现在，我们可以添加一个简单的REST端点来运行该查询。

  1. import java.util.Optional;
  3. import javax.inject.Inject;
  4. import javax.ws.rs.GET;
  5. import javax.ws.rs.Path;
  6. import javax.ws.rs.PathParam;
  7. import javax.ws.rs.core.Response;
  8. import javax.ws.rs.core.Response.Status;
  10. @Path("/movie")
  11. public class MovieCountResource {
  13.    // 注入前文的类进行查询
  14.    @Inject
  15.    InteractiveQueries interactiveQueries;
  17.    @GET
  18.    @Path("/data/{id}")
  19.    public Response movieCountData(@PathParam("id") int id) {
  20.        Optional<MoviePlayCountData> moviePlayCountData = interactiveQueries.getMoviePlayCountData(id);
  22.        // 根据结果判定返回值还是404
  23.        if (moviePlayCountData.isPresent()) {
  24.            return Response.ok(moviePlayCountData.get()).build();
  25.        } else {
  26.            return Response.status(Status.NOT_FOUND.getStatusCode(),
  27.                    "No data found for movie " + id).build();
  28.        }
  30.    }
  31. }

Kafka Stream实现的模式如下图所示：

扩展

Kafka Streams应用可以进行扩展，所以流会分布到多个实例中。在这种情况下，每个实例都包含聚合结果的一个子集，所以要想获得总的聚合结果，我们需要通过将REST API重定向到另外的实例来获取其他实例的数据。

Kafka Streams提供了一个API，可以知道要请求的数据是在本地Kafka Streams存储中还是在其他的主机中。

虽然这个过程并不复杂，但已经超出了本文的讨论范围。

结论

到目前为止，我们已经看到，将Quarkus应用连接到Apache Kafka并生产和消费主题中的消息/事件是很容易的。此外，还看到Kafka Streams让我们不仅可以消费Kafka中消息，还能够实时处理它们，进行转换、过滤等操作，例如以同步的方式消费结果数据。 这是一项强大的技术，当需要处理的数据不断变化时，它可以轻松扩展，提供实时的体验。

但是，我们还没有解决该架构的最后一个问题。通常情况下，数据并不是存储在一个地方。电影信息可能存储在关系型数据库中，而电影的播放信息则存储在一个Kafka主题中。那么，该如何保持这两个地方的信息更新，以便Kafka Streams能正确地连接数据呢？

这里有一个缺失的部分，名为Debezium的项目可以帮助我们解决这个问题。我们将用一整篇文章来介绍Debezium和Quarkus，敬请持续关注。

关于作者

Alex Soto

Alex Soto是红帽公司的开发者体验总监。他对Java领域、软件自动化充满热情，他相信开源软件模式。Soto是Manning的《Testing Java Microservices》和O’Reilly的《Quarkus Cookbook》两本书的共同作者，他还是多个开源项目的贡献者。自 2017 年以来，他一直是Java Champion，是国际演讲者和Salle URL 大学的教师。你可以在 Twitter 上关注他（Alex Soto ⚛️），随时了解 Kubernetes 和 Java 领域的动态。

查看英文原文：Kafka Streams and Quarkus: Real-Time Processing Events

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注