Quarkus应用程序中Kafka反序列化程序中的CDI上下文

Question

我有一个Quarkus项目，基于Kafka的Smallrye反应式消息传递。因为我想使用“复杂的pojo”，所以我需要一个定制的反序列化程序。

我想制作这两个类CDI bean，这样我就可以注入和使用我的自定义记录器，它是一个CDI bean。有没有办法实现这一点？

现在，我的注入记录器对象只是空的：

import org.apache.kafka.common.serialization.Serializer;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;

@ApplicationScoped
public class MySerializer implements Serializer<MyDto>
{
    @Inject MyLogger logger;

    @Override public byte[] serialize(String topicName, MyDto myDto)
    {
        // this causes a java.lang.NullPointerException
        logger.info("serializing");

        ...
    }
}

Answer 1

据我所知，你只能用kafka注册一个类名，它会在内部创建那个类，即。不使用CDI。

可能的解决方法：使注册对象成为CDI bean的薄包装，并将工作委托给该bean：

public class MySerializer implements Serializer<MyDto> {
    private MySerializerCdi delegate;

    public MySerializer() {
        delegate = CDI.current().select(MySerializerCdi.class).get();
    }

    @Override public byte[] serialize(String topicName, MyDto myDto) {
        return delegate.serialize(topicName, myDto);
    }
    ...
}

... 并相应地重命名您原来的CDI类。