Asynchronous Camel Component - doStop () เรียกทันที

Question

ฉันกำลังพยายามสร้างส่วนประกอบอูฐซึ่งใช้ API จากบริการภายนอก

เส้นทางของฉันมีดังนี้

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")

ฉันใช้งานผู้บริโภคส่วนประกอบของฉันดังนี้

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 ---- ");
    }
}

ทุกอย่างทำงานได้ดีและข้อมูลก็ถูกส่งผ่านเส้นทาง ปัญหาเดียวของฉันคือข้อมูลไม่ได้ส่งต่อไปยังส่วนถัดไปจนกว่ากระบวนการทั้งหมดจะเสร็จสมบูรณ์ และส่วนถัดไปทำงานแบบอะซิงโครนัส

ฉันดูตัวอย่างของ StreamConsumer และพยายามนำไปใช้กับโค้ดของฉันโดยใช้ runnable และ executorService แต่ถ้าฉันทำให้ผู้บริโภคหยุดทันทีที่เริ่ม

ฉันเปลี่ยนรหัสเป็น

public class MyComponentConsumer extends DefaultConsumer implements Runnable 

และเพิ่ม

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

และย้ายตรรกะของฉันไปไว้ในเมธอด run () แต่เธรดผู้บริโภคจะสิ้นสุดทันทีที่เริ่มต้น และตัวประมวลผล async ไม่ได้ถ่ายโอนข้อมูลอย่างถูกต้อง

มีวิธีอื่นในการใช้ฟังก์ชันที่ฉันต้องการหรือฉันเข้าใจผิดที่นี่ ความช่วยเหลือใด ๆ จะได้รับการชื่นชม

Answer 1

คุณใช้อูฐรุ่นอะไร

มีปัญหาในการจัดการสถานะของผู้บริโภคในอูฐ 2.x ซึ่งได้รับการแก้ไขแล้วในอูฐ 3.x CAMEL-12765ซึ่งอาจนำไปสู่ปัญหาที่คุณอธิบายไว้ที่นี่

หากคุณใช้อูฐ 2.x ให้ลองใช้ newScheduledThreadPool แทน newSingleThreadExecutor นอกจากนี้ผู้ดำเนินการกำหนดการ (นี่คือ 5L, TimeUnit.SECONDS) แทนการดำเนินการดำเนินการ (สิ่งนี้)

การเริ่มตัวดำเนินการล่าช้าอาจช่วยหลีกเลี่ยงปัญหาที่คุณกำลังเผชิญอยู่