@levinzhang 2022-08-02T06:09:37.000000Z 字数 14010 阅读 432

使用Apache Kafka实现Quarkus的反应式消息

by
* Alex Soto

摘要：

现在，数据处理和消费的方式与以往的实践是不一样的。在过去，数据会存储到一个数据库中，并且会进行批处理以获取分析结果。Apache Kafka是一个分布式的事件存储和流处理平台，能够实时地存储、消费和处理数据流。在本文中，我们将会学习如何使用Apache Kafka和Quarkus生产和消费数据。

核心要点

Apache Kafka（简称为Kafka）是一个分布式的事件存储和流处理平台，能够存储、消费和处理数据流。
Quarkus能够与使用Apache Kafka进行事件处理的反应式应用完美集成。
Quarkus Dev Services为我们提供了一种自动化启动Kafka集群的方式，以方便我们对应用进行测试。
Apache Kafka能够取代像JMS这样的消息系统，尤其是当我们需要实时流处理的时候。

什么是Apache Kafka以及为何要使用它？

现在，数据处理和消费的方式与以往的实践是不一样的。在过去，数据会存储到一个数据库中，并且会进行批处理以获取分析结果。

尽管这种方式依然占有一席之地，但是更现代化的平台能够让我们在数据进入系统的时候对其进行实时处理。

Apache Kafka（简称为Kafka）是一个分布式的事件存储和流处理平台，能够存储、消费和处理数据流。

要理解Kafka是如何运行的，我们需要掌握五个基本概念：

事件（Event，或者叫做消息）：事件是一个具有时间戳的键-值对，代表了存储在系统中需要进行处理的数据。从Kafka的角度来看，它就是一堆字节。
分区（Partition）：分区是生产和消费事件的地方。在分区中，能够保证事件的顺序。
主题（Topic）：一个主题是由一个或多个分区组成的。主题是开发人员使用的工作单元，能够消费或生产事件。
消费者（Consumer）：消费者订阅某个主题，这样每当事件发布到该主题时，消费者都会得到通知。
生产者（Producer）：生产者发布事件到主题（实际上是属于该主题的某个分区）中。

Apache Kafka的一个重要特点是它在创建时充分考虑到了可扩展性和容错性，使其非常适合高性能的应用。我们认为，Kafka可以取代一些传统的消息系统，比如Java Message Service（JMS）和Advanced Message Queuing Protocol（AMQP）。

Apache Kafka能够与当前使用的大多数语言进行集成，但是在本文中，我们将会讨论它与Java的集成，具体来讲是与Quarkus Java栈的集成。

Quarkus是什么？

Quarkus是一个全栈、Kubernetes原生的Java框架，适用于Java虚拟机（JVM）和原生编译环境，专门为容器中运行的Java进行了优化，使其成为Serverless、云和Kubernetes环境下的高效平台。

Quarkus没有重复发明轮子，而是使用了由标准/规范支撑的知名企业级框架，并且使它们能够借助GraalVM编译为二进制文件。

如何在Quarkus中集成Kafka？

Quarkus使用SmallRye Reactive Messaging项目实现与Apache Kafka的交互。

Quarkus入门

要开始使用Quarkus，最快捷的方式就是通过其初始化页面添加所需的依赖。每个服务可能会有不同的依赖，我们可以在Java 11或Java 17之间选择。要实现Quarkus与Kafka的集成，我们需要添加SmallRye Reactive Messaging - Kafka Connector扩展。

要开发的应用

假设我们是一家影视流媒体公司，其中有个用例就是保存电影。这的确可以通过传统数据库来实现，但是考虑到要实现良好的用户体验需要实时互动，因此我们决定将它们存储在Kafka中。

所以，我们会有两个服务，其中一个服务会在用户停止播放电影的时候生成一个事件，另外一个服务则会消费这些事件，并以服务器事件的方式对其进行展示和流式处理。

