如何解决如何在Quarkus处理阶段创建多个KafaRecords?
Kafka quickstart示例说明了如何使用Kafka在Quarkus中生成,处理,读取和写入消息。我想在已发布的消息中设置Kafka键,而我正在使用KafkaRecord
进行设置。如果处理阶段生成单个KafkaRecord
,但使用Multi
创建多个记录时失败,则此方法有效。下面的代码显示了Quarkus的Kafka快速入门示例中的PriceGenerator
类的简单扩展:
/**
* A bean producing random prices every 5 seconds.
* The prices are written to a Kafka topic (prices). The Kafka configuration is specified in the application configuration.
*/
@ApplicationScoped
public class PriceGenerator {
private Random random = new Random();
@Outgoing("convert-to-kafka")
public Multi<Integer> generate() {
return Multi.createFrom().ticks().every(Duration.ofSeconds(5))
.onOverflow().drop()
.map(tick -> random.nextInt(100));
}
@Incoming("convert-to-kafka")
@Outgoing("generated-price")
public Multi<KafkaRecord<String,Integer>> convert(int price) {
return Multi.createFrom()
.items(KafkaRecord.of("original",price),KafkaRecord.of("double",2*price));
}
}
这不起作用,并生成以下类广播异常:
2020-08-21 08:36:05,284 ERROR [io.sma.rea.mes.provider] (executor-thread-1) SRMSG00200: The method org.acme.kafka.PriceGenerator#convert has thrown an exception: java.lang.ClassCastException: class org.eclipse.microprofile.reactive.messaging.Message$$Lambda$753/0x0000000800564840 cannot be cast to class java.lang.Integer (org.eclipse.microprofile.reactive.messaging.Message$$Lambda$753/0x0000000800564840 is in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @7c9bdee9; java.lang.Integer is in module java.base of loader 'bootstrap')
at org.acme.kafka.PriceGenerator_SmallRyeMessagingInvoker_convert_168e29259fd7b03bb28d7d0c9b052712b01b02a4.invoke(PriceGenerator_SmallRyeMessagingInvoker_convert_168e29259fd7b03bb28d7d0c9b052712b01b02a4.zig:46)
一种解决方法是将convert
方法分为两个处理阶段,一个阶段使用Multi<Integer>
作为返回类型来复制价格消息,然后将其转换为
进入KafkaRecords
。如下所示:
@ApplicationScoped
public class PriceGenerator {
private Random random = new Random();
@Outgoing("duplicate")
public Multi<Integer> generate() {
return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
.onOverflow().drop()
.map(tick -> random.nextInt(100));
}
@Incoming("duplicate")
@Outgoing("convert-to-kafka")
public Multi<Integer> duplicate(int price) {
return Multi.createFrom()
.items(price,2* price);
}
@Incoming("convert-to-kafka")
@Outgoing("generated-price")
public KafkaRecord<String,Integer> convert(int price) {
return KafkaRecord.of("some key",price);
}
}
我相信返回Multi<KafkaRecord<~>>
类型的阶段也应该起作用,但是也许我错过了一些东西...
我已经使用Quarkus 1.7.0.Final运行以上示例。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。