Akka Stream中使用泛型类创建BroadcastHub的问题

我尝试从同元素类型的Source创建一个类型为ConsumerRecord<String, String>的BroadcastHub，但不允许使用泛型类类型。最初的代码如下：

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

我曾找到过类似问题的解决方案，但该方案似乎不适用于BroadcastHub。目前只能将泛型类包装成一个包装类，代码如下：

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

请问有没有更好的解决办法？

不需要额外创建包装类，有两种更简洁的方式处理这个泛型类型问题：

方式一：强制类型转换绕过泛型擦除 #

直接通过双层强制转换获取符合要求的Class对象，配合注解消除编译警告：

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

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

只要上游Source输出的确实是ConsumerRecord<String, String>类型，这种方式在运行时不会有类型安全问题。

方式二：借助TypeToken获取泛型类型（需Guava依赖） #

如果项目中已引入Guava库，可以用TypeToken精准获取泛型类型的原始类：

TypeToken<ConsumerRecord<String, String>> typeToken = new TypeToken<ConsumerRecord<String, String>>() {};
@SuppressWarnings("unchecked")
Class<ConsumerRecord<String, String>> recordClass = (Class<ConsumerRecord<String, String>>) typeToken.getRawType();

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

内容的提问来源于stack exchange，提问作者BeeHellTea