下图展示了应用的架构：

接下来，我们使用Quarkus实现这些服务并阐述一些内部的细节。

电影播放的生产者（Movie Plays Producer）

每当用户停止播放电影的时候，该服务会向Kafka PlaytimeMovies主题发送一个事件。该事件包含了电影的ID以及观看的总时间。为了便于显示，我们将会使用一个定时器自动触发模拟用户观看电影的逻辑。

当该服务启动的时候，它将会生成一些电影到Kafka Movies主题中。

创建项目

导航至Quarkus的初始页面并选择smallrye-reactive-messaging-kafka以便于集成Kafka。然后，选择Jackson扩展，用于实现事件在JSON和Java的对象-字节数组之间进行编排/解排。同时，取消选中生成Started Code的选项。

如下面的截图所示：

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

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

开发

我们创建两个POJO，其中一个代表Movie，另外一个代表PlayedMovie。

public class Movie {
   public int id;
   public String name;
   public String director;
   public String genre;
   public Movie(int id, String name, String director, String genre) {
       this.id = id;
       this.name = name;
       this.director = director;
       this.genre = genre;
   }
}

Movie包含了电影的id、name、director和genre。

public class PlayedMovie {
   public int id;
   public long duration;
   public MoviePlayed(int id, long duration) {
       this.id = id;
       this.duration = duration;
   }
}

PlayedMovie包含了id和duration字段，分别代表了已播放电影的标识符以及用户观看的时长。

我们还需要一个新的类MovieKafkaGenerator，它负责将电影存储到Kafka主题中并模拟播放电影。

想要向主题上发布事件，我们首先需要两个类，分别是@Outgoing注解和Record类，@Outgoing用来以通道（channel）的形式指定要将事件发送至何处，在这里我们将其配置为指向Movies主题，Record代表了事件的包装器，以键/值的方式进行声明。

现在，我们创建MovieKafkaGenerator类，它会生成电影到Kafka Movies主题中。

package org.acme.movieplays;
import java.time.Duration;
import java.util.List;
import java.util.Random;
import javax.enterprise.context.ApplicationScoped;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import io.smallrye.mutiny.Multi;
import io.smallrye.reactive.messaging.kafka.Record;
@ApplicationScoped
public class MovieKafkaGenerator {
   private List<Movie> movies = List.of(
       new Movie(1, "The Hobbit", "Peter Jackson", "Fantasy"),
       new Movie(2, "Star Trek: First Contact", "Jonathan Frakes", "Space"),
       new Movie(3, "Encanto", "Jared Bush", "Animation"),
       new Movie(4, "Cruella", "Craig Gillespie", "Crime Comedy"),
       new Movie(5, "Sing 2", "Garth Jennings", "Jukebox Musical Comedy")
   );
   // 填充电影到Kafka主题中
   @Outgoing("movies")                                         
   public Multi<Record<Integer, Movie>> movies() {
       return Multi.createFrom().items(movies.stream()
               .map(m -> Record.of(m.id, m))
       );
   }
}

在这个类中，有几件重要的事情需要注意：

这个类是CDI作用域的类（@ApplicationScoped）
简单起见，我们将电影定义在了一个列表中
@Outgoing注解用来设置将事件发往何处（即movies通道）。movies()方法返回的元素将会自动发往定义的通道中。返回的类型可能是反应式/异步类型（在Quarkus中，通常会是io.smallrye.mutiny.Multi），以包装事件的内容。在后面，我们会将该通道配置为指向一个主题。
Record（即事件/消息）使用电影的ID作为key，并使用电影对象作为值。

最后一步就是配置连接至Kafka实例的Quarkus参数。Quarkus应用可以通过src/main/resources/目录下的application.properties文件进行配置。

通过如下的通用属性，我们可以很容易地配置通道和主题之间的关系：

mp.messaging.outgoing.<channel_name>.topic=<topic>

