Spring Cloud Stream Dispatcher hat keine Abonnenten

Question

Spring Cloud Stream Dispatcher hat keine Abonnenten Fehler.

Nach einem erfolgreichen Frühjahr Boot-Container starten wir einige Benachrichtigungsmeldungen auf einem Kafka Thema und einige unserer Microservices tut gleiche Funktion setzen müssen, und aus diesem Grund schrieben wir ein gemeinsames Glas, die Definitionen setzen Kanal enthält und Versand utils. Und die Funktionalität funktioniert wie erwartet, solange wir das Dienstprogramm direkt nach dem Aufruf SpringApplication.run aufrufen.

Folgendes ist eines unserer Microservices Application Class Beispiel.

@SpringBootApplication 
public class Application { 
    public static void main(String[] args) { 
     ConfigurableApplicationContext context =SpringApplication.run(Application.class, args); 
     context.getBean(SchedulerConsumerUtils.class).registerOrRestartConsumerJobs(); 
    } 
} 

Die oben aufgebaut funktioniert wie erwartet, jedoch bringt diese unnötige Belastung der Entwickler den Kessel Template-Code auf jeder Micro zu schreiben. Um dies zu vermeiden, haben wir eine Aspect-Implementierung geschrieben, die die gleiche Funktion erfüllt. Bei unserem Aspekt-Ansatz stoßen wir jedoch auf den folgenden Fehler.

org.springframework.context.ApplicationContextException: Fehler Bohne 'outputBindingLifecycle' beginnen; verschachtelte Ausnahme ist org.springframework.messaging.MessageDeliveryException: Der Dispatcher hat keine Abonnenten für den Kanal 'schedulertestsvcs: dev: 1180.scheduledJobExecutionResponseOutput' .; verschachtelte Ausnahme ist org.springframework.integration.MessageDispatchingException: Dispatcher hat keine Abonnenten

Wir mehrere Ansätze wie Spring SmartLifeCycle versucht einen Griff auf allen Kafka Output/Input Kanal Start Abschluss zu bekommen, aber alle von ihnen laufen in der gleiche Fehler.

Im Anschluss an unsere Aspect Implementierung auf org.springframework.boot.SpringApplication.run (..)

@Aspect 
@Component 
public class SchedulerConsumerAspect { 

    @Autowired 
    protected ApplicationContext applicationContext; 
    @AfterReturning(value = "execution(* org.springframework.boot.SpringApplication.run(..))",returning = "result") 
    public void afterConsumerApplicationStartup(JoinPoint pjp, Object result) throws Throwable { 
     if(result!=null){ 
      ConfigurableApplicationContext context=(ConfigurableApplicationContext) result; 
      if(context.containsBean("schedulerConsumerUtils")){ 
       //For what ever reason the following call resulting in Dispatcher has no subscribers for channel error. 
       //TODO fix the above issue and enable the following call. 
       context.getBean(SchedulerConsumerUtils.class).registerOrRestartConsumerJobs(); 
      } 
     } 
    } 

} 

Während unserer Debug-Sitzungen fanden wir org.springframework.boot.SpringApplication.run aus (..) Aspect wurde während des Bootstrap-Prozesses mehrmals aufgerufen. Zuerst, wenn der Aspekt aufgerufen wurde, haben wir den Ergebniswert als null, nach einer gewissen Zeit ruft Spring Boot denselben Aspekt auf, diesmal ist das Ergebnis nicht null. Auch wenn das Ergebnis nicht null ist, gibt es keinen Empfänger, der die Komponente vollständig initialisiert hat. Deshalb sehen Sie eine Überprüfung für context.containsBean ("schedulerConsumerUtils"). Nach der Bean-Initialisierung sehen wir jedoch, dass Ausgabekanäle nicht vollständig gebunden sind.

Was ist der beste Weg, um die Fertigstellung der Spring Cloud Stream Kafka-Ausgabe-/Eingangskanalbindung in den Griff zu bekommen?

Warum funktioniert der Komponentenaufruf in SpringBoot Application, aber nicht in Aspect? Ich kämpfte an diesen paar Tagen konnte nicht die richtige Lösung finden. Jede Hilfe sehr geschätzt.

Answer 1

Ich folgte dem Vorschlag von diesem Beitrag Spring cloud stream - send message after application initalization und verwendete die 3. Option ApplicationRunner. Die ersten beiden Optionen haben bei mir nicht funktioniert.

@Component 
public class AppStartup implements ApplicationRunner { 
    @Autowired 
    protected ApplicationContext applicationContext; 

    @Override 
    public void run(ApplicationArguments args) throws Exception { 
     if(applicationContext!=null){ 
      applicationContext.getBean(SchedulerConsumerUtils.class).registerOrRestartConsumerJobs(); 
     } 
    } 
}