我創(chuàng)建了一個自定義 Flink RichSinkFunction 并嘗試JpaRepository在這個自定義類中自動裝配 a但我不斷地得到一個NullPointerException. 如果我在構造函數(shù)中自動裝配它，我可以看到已找到 JpaRepo - 但是當調用 invoke 方法時，我收到一個NullPointerException.public interface MessageRepo extends JpaRepository<Message, Long> {}@Componentpublic class MessageSink extends RichSinkFunction<Message> { private final transient MessageRepo messageRepo; //if i don't make this transient, i get the error message "The implementation of the RichSinkFunction is not serializable" @Autowired public MessageSink(MessageRepo messageRepo){ this.messageRepo = messageRepo; messageRepo.save(new Message()); //no issues when i do this } @Override public void invoke(Message message, Context context) { // the message is not null messageRepo.save(message); // NPE }有沒有人遇到過這個問題？看起來MessageSinkinvoke 方法是在單獨的線程中調用的，這就是為什么messageRepo總是null? 除了當我有自己的自定義接收器時，我的代碼的其他部分能夠使用 MessageRepo。
2 回答
慕桂英4014372
TA貢獻1871條經(jīng)驗 獲得超13個贊
我認為這里的問題是 flink 在分發(fā)給它的工作人員之前需要序列化自定義接收器函數(shù)。
通過標記 MessageRepo 傳輸，意味著當工作節(jié)點反序列化此函數(shù)時該字段將為空。通常，您會在 open 函數(shù)中初始化傳輸依賴項，該函數(shù)將在對象反序列化后調用。
慕無忌1623718
TA貢獻1744條經(jīng)驗 獲得超4個贊
我不太清楚原因，但我認為在注入 bean 時，spring boot 會優(yōu)先考慮您的服務類。當我嘗試為我的實體類編寫偵聽器時，我遇到了類似的問題。我就是這樣解決的。創(chuàng)建一個實現(xiàn) ApplicationContextAware 接口并覆蓋 setApplicationContext 方法的組件類。在您的類中有一個名為 getBean 的靜態(tài)方法，它將在您的第一個請求時自動裝配。示例代碼 ---
@Component
public class SpringBeansUtil implements ApplicationContextAware {
private static ApplicationContext context;
@SuppressWarnings("static-access")
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.context = applicationContext;
}
public static <T> T getBean(Class<T> beanClass) {
return context.getBean(beanClass);
}
}
然后在你的代碼中獲取 bean ------->> ClassName referenceName = (ClassName)SpringBeansUtil.getBean(ClassName.class);
添加回答
舉報
0/150
提交
取消