在我们的样例中，定义如下所示：

mp.messaging.outgoing.movies.topic=movies

你可能想知道Kafka broker的位置是在哪里配置的呢？对于本地开发来说，我们不需要进行配置，因为Quarkus为Kafka提供了Dev Services特性。Dev Services会提供所需的外部依赖的实例（如数据库实例、Kafka broker、Keycloak服务等），这些实例会在容器运行时中提供，比如Podman或者其他兼容OCI的工具。从开发者的角度来说，如果我们包括了某个扩展但是并没有对其进行配置的话，Quarkus会自动启动服务，并配置应用来使用它。

基于此，我们在开发过程中，并不需要其他的配置参数。Quarkus就会帮助我们实现这一点。

重要提示：要在运行样例，我们需要在本地机器上有一个正在运行的Docker主机。如果没有本地没有Docker的话，那就需要有一个部署好的Kafka broker，我们将会在后文介绍如何在Quarkus中配置这个“远程”实例。

Docker主机启动并运行起来之后，在终端中以Quarkus dev模式下启动应用：

./mvnw compile quarkus:dev

终端输出如下所示：

[INFO] Scanning for projects...
[INFO]
[INFO] -------------------&lt; org.acme:movie-plays-producer &gt;--------------------
[INFO] Building movie-plays-producer 1.0.0-SNAPSHOT
[INFO] --------------------------------[ jar ]---------------------------------
….
2022-03-21 11:37:24,474 INFO  [io.qua.sma.dep.processor] (build-8) Configuring the channel 'movies' to be managed by the connector 'smallrye-kafka'
2022-03-21 11:37:24,483 INFO  [io.qua.sma.rea.kaf.dep.SmallRyeReactiveMessagingKafkaProcessor] (build-30) Generating Jackson serializer for type org.acme.movieplays.Movie
--
--
Checking Docker Environment 2022-03-21 11:37:25,018 INFO  [org.tes.uti.ImageNameSubstitut
--
2022-03-21 11:37:28,500 INFO  [io.qua.kaf.cli.dep.DevServicesKafkaProcessor] (build-22) Dev Services for Kafka started. Other Quarkus applications in dev mode will find the broker automatically. For Quarkus applications in production mode, you can connect to this by starting your application with -Dkafka.bootstrap.servers=OUTSIDE://localhost:32769
2022-03-21 11:37:29,581 INFO  [io.quarkus] (Quarkus Main Thread) movie-plays-producer 1.0.0-SNAPSHOT on JVM (powered by Quarkus 2.7.3.Final) started in 6.666s.
2022-03-21 11:37:29,582 INFO  [io.quarkus] (Quarkus Main Thread) Profile dev activated. Live Coding activated.
2022-03-21 11:37:29,582 INFO  [io.quarkus] (Quarkus Main Thread) Installed features: [cdi, kafka-client, smallrye-context-propagation, smallrye-reactive-messaging, smallrye-reactive-messaging-kafka, vertx]

应用首先会被编译，然后自动配置Jackson序列化器（还记得我们在开始的时候就添加了该扩展），从而将Movie对象序列化为字节数组以存储在Kafka主题中。随后，Kafka broker会自动在localhost:32769启动，应用会自动配置连接到此处。最后，应用启动并运行，所有的电影会被插入到Kafka Movies主题中。

我们可以使用kcat工具来探查主题的内容。在终端窗口运行如下的命令，记得替换成你的Kafka broker地址：

kcat -b localhost:32769 -t movies -C -K:
:{"id":1,"name":"The Hobbit","director":"Peter Jackson","genre":"Fantasy"}
:{"id":2,"name":"Star Trek: First Contact","director":"Jonathan Frakes","genre":"Space"}
:{"id":3,"name":"Encanto","director":"Jared Bush","genre":"Animation"}
:{"id":4,"name":"Cruella","director":"Craig Gillespie","genre":"Crime Comedy"}
:{"id":5,"name":"Sing 2","director":"Garth Jennings","genre":"Jukebox Musical Comedy"}
% Reached end of topic movies [0] at offset 5

停掉应用，我们接下来添加生成电影播放的内容。

打开MovieKafkaGenerator类并添加如下代码：

private Random random = new Random();
@Inject
Logger logger;
@Outgoing("play-time-movies")
public Multi<Record<String, PlayedMovie>> generate() {
    return Multi.createFrom().ticks().every(Duration.ofMillis(1000))   
            .onOverflow().drop()
            .map(tick -> {
                Movie movie = movies.get(random.nextInt(movies.size()));
                int time = random.nextInt(300);
                logger.info("movie {0} played for {1} minutes", movie.name, time);
                // Region作为key
                return Record.of("eu", new PlayedMovie(movie.id, time));
    });
   }

在这个新方法中，有几件重要的事情需要注意：

生成的事件会被发送至play-time-movies通道。
每秒钟会触发一个新的事件。
这里会在列表中随机选择一部电影，并分配一个随机的播放时间。
在本例中，会创建一个Record（事件/消息），其中key代表了用户的区域，value则是PlayedMovie对象。

最后，打开application.properties文件并配置新的通道：

mp.messaging.outgoing.play-time-movies.topic=playtimemovies

再次启动应用，现在应用每秒钟都会生成一个新的事件。

./mvnw compile quarkus:dev

2022-03-21 12:36:01,297 INFO  [io.sma.rea.mes.kafka] (Quarkus Main Thread) SRMSG18258: Kafka producer kafka-producer-play-time-movies, connected to Kafka brokers 'OUTSIDE://localhost:32771', is configured to write records to 'playtimemovies'
2022-03-21 12:36:01,835 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Cruella played for 148 minutes
2022-03-21 12:36:02,336 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Star Trek: First Contact played for 288 minutes
2022-03-21 12:36:02,836 INFO  [org.acm.mov.MovieKafkaGenerator] (executor-thread-0) movie Cruella played for 176 minutes

每当有新事件发布时，控制台都会展示日志。现在，我们使用kcat来探查主题的内容。

kcat -b localhost:32773 -t playtimemovies -C -K:
eu:{"id":4,"duration":88}
eu:{"id":3,"duration":291}
eu:{"id":1,"duration":228}
eu:{"id":2,"duration":165}
eu:{"id":1,"duration":170}
eu:{"id":4,"duration":75}

电影播放的消费者（Movie Plays Consumer）

该服务负责消费来自Kafka主题的事件。被消费的事件是以HTTP服务器端事件的形式流向调用者的。事件是playedmovie数据，其中包含了播放的电影ID以及已观看的总时长。

创建项目

导航至Quarkus的初始页面，选择resteasy-reactive-jackson来实现JAX-RS反应式端点，借助Jackson实现对Java对象和JSON之间的编排/解排，并选择smallrye-reactive-messaging-kafka扩展实现与Kafka的集成。同时，取消选中Started Code生成选项。

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

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

开发

该服务会处理PlayedMovie事件，所以我们为该元素创建一个简单的POJO：

public class PlayedMovie {
   public int id;
   public long duration;
   public MoviePlayed(int id, long duration) {
       this.id = id;
       this.duration = duration;
   }
}

然后，创建名为PlayedMovieResource的新类，并创建一个JAX-RS反应式端点，以便于流入来自Kafka主题的事件。

import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
import org.eclipse.microprofile.reactive.messaging.Channel;
import io.smallrye.mutiny.Multi;
@Path("/movies")
public class PlayedMovieResource {
   @Channel("played-movies")
   Multi<MoviePlayed> playedMovies;
   @GET
   @Produces(MediaType.SERVER_SENT_EVENTS)
   public Multi<PlayedMovie> stream() {
       return playedMovives;
   }
}

这是一个很小的类，但是做了很多的事情：

使用@Path注解，暴露HTTP端点至/movies路径。
处理名为played-movies的通道上的事件。每当有新的事件发送至该通道（即事件发送至Kafka主题）时，它会自动发布到Multi实例中。
当使用HTTP GET方法调用/movies端点时，应用会在该通道中以流的方式传输接收到的事件。

最后，在application.properties文件中对通道进行配置，在这里我们配置了通道（包括主题和偏移策略）并且将监听端口修改为9090，这样的话就不会与监听8080端口的生产者服务冲突。

mp.messaging.incoming.movies-played.topic=playtimemovies
mp.messaging.incoming.movies-played.auto.offset.reset=earliest
%dev.quarkus.http.port=9090

在终端窗口中启动并运行movie-player-producer服务后，我们启动movie-player-consumer。在一个新的终端窗口中，以dev模式运行服务。

./mvnw compile quarkus:dev`

[INFO] Scanning for projects...
[INFO]
[INFO] -------------------< org.acme:movie-plays-consumer >--------------------
[INFO] Building movie-plays-consumer 1.0.0-SNAPSHOT
[INFO] --------------------------------[ jar ]---------------------------------
….
2022-03-21 17:59:08,079 INFO  [io.qua.sma.dep.processor] (build-13) Configuring the channel 'movies-played' to be managed by the connector 'smallrye-kafka'
2022-03-21 17:59:08,092 INFO  [io.qua.sma.rea.kaf.dep.SmallRyeReactiveMessagingKafkaProcessor] (build-33) Generating Jackson deserializer for type org.acme.movieplays.MoviePlayed
….
2022-03-21 17:59:10,617 INFO  [io.sma.rea.mes.kafka] (smallrye-kafka-consumer-thread-0) SRMSG18257: Kafka consumer kafka-consumer-movies, connected to Kafka brokers 'localhost:32771, belongs to the 'movie-plays-consumer'
….
2022-03-21 17:59:10,693 INFO  [io.quarkus] (Quarkus Main Thread) movie-plays-consumer 1.0.0-SNAPSHOT on JVM (powered by Quarkus 2.7.3.Final) started in 4.114s. Listening on: http://localhost:9090
2022-03-21 17:59:10,693 INFO  [io.quarkus] (Quarkus Main Thread) Profile dev activated. Live Coding activated.
2022-03-21 17:59:10,694 INFO  [io.quarkus] (Quarkus Main Thread) Installed features: [cdi, kafka-client, resteasy-reactive, resteasy-reactive-jackson, smallrye-context-propagation, smallrye-reactive-messaging, smallrye-reactive-messaging-kafka, vertx]

应用首先会被编译，然后自动配置Jackson序列化器（还记得我们在开始的时候就添加了该扩展），从而能够将存储在Kafka主题中的字节数组对象反序列化为Java对象。运行中的应用会探测到已经启动的Kafka集群并自动进行连接。最后，应用会在9090端口启动。

在新的窗口中，运行如下的命令，获取流式数据：

curl -N localhost:9090/movies

data:{"id":4,"duration":213}
data:{"id":4,"duration":3}
data:{"id":3,"duration":96}
data:{"id":5,"duration":200}
data:{"id":2,"duration":234}
data:{"id":1,"duration":36}
data:{"id":1,"duration":162}
data:{"id":3,"duration":88}

我们可以观察到来自Kafka主题的数据是如何自动进行流式处理并以HTTP请求的形式发出的。

上述的样例达成了两个目的，首先以Multi的形式注入一个通道，以便于接收事件，然后将这些事件发送至带有@Incoming注解的方法中。

停止消费者服务并添加如下的代码片段至PlayedMovieResource类中，以消费来自Kafka Movie主题的事件：

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.jboss.logging.Logger;
@Inject
Logger logger;
@Incoming("movies")
public void newMovie(Movie movie) {
   logger.infov("New movie: {0}", movie);
}

在本例中，每当有新的电影发布至movies通道（Movies主题）时，newMovie()方法就会被调用。方法的参数就是Kafka主题的负载。

在application.properties文件中配置通道，使其指向Movies主题。

mp.messaging.incoming.movies.topic=movies
mp.messaging.incoming.movies.auto.offset.reset=earliest

现在，再次启动movie-plays-consumer服务，我们会发现有些日志行打印出了电影的列表：

./mvnw compile quarkus:dev

[INFO] Scanning for projects...
[INFO]
[INFO] -------------------< org.acme:movie-plays-consumer >---------------------
[INFO] Building movie-plays-consumer 1.0.0-SNAPSHOT
[INFO] --------------------------------[ jar ]---------------------------------
…
2022-03-21 17:59:12,146 INFO  [io.sma.rea.mes.kafka] (vert.x-eventloop-thread-13) SRMSG18256: Initialize record store for topic-partition 'movies-0' at position -1.
2022-03-21 17:59:12,160 INFO  [org.acm.mov.MoviePlayedResource] (pool-1-thread-1) New movie: Movie [director=Peter Jackson, genre=Fantasy, id=1, name=The Hobbit]
2022-03-21 17:59:12,164 INFO  [org.acm.mov.MoviePlayedResource] (pool-1-thread-1) New movie: Movie [director=Jonathan Frakes, genre=Space, id=2, name=Star Trek: First Contact]
2022-03-21 17:59:12,165 INFO  [org.acm.mov.MoviePlayedResource] (pool-1-thread-1) New movie: Movie [director=Jared Bush, genre=Animation, id=3, name=Encanto]
2022-03-21 17:59:12,166 INFO  [org.acm.mov.MoviePlayedResource] (pool-1-thread-1) New movie: Movie [director=Craig Gillespie, genre=Crime Comedy, id=4, name=Cruella]
2022-03-21 17:59:12,167 INFO  [org.acm.mov.MoviePlayedResource] (pool-1-thread-1) New movie: Movie [director=Garth Jennings, genre=Jukebox Musical Comedy, id=5, name=Sing 2]

外部的Kafka Broker

你也可以使用自己的Kafka broker，只需要在application.properties文件中配置kafka.bootstrap.servers属性即可。

kafka-bootstrap.servers=kafka:9092

结论

到目前为止，连接Quarkus应用到Apache Kafka并通过主题生产和消费消息/事件是非常容易的。消费Kafka消息很简单，只要有消息生成，我们就可以得到它们，但是除此之外，我们也做不了其他的事情了。如果我们需要实时处理数据（比如过滤或操作事件）的话，又该怎么办呢？如果我们需要对事件做一些关联，又该怎么办呢（比如，playedmovie事件中包含电影的id，但是我们该如何与Movie主题联合起来获取电影的名字呢）？

当然，我们可以编写专门的代码来操作所有发送的数据。不过，Kafka Streams项目能够帮助我们在事件生成的时候来消费事件流，从而进行任意的转换、流连接等操作，并且有选择性地将新的数据写回到主题中。

Kafka Streams是一个很大的话题，它在解决实时处理问题上的各种功能令人印象深刻。我们将会用一篇专门的文章介绍Kafka Streams和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 领域的动态。

查看英文原文：Getting Started to Quarkus Reactive Messaging with Apache Kafka

使用Apache Kafka实现Quarkus的反应式消息

摘要：

核心要点

什么是Apache Kafka以及为何要使用它？

Quarkus是什么？

如何在Quarkus中集成Kafka？

Quarkus入门

要开发的应用

电影播放的生产者（Movie Plays Producer）

创建项目

开发

电影播放的消费者（Movie Plays Consumer）

创建项目

开发

外部的Kafka Broker

结论

关于作者

**Alex Soto **

内容目录

Alex Soto