如何使用Quarkus按拓扑启动 Kafka 流管道

Question

我正在学习QuarkusKafka流教程，不太明白如何启动管道。

在本教程中，org.apache.kafka.streams。StreamsBuilder用于构建org.apache.kafka.streams。描述管道的拓扑。构建拓扑的方法用＜code＞注释为products＜code＞。在本备忘单中，描述了这足以运行Kafka Streams管道。在本教程中，还公开了httpendpoint。这在我目前正在实现的服务中不是必需的。此外，在本示例中，从未显式调用提供程序方法。当我在没有endpoint的情况下启动应用程序时，管道不会启动。

在本教程中，管道使用拓扑显式实例化。但在这里，属性必须手动设置，并且配置不是从Quarkus.kafka流中获取的。

问题是：如何使用第一个教程中的拓扑生成器来启动它所描述的管道？最佳情况是来自Quarkus.kafka流的配置。

使用：

• Java OpenJDK 11.0.8
• Quarkus版本：1.8.0.最终版




Answer 1

匿名用户

解决了问题：无法保证完全正确。只是想分享解决我问题的方法。

最重要的是使用正确的@products<代码>javax.enterprise.inject。products必须用于拓扑producting方法javax.ws.rs.products还可以用于定义输出的媒体类型，但不是强制性的：

    @javax.ws.rs.Produces( MediaType.TEXT_PLAIN )
    @Produces
    @AlternativePriority( 1 )
    public Topology buildTopology() {
        ...
    }

Kafka 流的实例由框架在启动时自动构建。运行管道所需的全部内容如下：

@ApplicationScoped
public class YourApplication {
    private final KafkaStreams streams;

    public NasDistributorApplication( final KafkaStreams streams ) {
        this.streams = streams;
    }

    public void onStart( @Observes final StartupEvent startupEvent ) {
        streams.start();
    }

    public void onStop( @Observes final ShutdownEvent shutdownEvent ) {
        streams.close();
    }

}

通过< code > @ alternative priority(1)对< code >拓扑生成方法进行注释，可能有必要告诉< code>quarkus使用此方法来构建< code >拓扑。我不知道框架的内部结构，但我怀疑使用了默认拓扑，并且< code > @ alternative priority 给予自定义方法比默认< code >拓扑更高的优先级