Asynchrone Kamelkomponente - doStop () wird sofort aufgerufen

Question

Ich versuche, eine Kamelkomponente zu erstellen, die eine API von einem externen Dienst verwendet.

Meine Route ist wie folgt

from("myComponent:entity?from=&to=")
.to("seda:one")

from("seda:one")
.aggregate(constant(true), new GroupedBodyAggregationStrategy())
.completionSize(5)
.completionTimeout(5000)
.process( new Processor1() )
to("seda:two")

.
.
.


from("seda:five")
.to("myComponent2:entity")

Ich habe meinen Komponentenverbraucher wie folgt implementiert

public class MyComponentConsumer extends DefaultConsumer {

    public MyComponentConsumer(MyComponentEndpoint endpoint, Processor processor) {
        super(endpoint, processor);
    }

    @Override
    protected void doStart() throws Exception {
        super.doStart();
        flag = true;
        while ( flag ) {
            //external API call
            Resource resource = getNextResource();
            if ( resource.next() == null ) {
                flag = false;
            }
            Exchange ex = endpoint.createExchange(ExchangePattern.InOnly);
            ex.getIn().setBody(resource.toString());
            getAsyncProcessor().process(
                            ex
                            doneSync -> {
                                LOG.info("Message processed");
                            }
                    );
        }
    }

    @Override
    protected void doStop() throws Exception {
        super.doStop();
        System.out.println("stop ---- ");
    }
}

Alles hat gut funktioniert und die Daten haben sich auf der Strecke verbreitet. Mein einziges Problem war, dass die Daten erst nach Abschluss des gesamten Prozesses auf den nächsten Teil übertragen wurden. Und die nächsten Teile liefen asynchron.

Ich habe mir das Beispiel von StreamConsumer angesehen und versucht, es mithilfe einer ausführbaren Datei und eines executorService in meinen Code zu implementieren. Aber wenn ich das tue, hört der Verbraucher auf, sobald es beginnt.

Ich habe den Code in geändert

public class MyComponentConsumer extends DefaultConsumer implements Runnable 

und hinzugefügt

private ExecutorService executor;
getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this, "myComponent");
executor.execute(this);

und habe meine Logik in die run () -Methode verschoben. Der Consumer-Thread endet jedoch, sobald er beginnt. und der asynchrone Prozessor überträgt die Daten nicht richtig.

Gibt es eine andere Möglichkeit, die von mir benötigte Funktionalität zu implementieren, oder irre ich mich hier. Jede Hilfe wäre dankbar.

Answer 1

Welche Version von Kamel benutzt du?

Es gab ein Problem mit der Verwaltung des Verbraucherstatus in Kamel 2.x, das in Kamel 3.x CAMEL-12765 behoben wurde und zu dem hier beschriebenen Problem führen kann.

Wenn Sie sich auf camel 2.x befinden, versuchen Sie, newScheduledThreadPool anstelle von newSingleThreadExecutor zu verwenden. Auch executor.schedule (this, 5L, TimeUnit.SECONDS) anstelle von executor.execute (this).

Ein verzögerter Start des Executors kann dazu beitragen, das Problem zu vermeiden, mit dem Sie konfrontiert sind.