我對使用 spring 云流和 kafka 的服務(wù)有問題。該服務(wù)一直工作正常,但昨天開始報告啟動時出現(xiàn)一系列異常:一段時間后,我們會看到這樣的異常:Caused by: org.springframework.messaging.core.DestinationResolutionException: failed to look up MessageChannel with name '946859a6-bc27-466d-91ba-3da93af50ac9:1' in the BeanFactory.; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No bean named '946859a6-bc27-466d-91ba-3da93af50ac9:1' available與 kafka 的連接配置了一個屬性: spring.kafka.bootstrap-server = kafka.kafka:9092并且主題配置了 Spring Cloud 流屬性: spring.cloud.stream.bindings.[topic-name].destination = blah與 kafka 的交互通過 spring 與如下代碼集成進行:@MessagingGatewaypublic interface StreamGateway {
@Gateway(requestChannel = KafkaConfig.ENRICH_PAYMENT, replyChannel = ChannelNames.PAYMENT_REPLY, replyTimeout = 10000)
String processPayment(String payload);}//不同的類:private final StreamGateway gateway;...gateway.processPayment(message)這是在 azure kubernetes 部署上運行,而 kafka 與 spring boot 服務(wù)位于一個單獨的 pod 中。提前致謝。更新: 問題再次發(fā)生,一些進一步的調(diào)查強調(diào)了一些事情因為我們使用 spring 集成@MessagingGateway并@Gateway創(chuàng)建與 Kafka 的同步交互,所以沒有正常的主題StreamListener或訂閱者當主題存在滯后時會出現(xiàn)問題,即主題中存在超出主題偏移量的消息。缺少正常StreamListener意味著無法處理滯后消息。只有當 MessageGateway 建立連接時,才有可能從主題中讀取消息。解決問題的一種方法是讀取所有“滯后”消息,使滯后為 0。然后服務(wù)將正常啟動,但是如果我手動將消息發(fā)布到主題(與 MessageGateway 交互),則錯誤再次發(fā)生。第二個部分解決方案(我還沒有完全理解)是向@DependsOnMessageGateway添加一個注釋,表明它需要一個使用@InputSubscribableChannel 對象單獨創(chuàng)建的 bean 。這意味著 SubscribableChannel 必須在 MessageGateway 之前創(chuàng)建,因此創(chuàng)建了一個訂閱者,但是仍然沒有 StreamListener,因此當從主題中拉出滯后消息時仍然會拋出異常,無處可去 ??
Spring Cloud 流/Kafka 異常
慕尼黑8549860
2021-09-12 20:21:07