在 Akka Stream BroadcastHub 中使用泛型类

Question

我正在尝试创建一个

BroadcastHub

类型的

ConsumerRecord<String, String>

来自相同元素类型的

Source

，但不允许泛型类类型。

RunnableGraph<Source<ConsumerRecord<String, String>, NotUsed>> graph =
                source.toMat(
                        BroadcastHub.of(ConsumerRecord.class, 256),
                        Keep.right()
                );

发现类似的问题已得到解答here，但看起来没有类似的方法

BroadcastHub

目前我只能选择将泛型类型包装到包装类中

RunnableGraph<Source<ConsumerRecordWrapper, NotUsed>> graph =
                source.map(ConsumerRecordWrapper::new)
                        .toMat(
                                BroadcastHub.of(ConsumerRecordWrapper.class, 256),
                                Keep.right()
                        );

有更好的解决方案吗？

Answer 1

根本问题是擦除。 由于

ConsumerRecord.class

的存在只是为了指导类型推断（在运行时，无论类型如何，类对象都是相同的，并且有效地表现得像它始终

ConsumerRecord<Object, Object>

一样），因此可以安全地使用

Class

对象的未经检查的强制转换来让类型发挥作用。

所以这样的东西适合我编译

@SuppressWarnings("unchecked")
final Class<ConsumerRecord<String, String>> consumerRecordStringToStringClass = (Class<ConsumerRecord<String, String>>)(Class<?>)ConsumerRecord.class;

RunnableGraph<Source<ConsumerRecord<String, String>, NotUsed>> graph =
  source.toMat(
    BroadcastHub.of(consumerRecordStringToStringClass, 256),
    Keep.right()
  );

（这是其中一些答案的专业化）